lib.ob.OB Class Reference

An collection of OB's or Sequences. More...

Public Member Functions
def	reload_tpl (self, i)
	dummy, does nothing More...
def	load_tpl (self, t, append_flag=False)
	Import the template module and call its ctor. More...
def	run_from_file (self, fn)
	Loads a python file and executes it. More...
def	run_all (self)
	Run all loaded modules. More...
def	run_node (self, serial_number)
	Find node, run it. More...
def	wrap_modules (self)
	Instances the Sequence and wraps them in the ROOT Sequence. More...
def	wrap_and_run (self, wrapper=None)
	This is the important one. More...
def	get_tree (self)
	dict of dicts representing the tree More...
def	reset_tree (self)
	Resets the tree. More...
def	resume (self)
	Restart the Graph without recreating tasks. More...
def	retry_resume (self)
	Retries last failed mode and restarts tree execution. More...
def	resume_node (self, serial_number)
	Resumes node execution. More...
def	abort (self)
	Aborts execution. More...
def	nodes (self)
	Gets tree nodes. More...
def	find_node (self, nid)
	Finds a node in the tree based on its id. More...
def	find_node_by_number (self, serial_number)
	Search and return for a node with given serial number. More...
def	get_runtime_flags (self, serial_number)
	_----------------------------------------------------— Runtime Flags _----------------------------------------------------— More...
def	set_runtime_flag (self, serial_number, flag, value=True)
	Stores a node's runtime flag in dictionary. More...
def	mark_skip (self, serial_number, value=True)
	Marks the node for skip. More...
def	mark_pause (self, serial_number, flag=True)
	Marks a node for pause. More...
def	flip_flag (self, serial_number, flag)
def	flip_pause (self, serial_number)
def	flip_skip (self, serial_number)
def	checked_cb (self, serial_number)
	Checks (unblocks) a Checkbox node. More...
def	continue_seq_cb (self, node_result)
	Continue sequence after confirmation from popup dialog. More...
def	dumptree (self, fname)
def	set_variable (self, path, value)
	_----------------------------------------------------— More...
def	get_variable (self, path)
def	notify_state_change (self, node, ts=None, **kw)
	Published in the queue a node's state change. More...
def	reset (self)
def	load_ob (self, obname, append_flag=False, **kw)
	Loads OB file (json) More...
def	list_ob (self)

Static Public Member Functions
def	create_sequence (mod, *args, **kw)

Static Public Attributes
	tpmods = attr.ib(init=False, default=attr.Factory(list))
	lock = attr.ib(init=False, default=attr.Factory(asyncio.Semaphore))
	queue = attr.ib(init=False, default=attr.Factory(asyncio.Queue))
	wrapper = attr.ib(init=False, default=None)
	wrapper_task = attr.ib(init=False, default=None)
	transitions_q = attr.ib(init=False, default=None)
	monitor_task = attr.ib(init=False, default=None)
	monitor_flag = attr.ib(default=False)
	controller = cv.ContextVar("controller", default=None)
	ob_scripts = attr.ib(init=False, default=attr.Factory(list))
	runtime_flags = dict()
	ottoif = OttoClient()
dictionary	status_map_otto

Properties
	context = property
	context property More...

Detailed Description

An collection of OB's or Sequences.

This is the executor.

Member Function Documentation

abort()

def lib.ob.OB.abort

(

self

)

Aborts execution.

checked_cb()

def lib.ob.OB.checked_cb	(		self,
			serial_number
	)

Checks (unblocks) a Checkbox node.

continue_seq_cb()

def lib.ob.OB.continue_seq_cb	(		self,
			node_result
	)

Continue sequence after confirmation from popup dialog.

create_sequence()

def lib.ob.OB.create_sequence (   mod, *  args, **  kw  )

static

dumptree()

def lib.ob.OB.dumptree	(		self,
			fname
	)

---

find_node()

def lib.ob.OB.find_node	(		self,
			nid
	)

Finds a node in the tree based on its id.

find_node_by_number()

def lib.ob.OB.find_node_by_number	(		self,
			serial_number
	)

Search and return for a node with given serial number.

flip_flag()

def lib.ob.OB.flip_flag	(		self,
			serial_number,
			flag
	)

flip_pause()

def lib.ob.OB.flip_pause	(		self,
			serial_number
	)

flip_skip()

def lib.ob.OB.flip_skip	(		self,
			serial_number
	)

get_runtime_flags()

def lib.ob.OB.get_runtime_flags	(		self,
			serial_number
	)

_----------------------------------------------------— Runtime Flags _----------------------------------------------------—

Gets runtime flags from node with given serial number

get_tree()

def lib.ob.OB.get_tree

(

self

)

dict of dicts representing the tree

get_variable()

def lib.ob.OB.get_variable	(		self,
			path
	)

list_ob()

def lib.ob.OB.list_ob

(

self

)

load_ob()

def lib.ob.OB.load_ob	(		self,
			obname,
			append_flag = False,
		**	kw
	)

Loads OB file (json)

load_tpl()

def lib.ob.OB.load_tpl	(		self,
			t,
			append_flag = False
	)

Import the template module and call its ctor.

Create node's tasks

mark_pause()

def lib.ob.OB.mark_pause	(		self,
			serial_number,
			flag = True
	)

Marks a node for pause.

mark_skip()

def lib.ob.OB.mark_skip	(		self,
			serial_number,
			value = True
	)

Marks the node for skip.

nodes()

def lib.ob.OB.nodes

(

self

)

Gets tree nodes.

notify_state_change()

def lib.ob.OB.notify_state_change	(		self,
			node,
			ts = None,
		**	kw
	)

Published in the queue a node's state change.

reload_tpl()

def lib.ob.OB.reload_tpl	(		self,
			i
	)

dummy, does nothing

reset()

def lib.ob.OB.reset

(

self

)

reset_tree()

def lib.ob.OB.reset_tree

(

self

)

Resets the tree.

   Forces the engine to rebuild the tree by invalidating the current one.

resume()

def lib.ob.OB.resume

(

self

)

Restart the Graph without recreating tasks.

Finished tasks are ignored, pending task are executed.

resume_node()

def lib.ob.OB.resume_node	(		self,
			serial_number
	)

Resumes node execution.

Finds a paused node and restarts it

retry_resume()

def lib.ob.OB.retry_resume

(

self

)

Retries last failed mode and restarts tree execution.

run_all()

def lib.ob.OB.run_all

(

self

)

Run all loaded modules.

run_from_file()

def lib.ob.OB.run_from_file	(		self,
			fn
	)

Loads a python file and executes it.

run_node()

def lib.ob.OB.run_node	(		self,
			serial_number
	)

Find node, run it.

set_runtime_flag()

def lib.ob.OB.set_runtime_flag	(		self,
			serial_number,
			flag,
			value = True
	)

Stores a node's runtime flag in dictionary.

set_variable()

def lib.ob.OB.set_variable	(		self,
			path,
			value
	)

_----------------------------------------------------—

Sets variable value

wrap_and_run()

def lib.ob.OB.wrap_and_run	(		self,
			wrapper = None
	)

This is the important one.

   Starts the outside sequence (wrapper) which executes
   the user's sequence(s).

wrap_modules()

def lib.ob.OB.wrap_modules

(

self

)

Instances the Sequence and wraps them in the ROOT Sequence.

Member Data Documentation

controller

lib.ob.OB.controller = cv.ContextVar("controller", default=None)

static

lock

lib.ob.OB.lock = attr.ib(init=False, default=attr.Factory(asyncio.Semaphore))

static

monitor_flag

lib.ob.OB.monitor_flag = attr.ib(default=False)

static

monitor_task

lib.ob.OB.monitor_task = attr.ib(init=False, default=None)

static

ob_scripts

lib.ob.OB.ob_scripts = attr.ib(init=False, default=attr.Factory(list))

static

ottoif

lib.ob.OB.ottoif = OttoClient()

static

queue

lib.ob.OB.queue = attr.ib(init=False, default=attr.Factory(asyncio.Queue))

static

runtime_flags

lib.ob.OB.runtime_flags = dict()

static

status_map_otto

dictionary lib.ob.OB.status_map_otto

static

Initial value:

=  {
        T_STATE.RUNNING: "started",
        T_STATE.ABORTED: "aborted",
        T_STATE.FINISHED: "executed",
    }

tpmods

lib.ob.OB.tpmods = attr.ib(init=False, default=attr.Factory(list))

static

transitions_q

lib.ob.OB.transitions_q = attr.ib(init=False, default=None)

static

wrapper

lib.ob.OB.wrapper = attr.ib(init=False, default=None)

static

wrapper_task

lib.ob.OB.wrapper_task = attr.ib(init=False, default=None)

static

Property Documentation

context

lib.ob.OB.context = property

static

context property

---

The documentation for this class was generated from the following file:

• seq/seqlib/src/seq/lib/ob.py