# Copyright 2025 Apheleia
#
# Description:
# Apheleia Verification Library Manager Sequence
from collections.abc import MutableMapping, MutableSequence
import random
import avl
from ._item import ReadItem, SequenceItem, WriteItem
[docs]
class ManagerSequence(avl.Sequence):
[docs]
def __init__(self, name: str, parent: avl.Component) -> None:
"""
Initialize the sequence
Sequence of independently randomized Manageractions
:param name: Name of the sequence item
:param parent: Parent component of the sequence item
"""
super().__init__(name, parent)
self.i_f = avl.Factory.get_variable(f"{self.get_full_name()}.i_f", None)
"""Handle to interface - defines capabilities and parameters"""
self.n_items = avl.Factory.get_variable(f"{self.get_full_name()}.n_items", 1)
"""Number of items in the sequence (default 1)"""
self.wait_for = avl.Factory.get_variable(f"{self.get_full_name}.wait_for", None)
"""Default phase to wait on after a item sent to driver before next item"""
self._items_ = []
"""Executed items. Used to track end of sequence"""
[docs]
def done_cb(self, *args, **kwargs):
"""
Callback on item done event
:param args: Arguments
:param kwargs: Keyword arguments
"""
pass
[docs]
def control_cb(self, *args, **kwargs):
"""
Callback on item control event
:param args: Arguments
:param kwargs: Keyword arguments
"""
pass
[docs]
def data_cb(self, *args, **kwargs):
"""
Callback on item data event
:param args: Arguments
:param kwargs: Keyword arguments
"""
pass
[docs]
def response_cb(self, *args, **kwargs):
"""
Callback on item response event
:param args: Arguments
:param kwargs: Keyword arguments
"""
self.get_sequencer().drop_objection()
async def _send_(self, item : SequenceItem, randomize : bool = True, wait_for : str = None) -> SequenceItem:
"""
Send an item to the driver
:param item: Item to send
:param randomize: Randomize the item before sending (default True)
:param wait_for: Phase to wait on after item sent to driver before next item (default None)
:return: The item sent
"""
self.get_sequencer().raise_objection()
item.add_event("done", self.done_cb)
item.add_event("control", self.control_cb)
item.add_event("data", self.data_cb)
item.add_event("response", self.response_cb)
await self.start_item(item)
if randomize:
item.randomize()
else:
item.resize()
await self.finish_item(item)
# Track
self._items_.append(item)
if wait_for is not None:
await item.wait_on_event(wait_for)
return item
[docs]
async def next(self) -> SequenceItem:
"""
Get the next item in the sequence
"""
item = random.choice([WriteItem(f"from_{self.name}", self), ReadItem(f"from_{self.name}", self)])
return await self._send_(item, randomize=True, wait_for=self.wait_for)
[docs]
async def write(self, **kwargs) -> WriteItem:
"""
Send a write item to the driver
:param kwargs: Keyword arguments to set on the item
:return: The item sent
"""
item = WriteItem(f"from_{self.name}", self)
for k,v in kwargs.items():
if hasattr(item, k):
if isinstance(v, (MutableSequence | tuple)):
for _i,_v in enumerate(v):
item.set(k, _v, idx=_i)
elif isinstance(v, MutableMapping):
for _k,_v in v.items():
item.set(k, _v, idx=_k)
else:
item.set(k, v)
if "wait_for" in kwargs:
wait_for = kwargs["wait_for"]
else:
wait_for = self.wait_for
return await self._send_(item, randomize=False, wait_for=wait_for)
[docs]
async def read(self, **kwargs) -> ReadItem:
"""
Send a read item to the driver
:param kwargs: Keyword arguments to set on the item
:return: The item sent
"""
item = ReadItem(f"from_{self.name}", self)
for k,v in kwargs.items():
if hasattr(item, k):
if isinstance(v, (MutableSequence | tuple)):
for _i,_v in enumerate(v):
item.set(k, _v, idx=_i)
elif isinstance(v, MutableMapping):
for _k,_v in v.items():
item.set(k, _v, idx=_k)
else:
item.set(k, v)
if "wait_for" in kwargs:
wait_for = kwargs["wait_for"]
else:
wait_for = self.wait_for
return await self._send_(item, randomize=False, wait_for=wait_for)
[docs]
async def body(self) -> None:
"""
Body of the sequence
"""
self.info(f"Starting Manager sequence {self.get_full_name()} with {self.n_items} items")
for _ in range(self.n_items):
_item = await self.next()
[docs]
async def post_body(self) -> None:
"""
Wait for all items to have seen response
"""
await super().post_body()
for item in self._items_:
await item.wait_on_event("response")
self._items_.clear()
__all__ = ["ManagerSequence"]