sio3pack.workflow

Submodules

Package Contents

class sio3pack.workflow.Object(handle)[source]

A class to represent an object in a workflow. A object is a file, stored either locally or remotely.

Parameters:

handle (str) – The handle of the object.

replace_templates(replacements)[source]

Replace strings in the object with the given replacements.

Parameters:

  • replacements (dict[str, str]) – The replacements to make.

  • replacements

class sio3pack.workflow.ObjectList[source]

A class to represent a list of objects in a workflow.

append(obj)[source]

Append an object to the list.

Parameters:

obj (Object) – The object to append.

extend(objects)[source]

Extend the list with the given objects.

Parameters:

objects (list[Object]) – The objects to extend the list with.

union(other)[source]

Union the list with another list of objects.

Parameters:

other (ObjectList) – The other list to union with.

class sio3pack.workflow.ObjectsManager[source]

A class to manage objects in a workflow. Allows creation, retrieval, and management of objects.

create_object(handle)[source]

Create and return a new object.

Parameters:

handle (str) – The handle of the object.

Returns:

The created object.

Return type:

Object

add_object(obj)[source]

Add an object to the manager.

Parameters:

obj (Object) – The object to add.

get_object(handle)[source]

Get an object by its handle.

Parameters:

handle (str) – The handle of the object.

Return type:

Object

get_or_create_object(handle)[source]

Get an object by its handle, creating it if it does not exist.

Parameters:

handle (str) – The handle of the object.

Return type:

Object

find_by_regex_in_objects(regex, return_group)[source]

Find all occurrences of a regex in the task.

Parameters:

  • regex (str) – The regex to search for.

  • return_group (int) – The group to return.

Returns:

A list of matches.

Return type:

list[str]

class sio3pack.workflow.ExecutionTask(name, workflow, exclusive=False, hard_time_limit=None, extra_limit=None, output_register=None, pid_namespaces=1, processes=None, pipes=0, channels=None)[source]

Bases: Task

A task that executes a program.

Parameters:

  • name (str) – The name of the task.

  • workflow (Workflow) – The workflow the task belongs to.

  • exclusive (bool) – Whether the task is exclusive.

  • hard_time_limit (int) – The hard time limit.

  • extra_limit (int) – If set, the hard_time_limit for the task will be the maximum time limit of all resource groups plus this value.

  • output_register (int | str) – The output register of the task.

  • pid_namespaces (int) – The number of PID namespaces.

  • processes (list[Process]) – The processes of the task.

  • pipes (int) – The number of pipes available to the task.

  • channels (list[Channel]) – Configuration of the channels for the task.

  • output_register

classmethod from_json(data, workflow)[source]

Create a new execution task from a dictionary.

Parameters:

  • data (dict) – The dictionary to create the task from.

  • workflow (Workflow) – The workflow the task belongs to.

to_json(reg_map=None)[source]

Convert the task to a dictionary.

Parameters:

reg_map (dict[str, int]) – A mapping of register names to register numbers.

Return dict:

The dictionary representation of the task.

Return type:

dict

add_filesystem(filesystem)[source]

Add a filesystem to the task.

Parameters:

filesystem (Filesystem) – The filesystem to add.

add_mount_namespace(mount_namespace)[source]

Add a mount namespace to the task.

Parameters:

mount_namespace (MountNamespace) – The mount namespace to add.

add_resource_group(resource_group)[source]

Add a resource group to the task.

Parameters:

resource_group (ResourceGroup) – The resource group to add.

add_process(process)[source]

Add a process to the task.

Parameters:

process (Process) – The process to add.

replace_templates(replacements)[source]

Replace strings in the task with the given replacements.

Parameters:

replacements (dict[str, str]) – The replacements to make.

class sio3pack.workflow.ScriptTask(name, workflow, reactive=False, input_registers=None, output_registers=None, objects=None, script=None)[source]

Bases: Task

A task that runs a script.

Parameters:

  • name (str) – The name of the task.

  • workflow (Workflow) – The workflow the task belongs to.

  • reactive (bool) – Whether the task is reactive.

  • input_registers (list[int | str]) – The input registers of the task.

  • output_registers (list[int | str]) – The output registers of the task.

  • objects (list[Object]) – The objects the task uses.

  • script (str) – The script to run.

  • input_registers

  • output_registers

classmethod from_json(data, workflow)[source]

Create a new script task from a dictionary.

Parameters:

  • data (dict) – The dictionary to create the task from.

  • workflow (Workflow) – The workflow the task belongs to.

to_json(reg_map=None)[source]

Convert the task to a dictionary.

Parameters:

reg_map (dict[str, int]) – A mapping of register names to register numbers.

Returns:

The dictionary representation of the task.

Return type:

dict

replace_templates(replacements)[source]

Replace strings in the task with the given replacements.

Parameters:

replacements (dict[str, str]) – The replacements to make.

class sio3pack.workflow.Task[source]

Base class for a task.

classmethod from_json(data, workflow)[source]

Create a new task from a dictionary.

Parameters:

  • data (dict) – The dictionary to create the task from.

  • workflow (Workflow) – The workflow the task belongs to.

abstractmethod to_json(reg_map=None)[source]

Convert the task to a dictionary.

Parameters:

reg_map (dict[str, int]) – A mapping of register names to register numbers.

Return dict:

The dictionary representation of the task.

Return type:

dict

abstractmethod replace_templates(replacements)[source]

Replace strings in the task with the given replacements.

Parameters:

replacements (dict[str, str]) – The replacements to make.

class sio3pack.workflow.Workflow(name, external_objects=None, observable_objects=None, observable_registers=0, tasks=None)[source]

A class to represent a workflow. Number of registers is not required, as it is calculated automatically.

Parameters:

  • name (str) – The name of the workflow.

  • external_objects (ObjectList) – The external objects used in the workflow.

  • observable_objects (ObjectList) – The observable objects used in the workflow.

  • observable_registers (int) – The number of observable registers used in the workflow.

  • tasks (list[Task]) – The tasks in the workflow.

classmethod from_json(data)[source]

Create a new workflow from a dictionary.

Parameters:

data (dict) – The dictionary to create the workflow from.

get_num_registers()[source]

Get number of currently used registers.

Return type:

int

only_string_registers()[source]

Check if all registers in the workflow are strings.

Return bool:

True if all registers are strings, False otherwise.

Return type:

bool

to_json(to_int_regs=False)[source]

Convert the workflow to a dictionary.

Parameters:

to_int_regs (bool) – Whether to convert registers to integers.

Return dict:

The dictionary representation of the workflow.

Return type:

dict

add_task(task)[source]

Add a task to the workflow.

Parameters:

task (Task) – The task to add.

abstractmethod get_prog_files()[source]

Get all program files in the workflow.

Returns:

A list of program files.

Return type:

list[sio3pack.files.file.File]

add_external_object(obj)[source]

Add an external object to the workflow.

Parameters:

obj (Object) – The object to add.

add_observable_object(obj)[source]

Add an observable object to the workflow.

Parameters:

obj (Object) – The object to add.

replace_templates(replacements)[source]

Replace strings in the workflow with the given replacements.

Parameters:

  • replacements (dict[str, str]) – The replacements to make.

  • replacements

find_by_regex_in_objects(regex, return_group)[source]

Find all occurrences of the given regex in the workflow’s objects.

Parameters:

  • regex (str) – The regex to search for.

  • return_group (int) – The group to return.

Returns:

A list of occurrences.

Return type:

list[str]

union(other)[source]

Add another workflow to this workflow. Merge all objects and tasks.

Parameters:

other (Workflow) – Other workflow to merge into this.

class sio3pack.workflow.WorkflowManager(package, workflows)[source]

A class to manage workflows for a package. Allows to get workflows by name, manages default and user created workflows, and provides methods to get workflows for compiling files, generating tests, and verifying results. This class can be overridden to provide custom workflows in other package types.

Parameters:

  • package (Package) – The package for which the workflows are managed.

  • workflows (dict[str, Any]) – A dictionary of user defined workflows, where the key is the workflow name and the value is the Workflow object.

  • workflows

get(name)[source]

Get the workflow with the given name. If the workflow does not exist, return default workflow for this name from WorkflowManager.get_default().

Parameters:

name (str) – The name of the workflow.

Returns:

The workflow with the given name.

Return type:

sio3pack.workflow.workflow.Workflow

all()[source]

Get all user defined workflows.

Returns:

A dictionary of all workflows.

Return type:

dict[str, sio3pack.workflow.workflow.Workflow]

get_default(name)[source]

Get the default workflow for the given name. This method should be overridden by subclasses to provide the default workflow for the given name.

Parameters:

name (str) – The name of the workflow.

Returns:

The default workflow for the given name.

Return type:

sio3pack.workflow.workflow.Workflow

abstractmethod get_prog_files()[source]

Get all program files used in all graphs.

Return type:

list[sio3pack.files.File]

get_compile_file_workflow(file)[source]

A helper function to get a workflow for compiling the given file. The file should be a program file, not a test file. Returns a tuple of the workflow and the path to the compiled file. The files are not added as external or observable objects, since they don’t have to be.

Parameters:
Returns:

A tuple of the workflow and the path to the compiled file.

Return type:

tuple[sio3pack.workflow.workflow.Workflow, str]

get_unpack_operation(has_test_gen, has_verify, return_func=None)[source]

Get the operation for unpacking the package.

Parameters:

  • has_test_gen (bool) – Whether the package has test generation.

  • has_verify (bool) – Whether the package has verification.

  • return_func (callable) – A function to call with the results of the workflow execution.

Return type:

sio3pack.workflow.workflow_op.WorkflowOperation

abstractmethod get_run_operation(program, tests=None, return_func=None)[source]

Get the operation for running the program with the given tests.

Parameters:

  • program (File) – The program file to run.

  • tests (list[sio3pack.test.Test] | None) – The list of tests to run the program with or None if all tests should be run.

  • return_func (callable) – A function to call with the results of the workflow execution.

  • tests

Return type:

sio3pack.workflow.workflow_op.WorkflowOperation

abstractmethod get_user_out_operation(program, test, return_func=None)[source]

Get the operation for running the user output with the given program and test.

Parameters:

  • program (File) – The program file to run.

  • test (Test) – The test to run the program with.

  • return_func (callable) – A function to call with the results of the workflow execution.

Return type:

sio3pack.workflow.workflow_op.WorkflowOperation

abstractmethod get_test_run_operation(program, test, return_func=None)[source]

Get the operation for running the test with the given program.

Parameters:

  • program (File) – The program file to run.

  • test (File) – The test file to run the program with.

  • return_func (callable) – A function to call with the results of the workflow execution.

Return type:

sio3pack.workflow.workflow_op.WorkflowOperation

class sio3pack.workflow.WorkflowOperation(get_workflow_func, return_results=False, return_results_func=None, *wf_args, **wf_kwargs)[source]

A class to handle workflow operations, allowing for the retrieval of workflows and the return of results from those workflows.

Parameters:

  • get_workflow_func (callable) – Function to retrieve workflows.

  • return_results (bool) – Whether to return results from the workflow.

  • return_results_func (callable) – Function to handle returning results.

  • wf_args (list[Any]) – Additional positional arguments for the workflow function.

  • wf_kwargs (dict[str, Any]) – Additional keyword arguments for the workflow function.

get_workflow()[source]

A function to retrieve workflows. It yields workflows until the last one is reached. This function uses the provided get_workflow_func to get workflows and yields them one by one.