laborchestrator.orchestrator_interface module

class laborchestrator.orchestrator_interface.FormalLabwareManagerInterface

Bases: object

_abc_impl = <_abc._abc_data object>

abstractmethod add_labware(labware: List[Any])

register one or several labware to orchestrator

check_labware_presence(process_names: List[str] = None) → Tuple[List[Any], List[Any]]

Checks whether in the database there is labware registered as required for the process. (In the process is defined, where what labware is required). :param process_names: By default, it is all processes, have been added but not started. :return: A tuple [found, missing]. The first entry is a list of information on existing labware

and the second entry lists requirements of missing labware

abstractmethod create_labware_location_graph(labware_ids: List[str])

history of labware movements (esp. colocations) current state labware distribution

abstractmethod remove_labware(labware_ids: List[str])

Removes one ore several labware from the orchestrator

class laborchestrator.orchestrator_interface.FormalLoggingInterface

Bases: object

_abc_impl = <_abc._abc_data object>

abstractmethod get_log(start_datetime: datetime = None, stop_datetime: datetime = None, level: int = 0)

default for stop: now

abstract property logging_level

add description

Type:

TODO

class laborchestrator.orchestrator_interface.FormalOrchestratorConfigInterface

Bases: object

Resources are all … excluding labware

_abc_impl = <_abc._abc_data object>

abstractmethod add_lab_resources_from_database(URI: str)

Defines and configures the lab environment (types, names, functionalities, etc. of devices). :param URI: Name of the Lab Environment configuration file in YAML format,

according to the specification TODO:[@Mark: insert link]

Returns:

abstractmethod add_lab_resources_from_file(lab_env_filename: str)

Defines and configures the lab environment (types, names, functionalities, etc. of devices). :param lab_env_filename: Name of the Lab Environment configuration file in YAML format,

according to the specification TODO:[@Mark: insert link]

Returns:

class laborchestrator.orchestrator_interface.FormalProcessControllerInterface

Bases: object

A class for managing one or more processes in a laboratory. You can add processes (also while others are already running), monitor their progress. The class incorporates an intelligent scheduler. It adapts the schedule to delays, errors, manual interference and addition of processes. Before really adding a process, you can check what the new schedule would look like. The progress can be observed via gantt charts or workflow graphs.

_abc_impl = <_abc._abc_data object>

abstractmethod add_process(description: str | None = None, file_path: str | None = None, name: str | None = None) → str

Adds a process to be orchestrated, read by the selected process reader You have to either specify a description (i.e. the file content) or a file location. If no name is specified, the process will be named like P_2 (enumerated). If the given name is already taken, it is added a suffix. :return: process object :raises ParsingError

abstract property available_processes: List[ProcessDescription]

Checks the file-system(currently not implemented) or database for available saved processes. All found processes are returned with the available information. :return: A List of tuples [Name, description(if found), filepath(if found)] for each found process

abstract property gantt_chart_executed_processes

returns the gantt chart including that process.

past :param processes: List of names of all processes to include in the gantt chart. By default, its all. :return: The gantt chart as plotly figure.

abstract property gantt_chart_scheduled_processes

returns the gantt chart including that process.

future

Parameters:

processes – List of names of all processes to include in the gantt chart. By default, its all.

Returns:

The gantt chart as plotly figure.

get_parameter(param_name: str)

Used as a flexible and easily extendable method to retrieve parameters of the orchestrator

abstractmethod get_process_state(process_name: str) → ProcessExecutionState

returns current state of the process :param process_name: :return: ProcessExecutionState

abstract property in_time: bool

timing state of the orchestrator

Returns:

_True, if orchestrator is currently in time (=not delayed) _

Return type:

bool

abstractmethod pause_processes(process_names: List[str] = None) → bool

Pausing a process of a given name :return: bool, if process could be paused

abstract property processes: List[ProcessInfo]

_Lists all process info for all current processes_

Raises:

NotImplementedError – _description_

Returns:

_description_

Return type:

List[str]

abstractmethod remove_processes(process_names: List[str], return_labwares: bool = False, final_device: str = None)

TODO: improve description Removes all information of the specified process. If the flag is set, the scheduler will return all involved labware to their original position. You can also specify a storage device, where all labware shall be brought.

Parameters:

• process_names – Unique name of the process to remove.

• return_labwares – Shall the scheduler bring all involved labwares to some location? By default it is their starting position.

• final_device – If return_labware is set, you can specify a device where all involved labware shall be brought.

Returns:

Nothing

abstractmethod restart_process_from_datetime(process_uri: str, start: datetime = None) → bool

Restarts a process from a given point in time. There has to be a database interface implemented. :param process_uri: unique uri for the database interface to find information on the process :param start: Point in time from where to restart. The default start point is the last known state. :return: bool, if the process could be restarted

abstractmethod resume_processes(process_names: List[str] = None) → bool

Resume form Pausing a process of a given name :return: bool, if all process could be resumed

abstractmethod select_process_reader(process_reader=ProcessReader.PYTHONLABREADER)

Dependency injection of the process reader

set_parameter(param_name: str, new_value)

Used as a flexible and easily extendable method to set parameters of the orchestrator

abstractmethod set_process_priority(process_name: str, priority: int)

Changes the priority of an existing process

abstractmethod simulate_all_processes(speed: float) → bool

TODO: description 0-600x ?. :return: bool if all processes could be started

abstract property simulation_speed: float

abstractmethod start_processes(process_names: List[str] = None) → bool

Starts the specified added process. This will cause an initial scheduling. :return: bool if the process could be started

abstractmethod stop_processes(process_names: List[str] = None) → bool

Stops the specified process. All running operations will continue, but no follow up operation will be started.

Parameters:

process_names – Unique name of the process to stop

Returns:

bool, if process could be stopped ?

abstract property workflow_graph_executed_processes

control-flow or workflow ? Creates a graphviz Digraph visualizing the progress of scheduled processes. future TODO: we could add different formats, defined by an WFGFormat(Enum)

Parameters:

processes – A list of process names that shall be visualized. By default all processes will be included.

Returns:

WorkFlowGraphViz

abstract property workflow_graph_scheduled_processes

control-flow or workflow ? Creates a graphviz Digraph visualizing the progress of scheduled processes. future TODO: we could add different formats, defined by an WFGFormat(Enum)

Parameters:

processes – A list of process names that shall be visualized. By default all processes will be included.

Returns:

WorkFlowGraphViz

class laborchestrator.orchestrator_interface.FormalProcessStepControllerInterface

Bases: object

_abc_impl = <_abc._abc_data object>

abstractmethod get_process_step_state(step_id: str)

TODO: add description

abstractmethod insert_process_step(process_step: Any, parent_step_ids: List[str], child_step_ids: List[str], process_id: str | None = None, waiting_cost: Dict[str, float] = None, max_waiting_time: Dict[str, float] = None)

Adds the given process step inbetween the given parents and children with the (optional) given time constraints. If process_id is omitted, there is assumed to be at least one child or parent step. The new step is then assigned to the same process

abstractmethod interrupt_process_step(step_id: str)

Stops the process step as soon as possible and considers it finished

abstractmethod process_step_executed_externally(step_id: str, result)

TODO: improve description There are jobs that have to be done externally, e.g. by a human or external device. Call this method to inform the scheduler a external process step has finished the specified operation.

Parameters:

• step_id – unique id of the step, a external entity was supposed to do

• result – In case the job had some results, those should be given here

abstractmethod remove_process_step(step_id: str)

TODO: add description

abstractmethod reset_error_state(step_id: str, repeat_operation: bool = False)

TODO: improve description In case an error need to be resolved manually, use this command afterwards to proceed with the process. :param step_id: Unique identifier for an operation :param repeat_operation: Boolean whether the operation shall be tried again :return: nothing

abstractmethod retry_process_step(step_id: str)

TODO: add description

class laborchestrator.orchestrator_interface.FormalSampleManagerInterface

Bases: object

_abc_impl = <_abc._abc_data object>

abstractmethod get_sample_labware(labware_id: str, sample_id)

TODO: add description

abstract property sample_location_graph

add description

Type:

TODO

class laborchestrator.orchestrator_interface.LabwareInfo(URI: str, barcode: str)

Bases: object

URI: str

barcode: str

class laborchestrator.orchestrator_interface.OrchestratorInterface

Bases: ABC, FormalOrchestratorConfigInterface, FormalProcessControllerInterface, FormalLoggingInterface, FormalLabwareManagerInterface, FormalProcessStepControllerInterface

_abc_impl = <_abc._abc_data object>

class laborchestrator.orchestrator_interface.ProcessDescription(name: str, description: str | None = None, file_path: str | None = None)

Bases: object

description: str | None = None

file_path: str | None = None

name: str

class laborchestrator.orchestrator_interface.ProcessExecutionState(*values)

Bases: Enum

_State of a Process Execution_ TODO: check SiLA specification

Parameters:

Enum (_type_) – _description_

FINISHED = 4

IDLE = 1

PAUSED = 3

RUNNING = 2

SCHEDULED = 5

class laborchestrator.orchestrator_interface.ProcessInfo(name: str, priority: int, state: laborchestrator.orchestrator_interface.ProcessExecutionState)

Bases: object

name: str

priority: int

state: ProcessExecutionState

exception laborchestrator.orchestrator_interface.ProcessParsingError(message)

Bases: Exception

class laborchestrator.orchestrator_interface.ProcessReader(*values)

Bases: Enum

PYTHONLABREADER = 1