task

this module contains different types of tasks and some helper functions

class mosaic_orchestrator.task.Calculated

Used to mark a input of a subtask as calculated by the current task.

This means that the input is no longer settable from a higher hierarchy. Calculated inputs need to be passed to the run method of the subtask.

Example

This example shows how to set Calculated inputs for a child task:

class Addition(Task):
    summand1 = Input()
    summand2 = Input()

    def run(self, *args, **kwargs):
        return self.summand1.value + self.summand2.value


class MyTask(Task):
    t1 = Addition(inputs={"summand1": Calculated, "summand2": Calculated})

    def run(self, *args, **kwargs):
        t1_result = self.t1.run(inputs={"summand1":2, "summand2":3}) # will be 5
        ...

exception mosaic_orchestrator.task.CalculatedAccessException

raised if a calculated input

class mosaic_orchestrator.task.CheckResult(success: bool = True)

Encapsulates the results of a check of the dependencies and validations of a task hierarchy.

success

when True no errors were found and the task tree is ready to build and run.

inputs

a list of all settable inputs and their values.

errors

a list of errors. See ErrorType for the different types of errors.

append_error(error_type: ErrorType, message, source)

appends the given error

append_if_error(error_type: ErrorType, result: Result, source)

appends the given error only if the given result is failure

append_input(path: str, input: Input, parent_task: Task)

appends the given input

class mosaic_orchestrator.task.Const(value)

Used to mark a input of a subtask as constant, i.e. not settable from a higher hierarchy.

Example

When a value is passed to the constructor it is used for the target input. It is also possible to use the Const type as a marker to indicate that the default value of the target input shall be used als a constant:

class Addition(Task):
    summand1 = Input()
    summand2 = Input(default=0)

    def run(self, *args, **kwargs):
        return self.summand1.value + self.summand2.value


class MyTask(Task):
    t1 = Addition(inputs={
        "summand1": Const(2),
        "summand2": Const # will be fixed at default value 0
    })

    def run(self, *args, **kwargs):
        t1_result = self.t1.run() # will be 2 = (2 + 0)
        ...

class mosaic_orchestrator.task.Input(type: Optional[Type[TaskInputType]] = typing.Any, default: Optional[TaskInputType] = None, validate: Optional[Callable[[ATask, TaskInputType], Validation]] = None, description: Optional[str] = None, prompt: Optional[str] = None, transient: Optional[bool] = False, hidden: Optional[bool] = False)

Represents a input of a task. Input values can be set from a higher hierarchy.

This class may be subclassed to implement custom validation logic. All class variables in Task objects that are instances of this class are automatically converted to member variables by the framework. It is also possible to just annotate a variable (without assigning an actual value); in that case the input needs to be set from a higher hierarchy, i.e. it has no default value. Some subclasses for basic type checking are already implemented; see FileInput, StringInput, ListInput, …

default

default value of this input. Defaults to None which is interpreted as a missing value.

validate

an optional validation function that is applied to the input value once all task inputs have been set.

description

an optional description of the input.

transient

when True this input is ignored during caching.

Example

A simple example showing different types of inputs:

class MyTask(CachableTask):
    # must be set from outside, i.e. has no default value
    p1: Input

    # a input with a default value and a description
    p2 = Input(default="a string", description="this is a string input")

    # use a lambda to access other inputs or dependencies of a task. The passed task
    # is a self reference.
    p3 = StringInput(default=lambda task: str(task.p1.value))

    # inputs can provide custom validate logic via the constructor. The validate
    # function is called after all input values have been set.
    p4 = IntInput(
        default=0,
        validate=lambda task, value: Validation.of(value < 10, "value too large")
    )

    # inputs marked as transient will be ignored when the state of a task is
    # determined in the caching process.
    p5 = Input(default=0, transient=True)

property default: Optional[TaskInputType]

default value

task_hash() → bytes

default implementation of task hash

validate() → Validation

overwrite this method to implement custom validation logic.

class mosaic_orchestrator.task.InputDescription(path: str, input: Input, parent_task: Task)

holds a Input and the path of that object in the task tree

input: Input

parent_task: Task

path: str

exception mosaic_orchestrator.task.InputException

InputException is raised when a task input is not handled right, e.g. unexpected inputs in run()

class mosaic_orchestrator.task.Mutable

Used to mark a input of a subtask as calculated by default but still mutable from a higher hierarchy.

This means that the default value of this input is calculated at runtime (i.e. task execution). These default values need to be passed to the run method of the subtask. In general the behaviour is the same as Calculated inputs with the only difference that Mutable inputs can still be overridden from a higher hierarchy.

property value

class mosaic_orchestrator.task.Output(type: Optional[Type[TaskOutputType]] = typing.Any, default: Optional[TaskOutputType] = None, validate: Optional[Callable[[TaskOutputType], Validation]] = None, description: Optional[str] = None, prompt: Optional[str] = None, transient: Optional[bool] = False)

class mosaic_orchestrator.task.Task(inputs: Optional[Dict[str, object]] = None)

Abstract class that represents a unit of work within the Mosaic framework.

The task is the main abstraction of the Mosaic framework. It represents a (single-threaded) unit of work that can be reused in different places within an analog generator, much like a simple function in Python. To define a task this class must be subclassed and the run() method must be implemented. All work a task does is expected to take place within this method. Tasks are intended to be used hierarchically, i.e. a parent task can consist of multiple child tasks. This is done by declaring a (child) task as a class variable. The framework will automatically create instance variables of any child task. Therefore, child tasks should be accessed the same as other properties using the self reference. Inputs of child tasks can be set using the constructor.

Example

This simple example shows how to define and relate a parent task to a child task:

class Addition(Task):
    summand1 = Input()
    summand2 = Input()

    def run(self, *args, **kwargs):
        return self.summand1.value + self.summand2.value


class MyTask(Task):
    t1 = Addition(inputs={"summand1": 2, "summand2": 3})
    t2 = Addition(inputs={"summand1": 1, "summand2": 1})

    def run(self, *args, **kwargs):
        t1_result = self.t1.run() # will be 5
        t2_result = self.t2.run() # will be 2
        ...

copy(new_id: str) → ATask

Allows to copy a task dynamically during task execution.

The created instance is added as a child of this task using the passed id. All inputs are copied from their current state.

Parameters:

new_id – the id of the new task. Must be unique within this task.

Returns:

The new task instance.

property cwd: str

path to the root of the current working directory (cwd)

delete_run_dirs()

Deletes the run directory of this task.

dynamic_inputs() → Dict[str, Input]

Can be overridden to add dynamic Input e.g. from tools to the Task. Pdks and Tools are already available, other static defined Inputs not.

Returns:

A dict containing input names and Inputs to be added to the Task before the run() is called

dynamic_tasks() → Dict[str, Task]

Can be overridden to add dynamic Task e.g. from tools to the Task. Pdks and Tools are already available, other static defined Inputs not.

Returns:

A dict containing input names and Task to be added to the Task before the run() is called

get_fields_of_type(field_type: Type) → List

Return all fields of this task that are of type field_type or a subclass thereof.

get_outputs() → Dict[str, Output]

Get the outputs of the run() method. For this to work, the type annotation has to be set correctly.

property log: Logger

Obj:

Logger: a task specific logger instance. Logs are saved in the run directory of

each task as well as printed to stdout.

property mosaic

Get the mosaic instance

Returns:

mosaic instance

Return type:

TaskHierarchyBuilder

property path: Path

Obj:

Path: the unique path of this specific task instance starting from the root task.

abstract run(*args, **kwargs) → Union[Any, WorkProduct]

This method needs to be implemented when a new task is defined.

The current working directory of this task depends on its position in the task hierarchy, i.e. its path. The directory is created automatically as soon as this method is called. Custom runtime arguments are allowed.

validate_pdk(pdk: PDK) → Validation

Can be overridden to write custom validation logic for the PDK.

This method is called during the checking-phase, i.e. before the task tree is executed. Its intended use is to search the PDK for a specific functionality instead of a specific view. The default implementation returns ValidationSuccess.

Parameters:

pdk – The PDK object to validate.

Returns:

A Validation object indicating the result of the validation. ValidationSuccess in the

default implementation.

mosaic_orchestrator.task.TaskInputType

Type definition of Input

alias of TypeVar(‘TaskInputType’)

class mosaic_orchestrator.task.TaskNode(value: Task, children: List[Node], inputs: dict, injected_inputs: List[Tuple[Path, Optional[Task]]], parent: Optional[Task] = None)

hepler struct to holding a Task and additional inputs for building the task tree

injected_inputs: List[Tuple[Path, Optional[Task]]]

list of paths with tasks of the origin of the default inputs (in the hierarchy). The origin can be None, which means it was injected from toplevel.

inputs: dict

parent: Optional[Task] = None

property task: Task

returning task of the node

value: Task

mosaic_orchestrator.task.TaskOutputType

Type definition of Output

alias of TypeVar(‘TaskOutputType’)

class mosaic_orchestrator.task.TaskState(value: Optional[TaskStateType] = None)

Generic class that encapsulates the state of a task.

This class can be used similarly to a Input. It is used for caching and is intended to be used only from within a task. Hence, it can neither be modified from a parent task nor from the top level.

value: Optional[TaskStateType] = None

the value that is encapsulated in this instance.

mosaic_orchestrator.task.get_task_inputs(task: Task) → Dict[str, Input]

extracts all task input of a given Task, returning a dict of name and task input

mosaic_orchestrator.task.is_calculated(obj) → bool

Helper function do check if input is calculated

mosaic_orchestrator.task.is_calculated_or_mutable_or_const(obj) → bool

Helper function do check if input is calculated, mutable or constant

mosaic_orchestrator.task.is_const(obj) → bool

Helper function do check if input is constant

mosaic_orchestrator.task.is_mutable(obj) → bool

Helper function do check if input is mutable. A mutable input is a input that must be set when run() is being called but can be overwritten from above.