Skip to content

kiara_modules.core.dev

Modules that are useful for kiara as well as pipeline-development, as well as testing.

DummyModule

Module that simulates processing, but uses hard-coded outputs as a result.

create_input_schema(self)

The input schema for the dummy module is created at object creation time from the input_schemas config parameter.

Source code in core/dev.py 
def create_input_schema(self) -> typing.Mapping[str, ValueSchema]:
    """The input schema for the ``dummy`` module is created at object creation time from the ``input_schemas`` config parameter."""

    result = {}
    for k, v in self.config.get("input_schema").items():  # type: ignore
        schema = ValueSchema(**v)
        schema.validate_types(self._kiara)
        result[k] = schema
    return result

create_output_schema(self)

The output schema for the dummy module is created at object creation time from the output_schemas config parameter.

Source code in core/dev.py 
def create_output_schema(self) -> typing.Mapping[str, ValueSchema]:
    """The output schema for the ``dummy`` module is created at object creation time from the ``output_schemas`` config parameter."""

    result = {}
    for k, v in self.config.get("output_schema").items():  # type: ignore
        schema = ValueSchema(**v)
        schema.validate_types(self._kiara)
        result[k] = schema
    return result

process(self, inputs, outputs)

Returns the hardcoded output values that are set in the outputs config field.

Optionally, this module can simulate processing by waiting a configured amount of time (seconds -- specified in the delay config parameter).

Source code in core/dev.py 
def process(self, inputs: ValueSet, outputs: ValueSet) -> None:
    """Returns the hardcoded output values that are set in the ``outputs`` config field.

    Optionally, this module can simulate processing by waiting a configured amount of time (seconds -- specified in the ``delay`` config parameter).
    """

    time.sleep(self.config.get("delay"))  # type: ignore

    output_values: typing.Mapping = self.config.get("outputs")  # type: ignore

    value_dict = {}
    for output_name in self.output_names:
        if output_name not in output_values.keys():
            raise NotImplementedError()
            # v = self.output_schemas[output_name].type_obj.fake_value()
            # value_dict[output_name] = v
        else:
            value_dict[output_name] = output_values[output_name]
    outputs.set_values(**value_dict)

DummyProcessingModuleConfig pydantic-model

Configuration for the 'dummy' processing module.

delay: float pydantic-field

The delay in seconds from processing start to when the (dummy) outputs are returned.

input_schema: Dict[str, Dict] pydantic-field required

The input schema for this module.

output_schema: Dict[str, Dict] pydantic-field required

The output schema for this module.

outputs: Dict[str, Any] pydantic-field

The (dummy) output for this module.