sio3pack.workflow ================= .. py:module:: sio3pack.workflow Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/sio3pack/workflow/execution/index Package Contents ---------------- .. py:class:: Object(handle) A class to represent an object in a workflow. A object is a file, stored either locally or remotely. :param str handle: The handle of the object. .. py:method:: replace_templates(replacements) Replace strings in the object with the given replacements. :param dict[str, str] replacements: The replacements to make. .. py:class:: ObjectList A class to represent a list of objects in a workflow. .. py:method:: append(obj) Append an object to the list. :param Object obj: The object to append. .. py:method:: extend(objects) Extend the list with the given objects. :param list[Object] objects: The objects to extend the list with. .. py:method:: union(other) Union the list with another list of objects. :param ObjectList other: The other list to union with. .. py:class:: ObjectsManager A class to manage objects in a workflow. Allows creation, retrieval, and management of objects. .. py:method:: create_object(handle) Create and return a new object. :param str handle: The handle of the object. :return: The created object. .. py:method:: add_object(obj) Add an object to the manager. :param Object obj: The object to add. .. py:method:: get_object(handle) Get an object by its handle. :param str handle: The handle of the object. .. py:method:: get_or_create_object(handle) Get an object by its handle, creating it if it does not exist. :param str handle: The handle of the object. .. py:method:: find_by_regex_in_objects(regex, return_group) Find all occurrences of a regex in the task. :param regex: The regex to search for. :param return_group: The group to return. :return: A list of matches. .. py:class:: ExecutionTask(name, workflow, exclusive = False, hard_time_limit = None, extra_limit = None, output_register = None, pid_namespaces = 1, processes = None, pipes = 0, channels = None) Bases: :py:obj:`Task` A task that executes a program. :param name: The name of the task. :param Workflow workflow: The workflow the task belongs to. :param bool exclusive: Whether the task is exclusive. :param int hard_time_limit: The hard time limit. :param int extra_limit: If set, the hard_time_limit for the task will be the maximum time limit of all resource groups plus this value. :param int | str output_register: The output register of the task. :param int pid_namespaces: The number of PID namespaces. :param list[Process] processes: The processes of the task. :param int pipes: The number of pipes available to the task. :param list[Channel] channels: Configuration of the channels for the task. .. py:method:: from_json(data, workflow) :classmethod: Create a new execution task from a dictionary. :param dict data: The dictionary to create the task from. :param Workflow workflow: The workflow the task belongs to. .. py:method:: to_json(reg_map = None) Convert the task to a dictionary. :param reg_map: A mapping of register names to register numbers. :return dict: The dictionary representation of the task. .. py:method:: add_filesystem(filesystem) Add a filesystem to the task. :param Filesystem filesystem: The filesystem to add. .. py:method:: add_mount_namespace(mount_namespace) Add a mount namespace to the task. :param MountNamespace mount_namespace: The mount namespace to add. .. py:method:: add_resource_group(resource_group) Add a resource group to the task. :param ResourceGroup resource_group: The resource group to add. .. py:method:: add_process(process) Add a process to the task. :param Process process: The process to add. .. py:method:: replace_templates(replacements) Replace strings in the task with the given replacements. :param replacements: The replacements to make. .. py:class:: ScriptTask(name, workflow, reactive = False, input_registers = None, output_registers = None, objects = None, script = None) Bases: :py:obj:`Task` A task that runs a script. :param str name: The name of the task. :param Workflow workflow: The workflow the task belongs to. :param bool reactive: Whether the task is reactive. :param list[int | str] input_registers: The input registers of the task. :param list[int | str] output_registers: The output registers of the task. :param list[Object] objects: The objects the task uses. :param str script: The script to run. .. py:method:: from_json(data, workflow) :classmethod: Create a new script task from a dictionary. :param data: The dictionary to create the task from. :param workflow: The workflow the task belongs to. .. py:method:: to_json(reg_map = None) Convert the task to a dictionary. :param reg_map: A mapping of register names to register numbers. :return: The dictionary representation of the task. .. py:method:: replace_templates(replacements) Replace strings in the task with the given replacements. :param replacements: The replacements to make. .. py:class:: Task Base class for a task. .. py:method:: from_json(data, workflow) :classmethod: Create a new task from a dictionary. :param dict data: The dictionary to create the task from. :param Workflow workflow: The workflow the task belongs to. .. py:method:: to_json(reg_map = None) :abstractmethod: Convert the task to a dictionary. :param reg_map: A mapping of register names to register numbers. :return dict: The dictionary representation of the task. .. py:method:: replace_templates(replacements) :abstractmethod: Replace strings in the task with the given replacements. :param replacements: The replacements to make. .. py:class:: Workflow(name, external_objects = None, observable_objects = None, observable_registers = 0, tasks = None) A class to represent a workflow. Number of registers is not required, as it is calculated automatically. :param str name: The name of the workflow. :param ObjectList external_objects: The external objects used in the workflow. :param ObjectList observable_objects: The observable objects used in the workflow. :param int observable_registers: The number of observable registers used in the workflow. :param list[Task] tasks: The tasks in the workflow. .. py:method:: from_json(data) :classmethod: Create a new workflow from a dictionary. :param data: The dictionary to create the workflow from. .. py:method:: get_num_registers() Get number of currently used registers. .. py:method:: only_string_registers() Check if all registers in the workflow are strings. :return bool: True if all registers are strings, False otherwise. .. py:method:: to_json(to_int_regs = False) Convert the workflow to a dictionary. :param bool to_int_regs: Whether to convert registers to integers. :return dict: The dictionary representation of the workflow. .. py:method:: add_task(task) Add a task to the workflow. :param Task task: The task to add. .. py:method:: get_prog_files() :abstractmethod: Get all program files in the workflow. :return: A list of program files. .. py:method:: add_external_object(obj) Add an external object to the workflow. :param Object obj: The object to add. .. py:method:: add_observable_object(obj) Add an observable object to the workflow. :param Object obj: The object to add. .. py:method:: replace_templates(replacements) Replace strings in the workflow with the given replacements. :param dict[str, str] replacements: The replacements to make. .. py:method:: find_by_regex_in_objects(regex, return_group) Find all occurrences of the given regex in the workflow's objects. :param str regex: The regex to search for. :param int return_group: The group to return. :return: A list of occurrences. .. py:method:: union(other) Add another workflow to this workflow. Merge all objects and tasks. :param Workflow other: Other workflow to merge into this. .. py:class:: WorkflowManager(package, workflows) 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. :param Package package: The package for which the workflows are managed. :param dict[str, Workflow] workflows: A dictionary of user defined workflows, where the key is the workflow name and the value is the Workflow object. .. py:method:: get(name) Get the workflow with the given name. If the workflow does not exist, return default workflow for this name from :meth:`WorkflowManager.get_default`. :param str name: The name of the workflow. :return: The workflow with the given name. .. py:method:: all() Get all user defined workflows. :return: A dictionary of all workflows. .. py:method:: get_default(name) Get the default workflow for the given name. This method should be overridden by subclasses to provide the default workflow for the given name. :param name: The name of the workflow. :return: The default workflow for the given name. .. py:method:: get_prog_files() :abstractmethod: Get all program files used in all graphs. .. py:method:: get_compile_file_workflow(file) 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. :param File | str file: The file (or the path to the file) to compile. :return: A tuple of the workflow and the path to the compiled file. .. py:method:: get_unpack_operation(has_test_gen, has_verify, return_func = None) Get the operation for unpacking the package. :param bool has_test_gen: Whether the package has test generation. :param bool has_verify: Whether the package has verification. :param callable return_func: A function to call with the results of the workflow execution. .. py:method:: get_run_operation(program, tests = None, return_func = None) :abstractmethod: Get the operation for running the program with the given tests. :param File program: The program file to run. :param list[Test] | None tests: The list of tests to run the program with or None if all tests should be run. :param callable return_func: A function to call with the results of the workflow execution. .. py:method:: get_user_out_operation(program, test, return_func = None) :abstractmethod: Get the operation for running the user output with the given program and test. :param File program: The program file to run. :param Test test: The test to run the program with. :param callable return_func: A function to call with the results of the workflow execution. .. py:method:: get_test_run_operation(program, test, return_func = None) :abstractmethod: Get the operation for running the test with the given program. :param File program: The program file to run. :param File test: The test file to run the program with. :param callable return_func: A function to call with the results of the workflow execution. .. py:class:: WorkflowOperation(get_workflow_func, return_results = False, return_results_func = None, *wf_args, **wf_kwargs) A class to handle workflow operations, allowing for the retrieval of workflows and the return of results from those workflows. :param callable get_workflow_func: Function to retrieve workflows. :param bool return_results: Whether to return results from the workflow. :param callable return_results_func: Function to handle returning results. :param list[Any] wf_args: Additional positional arguments for the workflow function. :param dict[str, Any] wf_kwargs: Additional keyword arguments for the workflow function. .. py:method:: get_workflow() 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.