laborchestrator.pythonlab_reader module

Contains an implementation of ProcessReader for PythonLab processes

class laborchestrator.pythonlab_reader.PythonLabReader

Bases: ProcessReader

_abc_impl = <_abc._abc_data object>

static costs_from_prio(prio: float)

static fill_transfer_times(smp)

static find_origin(idx, data: Dict[str, Any], topo_sort: List[str], g: DiGraph) → UsedDevice | None

static get_available_processes(file_dir: str) → List[Any]

Searches a directory for available process descriptions. :param file_dir: :return: A list containing all found process descriptions (in their native language).

static preprocess_wfg(plp: PLProcess)

Some regularisation, translation of priorities into waiting costs and filling of missing values.

static read_computations(g: DiGraph)

static read_containers(g: DiGraph, p: PLProcess)

static read_if_nodes(g: DiGraph)

static read_jobs(g: DiGraph)

Extracts all information on jobs in a workflow. The attributes of the jobs are assumed to be in the data-dictionary which networkx provides for each node.

Parameters:

g – The graph containing the whole workflow

Returns:

A list of all jobs described in the graph

static read_precedences(smp: SMProcess, g: DiGraph)

static read_process(process: PLProcess | None = None, file_path: str | None = None, name: str | None = None, src=None, **kwargs) → SMProcess

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

static read_variables(g: DiGraph)

static read_wait_cons(smp: SMProcess, g: DiGraph)

static relabel_nodes(p, prefix)

Adds ‘[ProcessName]’ to all node labels (This makes them unique in the schedule, if the ProcessName is unique)

laborchestrator.pythonlab_reader.read_process_test(process: ~pathlib.Path = <typer.models.OptionInfo object>, show_wfg: bool = <typer.models.OptionInfo object>)

This is intended for testing the import and reading of a PythonLab process.

laborchestrator.pythonlab_reader.try_to_read_process()