Skip to content

metadata

Classes

MetadataModuleConfig

Bases: KiaraModuleConfig

Source code in kiara/modules/included_core_modules/metadata.py 
20
21
22
23
class MetadataModuleConfig(KiaraModuleConfig):

    data_type: str = Field(description="The data type this module will be used for.")
    kiara_model_id: str = Field(description="The id of the kiara (metadata) model.")

Attributes

data_type: str = Field(description='The data type this module will be used for.') class-attribute
kiara_model_id: str = Field(description='The id of the kiara (metadata) model.') class-attribute

ExtractMetadataModule

Bases: KiaraModule

Base class to use when writing a module to extract metadata from a file.

It's possible to use any arbitrary kiara module for this purpose, but sub-classing this makes it easier.

Source code in kiara/modules/included_core_modules/metadata.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class ExtractMetadataModule(KiaraModule):
    """Base class to use when writing a module to extract metadata from a file.

    It's possible to use any arbitrary *kiara* module for this purpose, but sub-classing this makes it easier.
    """

    _config_cls = MetadataModuleConfig
    _module_type_name: str = "value.extract_metadata"

    def _retrieve_module_characteristics(self) -> ModuleCharacteristics:

        return ModuleCharacteristics(
            is_idempotent=True, is_internal=True, unique_result_values=True
        )

    def create_inputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        data_type_name = self.get_config_value("data_type")
        inputs = {
            "value": {
                "type": data_type_name,
                "doc": f"A value of type '{data_type_name}'",
                "optional": False,
            }
        }
        return inputs

    def create_outputs_schema(
        self,
    ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

        kiara_model_id: str = self.get_config_value("kiara_model_id")

        # TODO: check it's subclassing the right class

        outputs = {
            "value_metadata": {
                "type": "internal_model",
                "type_config": {"kiara_model_id": kiara_model_id},
                "doc": "The metadata for the provided value.",
            }
        }

        return outputs

    def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

        value = inputs.get_value_obj("value")

        kiara_model_id: str = self.get_config_value("kiara_model_id")

        model_registry = ModelRegistry.instance()
        metadata_model_cls: Type[ValueMetadata] = model_registry.get_model_cls(kiara_model_id=kiara_model_id, required_subclass=ValueMetadata)  # type: ignore

        metadata = metadata_model_cls.create_value_metadata(value=value)

        if not isinstance(metadata, metadata_model_cls):
            raise KiaraProcessingException(
                f"Invalid metadata model result, should be class '{metadata_model_cls.__name__}', but is: {metadata.__class__.__name__}. This is most likely a bug."
            )

        outputs.set_value("value_metadata", metadata)

Attributes

_config_cls = MetadataModuleConfig class-attribute

Functions

create_inputs_schema() -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]
Source code in kiara/modules/included_core_modules/metadata.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def create_inputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    data_type_name = self.get_config_value("data_type")
    inputs = {
        "value": {
            "type": data_type_name,
            "doc": f"A value of type '{data_type_name}'",
            "optional": False,
        }
    }
    return inputs
create_outputs_schema() -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]
Source code in kiara/modules/included_core_modules/metadata.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def create_outputs_schema(
    self,
) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]:

    kiara_model_id: str = self.get_config_value("kiara_model_id")

    # TODO: check it's subclassing the right class

    outputs = {
        "value_metadata": {
            "type": "internal_model",
            "type_config": {"kiara_model_id": kiara_model_id},
            "doc": "The metadata for the provided value.",
        }
    }

    return outputs
process(inputs: ValueMap, outputs: ValueMap) -> None
Source code in kiara/modules/included_core_modules/metadata.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def process(self, inputs: ValueMap, outputs: ValueMap) -> None:

    value = inputs.get_value_obj("value")

    kiara_model_id: str = self.get_config_value("kiara_model_id")

    model_registry = ModelRegistry.instance()
    metadata_model_cls: Type[ValueMetadata] = model_registry.get_model_cls(kiara_model_id=kiara_model_id, required_subclass=ValueMetadata)  # type: ignore

    metadata = metadata_model_cls.create_value_metadata(value=value)

    if not isinstance(metadata, metadata_model_cls):
        raise KiaraProcessingException(
            f"Invalid metadata model result, should be class '{metadata_model_cls.__name__}', but is: {metadata.__class__.__name__}. This is most likely a bug."
        )

    outputs.set_value("value_metadata", metadata)