"""Implements Parallel node type for the sequenceer.
"""
import attr
from .state import T_STATE
from .sequence import Sequence
from .. import ob
[docs]@attr.s
class Parallel(Sequence):
"""
Parallel node definition.
Use the :meth:`create` method to build properly :class:`Parallel` nodes.
Since it inherits from :class:`Sequence` it has access to the same context variables.
Example:
One can access the running Sequence using Sequence.current_seq context variable.
>>> s = Parallel.create(a,b, name="my sequence")
# execute Sequence
>>> await s.start()
# Get running Sequence from Inside function a():
>>> def a():
... current_seq = Sequence.current_seq
... # now current_seq is the node s
... assert current_seq == s
"""
seq_args = attr.ib(kw_only=True, default=attr.Factory(list), repr=False)
def __attrs_post_init__(self):
if self.name is None:
self.name = "Par"
super().__attrs_post_init__()
self.seq = self.seq_args
def __repr__(self):
return "<Parallel (%s)...>" % (self.id)
[docs] def make_sequence(self, parent_tpl=None):
ctrl = ob.OB.controller.get()
G = self.graph
st = self.start_node
end = self.end_node
G.add_node(st.id, node=st)
G.add_node(end.id, node=end)
st.state = end.state = T_STATE.NOT_STARTED
ctrl = ob.OB.controller.get()
if ctrl:
self.runtime_flags = ctrl.runtime_flags.get(self.serial_number, 0)
for el in self.seq:
G.add_node(el.id, node=el)
G.add_edge(st.id, el.id)
G.add_edge(el.id, end.id)
el.state = T_STATE.NOT_STARTED
el.make_sequence(parent_tpl)
[docs] @staticmethod
def create(*args, **kw):
"""Creates a :class:`Parallel` node
Args:
*args: Variable length list of nodes or coroutines that compose the sequence.
Keyword Args:
id: Node id
name: node name
"""
a = Parallel(seq_args=args, **kw)
return a