# loop.py
"""Implements Loops for the sequencer.
A Loop node is composed of three parts.
* *Initialization* (``optional``) initiates variables to be used
in the loop's activity and its exit-condition; it is executed
once on entry to the loop
* *condition* (``callable``) Check the loop's condition.
It shall return *False* to interrupt the Loop. Otherwise it
shall *return* **True**.
* *block* (``Sequence``) The loop's body. It is the sequence of
nodes to be executed repeatedly until the Loop's condition
produces a False value.
"""
import logging
import contextvars as cv
import attr
import networkx as nx
from .action import make_node
from .sequence import Sequence
from .state import T_STATE
from .. import ob
user_logger = logging.getLogger("seq.user")
[docs]@attr.s
class Loop(Sequence):
"""Loop node definition.
Use the :meth:`create` method to build properly :class:`Loop` node.
Since it inherits from :class:`Sequence` it has access to the same context variables.
============ =======================================================
Context Variables
---------------------------------------------------------------------
Name Desc
============ =======================================================
current_seq The parent of the current node (from :class:`Sequence`)
root Top level DAG's root (from :class:`Sequence`)
index The Loop's current index (starts at 0)
============ =======================================================
Keyword Args:
id (Optional str): Node id.
name (Optional str): Node name.
init (Optional): Initialization node.
condition (callale): A Python method that returns a boolean value.
block (list): The loop's body.
"""
# my attributes
block_args = attr.ib(kw_only=True, default=attr.Factory(list), repr=False)
condition = attr.ib(kw_only=True, default=None, repr=False)
init = attr.ib(kw_only=True, default=None, repr=False)
block = attr.ib(init=False, default=None, repr=False)
node_value = attr.ib(init=False, default=0, repr=False)
parent_tpl = attr.ib(default=None, repr=False)
# Loop's contextvar
index = cv.ContextVar("index", default=0)
def __attrs_post_init__(self):
if self.name is None:
self.name = "Loop"
super().__attrs_post_init__()
self.set_block(*self.block_args)
if self.condition is None:
self.condition = self._condition
self.condition = make_node(self.condition)
self.condition.can_skip = False
if self.init is None:
self.init = self._init_loop
self.init = make_node(self.init)
self.init.can_skip = False
[docs] async def end_step(self):
"""Standard Loop's end step.
Evaluates the Loop's final state.
"""
self.state = T_STATE.FINISHED
user_logger.info("Loop %s finished", self.name)
G = self.graph
states = [
G.nodes[key]["node"].in_error for key in nx.topological_sort(G)
]
# grab node's state and check for ERROR
if any(states):
self.in_error = True
[docs] def make_sequence(self, parent_tpl=None):
"""Creates the Loop execution graph"""
G = self.graph
st = self.start_node
end = self.end_node
self.parent_tpl = parent_tpl
G.add_node(st.id, node=st)
G.add_node(self.condition.id, node=self.condition, label="Cond")
if self.block:
G.add_node(self.block.id, node=self.block, label="Loop body")
G.add_node(end.id, node=end)
# Adds runtime flags
ctrl = ob.OB.controller.get()
if ctrl:
self.runtime_flags = ctrl.runtime_flags.get(self.serial_number, 0)
[docs] def set_block(self, *args):
"""Assigns the sequence to the loop's block"""
if args:
self.block = Sequence.create(
*args, name="block", id="block_" + self.id
)
self.block.make_sequence(self.parent_tpl)
[docs] def make_task(self, node, input_list, resume):
from ..seqtask import LoopTask
return LoopTask(self.id, node, input_list, resume=resume)
async def _condition(self):
return False
async def _init_loop(self):
pass
[docs] def nodes(self):
return [
self.start_node.id,
self.condition.id,
self.block.id,
self.end_node.id,
]
[docs] def get_node(self, node_id):
# return self.block.get_node(node_id)
if node_id == self.condition.id:
return self.condition
elif node_id == self.block.id:
return self.block
elif node_id == self.init.id:
return self.init
elif node_id == self.start_node.id:
return self.start_node
elif node_id == self.end_node.id:
return self.end_node
else:
return self.block.get_node(node_id)
[docs] @staticmethod
def create(*args, **kw):
"""Creates a :class:`Loop` node
Args:
*args: Variable length list of nodes or coroutines that comprises the Loop`s body.
Keyword Args:
id: Node id
name: node name
init (node) : initialization node :class:`Action` or :class:`ActionInThread`.
condition(node): condition node :class:`Action` or :class:`ActionInThread`.
Returns:
A new :class:`Loop` object
Example:
Creating a loop.
.. code-block:: python
def eval_condition():
return False
class Tpl:
def initialize(self, context):
# performs some initialization
pass
async def a():
pass
async def b():
pass
@staticmethod
def create()
t = MyClass()
l = Loop.create(t.a, t.b, condition=eval_condition, init=t.initialize)
"""
a = Loop(block_args=args, **kw)
return a