seqlib.nodes package

Sequencer nodes

Implements all node types and utility functions.

Action nodes

Implements Action and ActionInThread nodes.

Actions are the leaf nodes in any Sequencer script. They execute python methods or functions.

Coroutines are associated to Action nodes while normal callable objects must be associated to ActionInThread nodes.

class seq.lib.nodes.action.Action(f=None, parent_tpl=None, *, id=None, name=None, runtime_flags=0, description='')

Bases: seq.lib.nodes.interface._BaseNode

Action ctor.

Action nodes executes coroutines in a Sequencer script.

Parameters

f (coroutine) – The coroutine the node will execute

Keyword Arguments

• id (str) – Unique id. If not provided an unique identifier is assigned.

• name (str) – Node name. If not provided a name is assigned.

Raises

TypeError – is f is not a coroutine.

current_node = <ContextVar name='current_node' default=None>

property context

Get context from the running Sequence

property full_state

Gets the node state.

property state

Gets the node state.

make_sequence(parent_tpl=None)

Not much to do here. Link back to Template object (if any) Get runtime flags from controller object.

class seq.lib.nodes.action.ActionInThread(f=None, parent_tpl=None, *, id=None, name=None, runtime_flags=0, description='')

Bases: seq.lib.nodes.action.Action

ActionInThread ctor.

ActionInThread nodes executes python callables (functions or methods) in a Sequencer script.

Parameters

f (callable) – The callable the node will execute

Keyword Arguments

• id (str) – Unique id. If not provided an unique identifier is assigned.

• name (str) – Node name. If not provided a name is assigned.

Raises

TypeError – is f is not a python method or function.

class seq.lib.nodes.action.StartNode(*args, **kw)

Bases: seq.lib.nodes.action.Action

This is Starting Node.

marks node as unskipabble

class seq.lib.nodes.action.EndNode(*args, **kw)

Bases: seq.lib.nodes.action.Action

This is the finish Node.

marks node as unskippable

seq.lib.nodes.action.make_node(s, **kw)

Makes an Action nodes out of a coroutine.

Parameters

s (coroutine) – target coroutine to be passed to Action ctor.

Keyword Arguments

**kw – Passed straight to Action ctor.

Returns

Action node associated to s coroutine.

Raises

TypeError – if ‘s’ is not a coroutine.

Checkbox node

Implements a Checkbox node.

Checkbox pauses execution of a Sequencer script until the ‘check’ command is given to it. That is to say, until the user confirms some condition and allows the script to continue.

Warning

This node is intended to use on interactive Sequencer scripts only. Do not seq run a script that contains it. It will hang forever.

class seq.lib.nodes.checkbox.Checkbox(parent_tpl=None, *, id=None, name=None, runtime_flags=0, description='')

Bases: seq.lib.nodes.action.Action

Checkbox ctor.

Checkbox node waits for user to confirm some condition.

Parameters

msg (str) – Checkbox message

Keyword Arguments

• id (str) – Unique id. If not provided an unique identifier is assigned.

• name (str) – Node name. If not provided a name is assigned.

checked()

Release the checkbox allowing the Sequencer script to continue its execution

static create(msg, **kw)

Checkbox node constructor.

Parameters

• msg (str) – Message to display along with the checkbox.

• *kw – standard sequencer node keywords.

Keyword Arguments

• id – Node id

• name – node name

Container nodes

class seq.lib.nodes.Sequence(*, id=None, name=None, runtime_flags=0, description='')

Bases: seq.lib.nodes.interface._BaseNode

Basic sequencer node.

Use create() to properly build Sequence objects.

state

node state

Context Variables
Name	Desc
current_seq	The parent of the current node
root	Top level DAG’s root

Examples

>>> s = Sequence.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

current_seq = <ContextVar name='current_seq' default=None>

root = <ContextVar name='root' default=None>

property context

Get context dictionary, preferably from root node.

property G

returns the graph object

async start_step()

Standard sequence’s start step.

Sets the sequence’s state to RUNNING

async end_step()

Standard sequence’s end step.

Evaluates the sequence’s final state. Collects node’s result and put them in the sequence’s result attribute.

property start_node

Returns the start node.

If it does not exist, it creates it.

property end_node

Returns the end node.

If it does not exist, it creates it.

append(s)

Appends a node to the Sequence

make_sequence(parent_tpl=None)

Builds this sequence execution graph. Joins the Sequence’s nodes together.

create_node_tasks(resume=False)

Creates Task object associated to this node

reschedule_node(node_id)

Reschedule a node for execution

start(make_sequence=True, resume=False)

This is the entry point for Sequence execution.

Returns

Returns the SeqTask object that executes the sequence

Raises

Exception – Any exception received is re-raised and the sequence is aborted.

async run()

Runs the node – This executes the task

async execute(resume=False, propagate=False)

Executes node – this just creates the asyncio task

async resume()

Resume node execution

main_task()

Returns the objective node of the sequence – the end node

abort()

Aborts the sequence.

Goes trough the full graph and aborts the tasks associated to nodes (if any). Do not allow nodes to run by taking away its running_checkpoint attribute.

nodes()

Return nodes from Graph

get_node(node_id)

Get node by id

get_task(node_id)

Get task by node_id

property parameters

Return parameters

par(k)

Get a parameter value

set(p)

Sets the value of a parameter

property state

Gets the node state.

property full_state

Gets the node state.

static get_context()

static create(*args, **kw)

Sequence node constructor.

Parameters

*args – Variable length list of nodes or coroutines that compose the sequence.

Keyword Arguments

• id – Node id

• name – node name

class seq.lib.nodes.Parallel(*, id=None, name=None, runtime_flags=0, description='', seq_args=NOTHING)

Bases: seq.lib.nodes.sequence.Sequence

Parallel node definition.

Use the create() method to build properly Parallel nodes. Since it inherits from 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

make_sequence(parent_tpl=None)

Builds this sequence execution graph. Joins the Sequence’s nodes together.

static create(*args, **kw)

Creates a Parallel node

Parameters

*args –

Variable length list of nodes or coroutines that compose the sequence.

Keyword Args:

id: Node id name: node name

class seq.lib.nodes.Loop(parent_tpl=None, *, id=None, name=None, runtime_flags=0, description='', block_args=NOTHING, condition=None, init=None)

Bases: seq.lib.nodes.sequence.Sequence

Loop node definition.

Use the create() method to build properly Loop node. Since it inherits from Sequence it has access to the same context variables.

Context Variables
Name	Desc
current_seq	The parent of the current node (from Sequence)
root	Top level DAG’s root (from Sequence)
index	The Loop’s current index (starts at 0)

Keyword Arguments

• 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.

index = <ContextVar name='index' default=0>

async end_step()

Standard Loop’s end step. Evaluates the Loop’s final state.

make_sequence(parent_tpl=None)

Creates the Loop execution graph

set_block(*args)

Assigns the sequence to the loop’s block

make_task(node, input_list, resume)

Creates the task object that executes the node.

nodes()

Return nodes from Graph

get_node(node_id)

Get node by id

static create(*args, **kw)

Creates a Loop node

Parameters

*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 Action or ActionInThread. condition(node): condition node Action or ActionInThread.

Returns:

A new Loop object

Example:

Creating a loop.

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)

class seq.lib.nodes.Template(tpl_name=None, params=NOTHING, content=NOTHING, *, id=None, name=None, runtime_flags=0, description='')

Bases: seq.lib.nodes.sequence.Sequence

The Template node

A Template is just a Sequence node with access to a set of variables as defined in an ObservingBlock.

The ObservingBlock takes care of instantiating the Template objects as needed by the Observation Block.

Context Variables
Name	Desc
current_tpl	The current Template being executed

Warning

Objects of type Template should not be directly constructed. Leave that for the ObservingBlock.

P = <ContextVar name='P' default={}>

current_tpl = <ContextVar name='current_tpl' default=None>

async execute(resume=False, propagate=False)

Executes node – this just creates the asyncio task

make_sequence(parent_tpl=None)

Builds this sequence execution graph. Joins the Sequence’s nodes together.

static from_dict(d)

Creates a Template from keywords in dictionary.

static create(d, *args, **kw)

Creates a Template node

Parameters

*args –

Variable length list of nodes or coroutines that composes the sequence.

Keyword Args:

id: Node id. name: node name.

class seq.lib.nodes.ObservingBlock(fname=None, *, id=None, name=None, runtime_flags=0, description='', from_otto=False)

Bases: seq.lib.nodes.sequence.Sequence

Represents a Sequence from an OB file (json). The JSON file contains a lot of data, however the Sequencer only cares about the templates section:

"templates": [
        {
            "templateName": "seq.samples.a",
            "type": "string"
        },
        {
            "templateName": "seq.samples.tpa",
            "type": "string",
            "parameters": [
                {
                    "name": "par_b",
                    "type": "integer",
                    "value": 0
                },
                {
                    "name": "par_c",
                    "type": "number",
                    "value": 77
                }
            ]
        }
    ],

Inside the templates section there is list of templateName objects. The templateName defines a Python module that can be directly imported (it is reachable from PYTHONPATH).

Each template might contain a list of parameters that can be accessed from corresponding python code that implements the template.

template_parameters = {}

property parameters

Returns a dictionary with template’s parameters

The returned dictionary enumerates the OB’s templates as keys and the values are Template’s parameters

load()

Loads json file

static create(f, *args, **kw)

Creates a ObservingBlock node

Parameters

f –

JSON file with OB definition

Keyword Args:

id: Node id name: node name

property module

Returns OB filename

property doc

Returns OB name from the corresponding kw

clone(*args, **kw)

Fake ctor for OB nodes