pythonlab.pythonlab_reader module

Module with all necessary utility to parse a PythonLab process into a networkx graph. todo: use pydentic to parse it into defined dataclasses

class pythonlab.pythonlab_reader.PLProcessReader

Bases: object

static ast_get_process(process_type, src=None)

static contract_dummys(g: DiGraph)

This function contracts the dummy nodes in the workflow graph. To keep track of which labware participated in which process step throughout exploring if-clauses, break commands, etc. the reader includes dummy nodes into the workflow graph. In this function (after the whole process was read) these dummies are removed by contraction, i.e., remove them and connect all sources of incoming edges to all targets of outgoing edges.

static execute_scope(body: List[AST], scope: Dict[str, Any], runtime_vars: List[str], src: str, devices: List[str], plp: PLProcessSimulator)

Goes through the list of code lines and constructs the workflow graph by manipulation the process state and executing certain lines of process description code. Ment to be called recursively :param plp: :param devices: List of variables that correspond to ServiceResources :param src: source code of the process description :param body: list of parsed code fragments :param scope: variables assigned so var :param runtime_vars: list of variable names in current scope that are evaluated at runtime :return: Nothing

static handle_assign(asg: Assign, scope: Dict[str, Any], runtime_vars: List[str], src: str, devices: List[str], plp: PLProcessSimulator)

static handle_break(scope: Dict[str, Any], plp: PLProcessSimulator)

static handle_expr(expr: Expr, scope: Dict[str, Any], runtime_vars: List[str], src: str)

static handle_for(node: For, scope: Dict[str, Any], runtime_vars: List[str], src: str, devices: List[str], plp: PLProcessSimulator)

static handle_if(node: If, scope: Dict[str, Any], runtime_vars: List[str], src: str, devices: List[str], plp: PLProcessSimulator)

static parse_process(plp: PLProcess, src=None, scope: dict | None = None)

The main function. It takes a PythonLabProcess and derives a SMProcess from it. The Information is stored in Job and LabwareInfo classes from the structures-module.

static parse_process_from_file_path(file_path: str | Path)

static parse_process_from_instance(plp: PLProcess)

static parse_process_from_source_code(src: str)

class pythonlab.pythonlab_reader.PLProcessSimulator(process_type: Type[PLProcess])

Bases: PLProcess, ABC

A utility class to help going through actions in a PythonLabProcess while linking them correctly. This is done by systematically recursively exploring the abstract syntax tree and executing all functions of ServiceResources while keeping track of the labware. Each call of a function of a ServiceResource adds a new process step node to the workflow graph. The edges are determined by consecutive participation of labware in two steps.

_abc_impl = <_abc._abc_data object>

add_break_node()

add_computation(name, var_names, fct_code, needed_scope)

add_edge(s, t, **kwargs)

add_if_node(var_names, decision_code, needed_scope)

add_node(attr)

add_process_step(resource: ServiceResource, labware: List[LabwareResource], is_movement: bool = False, **kwargs)

This Method will be overwritten when the process is parsed. It is designed for service resources to add process steps to the workflow. :param service: :param labware: :param is_movement: :param kwargs: :return:

add_var_nodes(*args)

finalize_break_node(break_node)

finalize_if_construction(if_node, orig_state, last_break=-1)

get_state()

handle_labels(t, **kwargs)

Very preliminary handling of relations between process steps independent of labware :param t: :param kwargs: :return:

join_state(status)

prepare_false_execution(if_node)

prepare_true_execution(if_node)

register_labware_resource(resource)

set_starting_position(resource: LabwareResource, device: ServiceResource, position: int)

This method gets called when setting the starting position of a labware resource. It exists to be overwritten. :param resource: :param device: :param position: :return:

set_state(status)

visualize_workflow_graph(show=True)

pythonlab.pythonlab_reader.find_pl_process_class(module: ModuleType) → type[PLProcess]

pythonlab.pythonlab_reader.get_source_code(src: str, node: AST) → str