lib.ob.OB Class Reference

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

Public Member Functions
def	context (self)
	context property More...
def	reload_tpl (self, i)
	dummy, does nothing More...
def	load_tpl (self, t)
def	run_from_file (self, fn)
def	run_all (self)
def	wrap_modules (self)
	Instances the Sequence and wraps them in the ROOT Sequence. More...
def	wrap_and_run (self)
def	get_tree (self)
def	reset_tree (self)
def	resume (self)
def	retry_resume (self)
def	resume_node (self, serial_number)
def	abort (self)
def	nodes (self)
def	tplist (self)
def	find_node (self, nid)
	Finds a node in the tree based on its id. More...
def	find_node_by_number (self, serial_number)
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)
def	mark_pause (self, serial_number, flag=True)
def	flip_flag (self, serial_number, flag)
def	flip_pause (self, serial_number)
def	flip_skip (self, serial_number)
def	set_params (self, v)
	_----------------------------------------------------— More...
def	notify_state_change (self, node, ts=None)
	Published in the queue a node's state change. More...
def	reset (self)
def	load_ob (self, obname)

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

Public Attributes
	cls

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()

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

context()

def lib.ob.OB.context

(

self

)

context property

create_sequence()

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

static

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
	)

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 _----------------------------------------------------—

get_tree()

def lib.ob.OB.get_tree

(

self

)

dict of dicts representing the tree

load_ob()

def lib.ob.OB.load_ob	(		self,
			obname
	)

Loads OB file (json)

load_tpl()

def lib.ob.OB.load_tpl	(		self,
			t
	)

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
	)

mark_skip()

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

nodes()

def lib.ob.OB.nodes

(

self

)

Gets tree nodes

notify_state_change()

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

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

)

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
	)

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.

set_params()

def lib.ob.OB.set_params	(		self,
			v
	)

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

set_runtime_flag()

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

Stores a node's runtime flag in dictionary.

tplist()

def lib.ob.OB.tplist

(

self

)

Get list of modules loaded

wrap_and_run()

def lib.ob.OB.wrap_and_run

(

self

)

This is the important one.

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

wrap_modules()

def lib.ob.OB.wrap_modules

(

self

)

Instances the Sequence and wraps them in the ROOT Sequence.

Member Data Documentation

cls

lib.ob.OB.cls

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

queue

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

static

runtime_flags

lib.ob.OB.runtime_flags = dict()

static

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

---

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

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