mosaic

This module contains the mosaic-orchestrator framework core functionality

class mosaic_orchestrator.mosaic.Mosaic(root: Type[T], caching, plugin_enabled)[source]

This is the main entry point of the framework. Use create to start building an executable task tree.

Build an instance of a task from the builder object.

>>> instance = Mosaic.create(root=ExampleTask).build()
>>> instance.run()
0.0

Raises:: MosaicException – If an errors occures during check()
Returns:: An instance of the task
Return type:: T

check() → CheckResult[source]

Perform checks without running a task. Performed checks are: - Missing parameters - Validation of input parameters - Checking of tools availability - Checking of PDK completness (are all PDKItems available?)

Returns:: Container of check results
Return type:: CheckResult

static create(root: Type[T], caching: bool = True, plugin_enabled: bool = True) → TaskHierarchyBuilder[T][source]

creates a builder for a task hierarchy given a root task.

Parameters:

root – the type of the desired root task. Notice that actual instantiation is done by the framework.
caching – when False all caching is disabled for hierarchies build with this builder.
plugin_enabled – Enables die plugin discovery for PDKs and tools.

Returns:

a TaskHierarchyBuilder for the given root type.

Example

This is a simple example that shows how to build a simple mosaic_orchestrator task tree:

builder = Mosaic.create(root=MyTask)                        .with_pdk(MyPdkImpl())                        .register(AtoolInterface, MyToolImpl())                        .register(AnotherDependency, AnotherDependency())                        .with_inputs({
        "x1": 5,
        "x2": 3,
        "t1.p5": "REF",
    })

if builder.check().success:
    instance = builder.build()
    instance.run()

default_cache_directory = '.mosaic_orchestrator'

default_work_directory = 'mosaic_orchestrator.work'

export_inputs_to_dict(skip_defaults: bool = True, skip_hidden: bool = True) → dict[source]

Run check() and write all inputs into a dictionary that can be used to save it in any kind of format of your choice (e.g. JSON/YMAL). You can later load them again via import_inputs_to_dict().

>>> builder = Mosaic.create(root=ExampleTask).with_inputs({"width": 20})
>>> builder.export_inputs_to_dict(skip_defaults=False, skip_hidden=False)
{'ExampleTask.width': 20}

Parameters:

skip_defaults (bool, optional) – Skip inputs that have value==defaults. Defaults to True.
skip_hidden (bool, optional) – Skip hidden inputs. Defaults to True.

Returns:

Dictionary with input path as keys and input values as values

Return type:

dict

static get_installed_pdks() → List[PDK][source]: lists all installed PDKs

static get_installed_tools() → List[Tool][source]: lists all installed Tools

import_inputs_from_dict(input_dict: dict[str, Any]) → TaskHierarchyBuilder[T][source]

Import parameters from a dictionary that has been created via export_inputs_from_dict. It can be used to save and restore inputs of root tasks.

>>> builder = Mosaic.create(root=ExampleTask).with_inputs({"width": 20.0})
>>> exported_dict = builder.export_inputs_to_dict(skip_defaults=False, skip_hidden=False)
>>> builder = Mosaic.create(root=ExampleTask).import_inputs_from_dict(exported_dict)
>>> result = builder.check()
>>> result.inputs[0].input.value
20.0

Parameters:

input_dict (dict[str, Any]) – Dictionary in the form of path->obj representation

Raises:

MosaicException – If a path is not found
MosaicException – Dataclass value cannot be initialized

Returns:

The builder instance

Return type:

TaskHierarchyBuilder[T]

log_formatter = <logging.Formatter object>: the formatter used for logging. It is used for stdout as well as log files. Can be set globally

log_level = 20: the log level of all loggers. Can be set globally

register(key: Union[Type, str], value: any, overwrite: bool = False) → TaskHierarchyBuilder[T][source]

registers a dependency based on its type or its path in the hierarchy.

This is the intended way to make tool implementations available for tasks. Subsequent calls with the same key overwrite the initial value.

Parameters:

key – value will be injected into all tasks that have members annotated with this type. When key is str it is expected to be a path to a task input; in this case value is set to this input.
value – this object will be injected when key matches
overwrite – Overwrite tool/object if it has already been registered

Returns:

a self reference to allow method chaining.

register_execution_tracker(tracker: ExecutionTracker) → TaskHierarchyBuilder[T][source]

Register an execution tracker

Parameters:: tracker (ExecutionTracker) – Execution tracker to add

with_cache_directory(path: Path) → TaskHierarchyBuilder[T][source]

sets the cache directory for this hierarchy.

Parameters:: path – a path to the target folder. The folder will be created if it does not exist.
Returns:: a self reference to allow method chaining.

with_cfg(config: Union[Config, str]) → TaskHierarchyBuilder[T][source]

loads a Config either from an object or from a python script.

This method can be called repeatedly to merge multiple configurations.

Parameters:: config – a Config object that will be applied to this hierarchy. Alternatively, pass a path to a python script containing a MosaicConfig class; This class will be automatically isntantiated and loaded.
Returns:: a self reference to allow method chaining.

with_inputs(inputs: Dict[str, object]) → TaskHierarchyBuilder[T][source]

Add inputs to the task.

>>> Mosaic.create(root=ExampleTask).with_inputs({"width": 10.0}).build().run()
10.0

Parameters:: inputs – Inputs to the task
Returns:: The builder instance
Return type:: TaskHierarchyBuilder[T]

with_pdk(pdk: PDK) → TaskHierarchyBuilder[T][source]

sets the pdk for this hierarchy. Subsequent calls overwrite the inital value.

Parameters:: pdk – the instantiated pdk object.
Returns:: a self reference to allow method chaining.

with_run_directory(path: Path) → TaskHierarchyBuilder[T][source]

sets the run directory for this hierarchy.

Parameters:: path – a path to the target folder. The folder will be created if it does not exist.
Returns:: a self reference to allow method chaining.