laborchestrator.structures module

Here we define the structures of a process, a scheduling instance etc.

class laborchestrator.structures.Computation(name: str, var_name: str, evaluation: Callable[[Dict[str, Any]], Any], prior: List[str] = <factory>, result: Any = None, status: laborchestrator.structures.StepStatus = <StepStatus.WAITING: 0>, opacity: float = 1)

Bases: object

evaluation: Callable[[Dict[str, Any]], Any]

name: str

opacity: float = 1

prior: List[str]

result: Any = None

status: StepStatus = 0

var_name: str

class laborchestrator.structures.ContainerInfo(name: str, current_device: str, current_pos: int, start_device: UsedDevice, filled: bool = True, content: str = '', labware_type: str = '', barcode: str | None = None, finished: bool = False, lidded: bool = False, lid_site: Any = None, in_error_state: bool = False, is_reagent: bool = False)

Bases: object

Editable data collection for containers going through a lab process.

barcode: str | None = None

content: str = ''

current_device: str

current_pos: int

filled: bool = True

finished: bool = False

in_error_state: bool = False

is_reagent: bool = False

labware_type: str = ''

lid_site: Any = None

lidded: bool = False

name: str

start_device: UsedDevice

class laborchestrator.structures.IfNode(name: str, evaluation: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], bool], decision: bool | None = None, prior: ~typing.List[str] = <factory>, true_tree: ~typing.List[str] = <factory>, false_tree: ~typing.List[str] = <factory>, status: ~laborchestrator.structures.StepStatus = StepStatus.WAITING, opacity: float = 1)

Bases: object

Represents a constraint in the workflow graph

decision: bool | None = None

evaluation: Callable[[Dict[str, Any]], bool]

false_tree: List[str]

name: str

opacity: float = 1

prior: List[str]

status: StepStatus = 0

true_tree: List[str]

class laborchestrator.structures.MoveStep(name: str, cont_names: ~typing.List[str], function: str, duration: float, process_name: str = '', status: ~laborchestrator.structures.StepStatus = StepStatus.WAITING, start: ~datetime.datetime | None = None, finish: ~datetime.datetime | None = None, prior: ~typing.List[str] = <factory>, used_devices: ~typing.List[~laborchestrator.structures.UsedDevice] = <factory>, label: str = 'LABEL', wait_cost: ~typing.Dict[str, float] = <factory>, max_wait: ~typing.Dict[str, int] = <factory>, min_wait: ~typing.Dict[str, int] = <factory>, wait_to_start_costs: float = 0, is_start: bool = False, opacity: float = 1, result: ~typing.Any = None, data: ~typing.Dict = <factory>, pref_dest_pos: int | None = None, destination_pos: int = 0, origin_pos: int = 0)

Bases: ProcessStep

Operation resembling a movement of a container

destination_pos: int = 0

property origin_device: UsedDevice | None

origin_pos: int = 0

pref_dest_pos: int | None = None

property target_device: UsedDevice | None

class laborchestrator.structures.ProcessStep(name: str, cont_names: ~typing.List[str], function: str, duration: float, process_name: str = '', status: ~laborchestrator.structures.StepStatus = StepStatus.WAITING, start: ~datetime.datetime | None = None, finish: ~datetime.datetime | None = None, prior: ~typing.List[str] = <factory>, used_devices: ~typing.List[~laborchestrator.structures.UsedDevice] = <factory>, label: str = 'LABEL', wait_cost: ~typing.Dict[str, float] = <factory>, max_wait: ~typing.Dict[str, int] = <factory>, min_wait: ~typing.Dict[str, int] = <factory>, wait_to_start_costs: float = 0, is_start: bool = False, opacity: float = 1, result: ~typing.Any = None, data: ~typing.Dict = <factory>)

Bases: object

Editable data collection for a general operations in a lab process.

property cont

cont_names: List[str]

data: Dict

duration: float

finish: datetime | None = None

function: str

is_start: bool = False

label: str = 'LABEL'

property main_device: UsedDevice | None

max_wait: Dict[str, int]

min_wait: Dict[str, int]

name: str

opacity: float = 1

prior: List[str]

priors_done(job_by_name: Dict)

Checks whether all prerequisite operations are finished

process_name: str = ''

result: Any = None

start: datetime | None = None

status: StepStatus = 0

used_devices: List[UsedDevice]

wait_cost: Dict[str, float]

wait_to_start_costs: float = 0

class laborchestrator.structures.SMProcess(name: str, priority: int = 0)

Bases: object

Encapsulates the information about an experimental process (as workflow graph) and provides utility functions. This structure is independent of the process description language

_status: ProcessExecutionState

computations: List[Computation]

containers: List[ContainerInfo]

experiment_uuid: str

get_info() → str

if_nodes: List[IfNode]

min_start: datetime | None

name: str

property starting_nodes: list[ProcessStep]

property status

This automatically sets the status to FINISHED if all steps are finished :return:

steps: List[ProcessStep]

update_reagent_opacity()

This method sets the opacity of all operations of reagents :return:

variables: List[Variable]

property wfg: DiGraph

Having the workflow as nx.DiGraph is very useful, but saving it in memory means a second ‘source of truth’. So, we create it on the fly. :return: the whole workflow as nx.DiGraph

class laborchestrator.structures.Schedule

Bases: Dict[str, ScheduledAssignment]

class laborchestrator.structures.ScheduledAssignment(start: datetime.datetime, machine_prior: List[str] = <factory>, participating: Dict[str, str] = <factory>)

Bases: object

property device: str | None

name of the device this job is scheduled on

Type:

return

machine_prior: List[str]

participating: Dict[str, str]

start: datetime

class laborchestrator.structures.SchedulingInstance

Bases: object

add_process(process: SMProcess)

property combined_wfg: DiGraph

computation_by_id: Dict[str, Computation]

container_info_by_name: Dict[str, ContainerInfo]

property definite_computation_by_id

property definite_if_node_by_id

property definite_step_by_id: Dict[str, ProcessStep]

property definite_var_by_id

deleted_operable: Dict[str, ProcessStep | Variable | IfNode | Computation]

gannt_chart()

if_node_by_id: Dict[str, IfNode]

naming_consistent(process: SMProcess)

Check whether the names of the jobs correspond to the indices in the workflow graph.

property operable: Dict[str, ProcessStep | Variable | IfNode | Computation]

process_by_name: Dict[str, SMProcess]

property process_stati_by_name: Dict[str, ProcessExecutionState]

remove_operable(idx: str)

Deletes a process step, variable, computation or decision from the structure including all references.

remove_process(process_id)

Purges the process and its content from all dictionaries :param process_id: :return:

property running_processes_names: List[str]

return a list of the names of all running processes

schedule: Schedule

schedule_violated(tolerance: float = 10) → bool

set_schedule(schedule: Schedule)

saves the schedule and writes its information into all affected job info :param schedule: :return:

start_process(process_id)

step_by_id: Dict[str, ProcessStep]

stop_process(process_id)

unique_job_names(process: SMProcess)

Checks whether none of the processes operations names is already given to an operation.

update_reagent_opacity()

This method sets the opacity of all operations of reagent :return:

var_by_id: Dict[str, Variable]

visualize_wfg()

class laborchestrator.structures.StepStatus(*values)

Bases: IntEnum

ERROR = 3

FINISHED = 2

RUNNING = 1

WAITING = 0

class laborchestrator.structures.UsedDevice(device_type: Any, name: str | None = None, preferred: str | None = None, tag: str = '')

Bases: object

device_type: Any

name: str | None = None

preferred: str | None = None

tag: str = ''

class laborchestrator.structures.Variable(name: str, var_name: str, result: ~typing.Any = None, prior: ~typing.List[str] = <factory>, status: ~laborchestrator.structures.StepStatus = StepStatus.WAITING, opacity: float = 1)

Bases: object

Representing a variable node of the workflow graph

name: str

opacity: float = 1

prior: List[str]

result: Any = None

status: StepStatus = 0

var_name: str