Skip to content

render_value

Attributes

logger = structlog.getLogger() module-attribute

Classes

RenderValueDetails

Bases: BaseOperationDetails

A model that contains information needed to describe an 'extract_metadata' operation.

Source code in kiara/operations/included_core_operations/render_value.py 
28
29
30
31
32
class RenderValueDetails(BaseOperationDetails):
    """A model that contains information needed to describe an 'extract_metadata' operation."""

    source_data_type: str = Field(description="The data type that will be rendered.")
    target_data_type: str = Field(description="The rendered data type.")

Attributes

source_data_type: str = Field(description='The data type that will be rendered.') class-attribute
target_data_type: str = Field(description='The rendered data type.') class-attribute

RenderValueOperationType

Bases: OperationType[RenderValueDetails]

An operation that renders a value.

A 'render_value' operation typically is named follwing the pattern:

 'render.<source_type>.as.<target_type>'
It has 2 inputs
  • 'value': the value to render
  • 'reneer_config' a target type-specific configuration dict
And one output
Source code in kiara/operations/included_core_operations/render_value.py 
 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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
class RenderValueOperationType(OperationType[RenderValueDetails]):
    """An operation that renders a value.

    A 'render_value' operation typically is named follwing the pattern:

    ```
     'render.<source_type>.as.<target_type>'
    ```

    It has 2 inputs:
      - 'value': the value to render
      - 'reneer_config' a target type-specific configuration dict

    And one output:
      - `render_value_result`: using internal type [RenderValueResultDataType][kiara.data_types.included_core_types.internal.render_value.RenderValueResultDataType]

    """

    _operation_type_name = "render_value"

    def _calculate_op_id(cls, source_type: str, target_type: str):

        if source_type == "any":
            operation_id = f"render.as.{target_type}"
        else:
            operation_id = f"render.{source_type}.as.{target_type}"

        return operation_id

    def retrieve_included_operation_configs(
        self,
    ) -> Iterable[Union[Mapping, OperationConfig]]:

        result = {}
        for name, module_cls in self._kiara.module_type_classes.items():

            if not issubclass(module_cls, RenderValueModule):
                continue

            for (
                source_type,
                target_type,
            ) in module_cls.retrieve_supported_render_combinations():
                if source_type not in self._kiara.data_type_names:
                    log_message("ignore.operation_config", operation_type="render_value", module_type=module_cls._module_type_name, source_type=source_type, target_type=target_type, reason=f"Source type '{source_type}' not registered.")  # type: ignore
                    continue
                if target_type not in self._kiara.data_type_names:
                    log_message(
                        "ignore.operation_config",
                        operation_type="render_value",
                        module_type=module_cls._module_type_name,
                        source_type=source_type,  # type: ignore
                        target_type=target_type,
                        reason=f"Target type '{target_type}' not registered.",
                    )
                    continue
                func_name = f"render__{source_type}__as__{target_type}"
                attr = getattr(module_cls, func_name)
                doc = DocumentationMetadataModel.from_function(attr)
                mc = {"source_type": source_type, "target_type": target_type}
                oc = ManifestOperationConfig(
                    module_type=name, module_config=mc, doc=doc
                )
                op_id = self._calculate_op_id(
                    source_type=source_type, target_type=target_type
                )
                result[op_id] = oc

        for data_type_name, data_type_class in self._kiara.data_type_classes.items():
            for attr in data_type_class.__dict__.keys():
                if not attr.startswith("render_as__"):
                    continue

                target_type = attr[11:]
                if target_type not in self._kiara.data_type_names:
                    log_message(
                        "operation_config.ignore",
                        operation_type="render_value",
                        source_type=data_type_name,
                        target_type=target_type,
                        reason=f"Target type '{target_type}' not registered.",
                    )  # type: ignore

                # TODO: inspect signature?
                doc = DocumentationMetadataModel.from_string(
                    f"Render a '{data_type_name}' value as a {target_type}."
                )
                mc = {
                    "source_type": data_type_name,
                    "target_type": target_type,
                }
                oc = ManifestOperationConfig(
                    module_type="render.value", module_config=mc, doc=doc
                )

                result[f"_type_{data_type_name}_{target_type}"] = oc

        return result.values()

    def check_matching_operation(
        self, module: "KiaraModule"
    ) -> Union[RenderValueDetails, None]:

        if len(module.inputs_schema) != 2:
            return None

        if len(module.outputs_schema) != 1:
            return None

        if "value" not in module.inputs_schema.keys():
            return None

        if (
            "render_config" not in module.inputs_schema.keys()
            or module.inputs_schema["render_config"].type != "dict"
        ):
            return None

        if (
            "render_value_result" not in module.outputs_schema.keys()
            or module.outputs_schema["render_value_result"].type
            != "render_value_result"
        ):
            return None

        source_type = module.inputs_schema["value"].type
        target_type = module.get_config_value("target_type")

        if source_type == "any":
            op_id = f"render.as.{target_type}"
        else:
            op_id = f"render.{source_type}.as.{target_type}"

        details = RenderValueDetails.create_operation_details(
            module_inputs_schema=module.inputs_schema,
            module_outputs_schema=module.outputs_schema,
            operation_id=op_id,
            source_data_type=source_type,
            target_data_type=target_type,
            is_internal_operation=True,
        )

        return details

    def get_render_operations_for_source_type(
        self, source_type: str
    ) -> Mapping[str, Operation]:
        """Return all render operations for the specified data type.

        Arguments:
            source_type: the data type to render

        Returns:
            a mapping with the target type as key, and the operation as value
        """

        if source_type not in self._kiara.data_type_names:
            source_type = "any"

        lineage = self._kiara.type_registry.get_type_lineage(data_type_name=source_type)

        result: Dict[str, Operation] = {}

        for data_type in lineage:

            for op_id, op in self.operations.items():
                op_details = self.retrieve_operation_details(op)
                match = op_details.source_data_type == data_type
                if not match:
                    continue
                target_type = op_details.target_data_type
                if target_type in result.keys():
                    continue
                result[target_type] = op

        return result

    def get_render_operations_for_target_type(
        self, target_type: str
    ) -> Mapping[str, Operation]:
        """Return all render operations that renders to the specified data type.

        Arguments:
            target_type: the result data type

        Returns:
            a mapping with the source type as key, and the operation as value
        """

        # TODO: consider type lineages

        if target_type not in self._kiara.data_type_names:
            raise Exception(f"Invalid target data type: {target_type}")

        result: Dict[str, Operation] = {}

        for op_id, op in self.operations.items():
            op_details = self.retrieve_operation_details(op)
            match = op_details.target_data_type == target_type
            if not match:
                continue
            source_type = op_details.source_data_type
            if source_type in result.keys():
                continue

            result[source_type] = op

        return result

    def get_render_operation(
        self, source_type: str, target_type: str
    ) -> Union[Operation, None]:

        all_ops = self.get_render_operations_for_source_type(source_type=source_type)
        return all_ops.get(target_type, None)

Functions

retrieve_included_operation_configs() -> Iterable[Union[Mapping, OperationConfig]]
Source code in kiara/operations/included_core_operations/render_value.py 
 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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def retrieve_included_operation_configs(
    self,
) -> Iterable[Union[Mapping, OperationConfig]]:

    result = {}
    for name, module_cls in self._kiara.module_type_classes.items():

        if not issubclass(module_cls, RenderValueModule):
            continue

        for (
            source_type,
            target_type,
        ) in module_cls.retrieve_supported_render_combinations():
            if source_type not in self._kiara.data_type_names:
                log_message("ignore.operation_config", operation_type="render_value", module_type=module_cls._module_type_name, source_type=source_type, target_type=target_type, reason=f"Source type '{source_type}' not registered.")  # type: ignore
                continue
            if target_type not in self._kiara.data_type_names:
                log_message(
                    "ignore.operation_config",
                    operation_type="render_value",
                    module_type=module_cls._module_type_name,
                    source_type=source_type,  # type: ignore
                    target_type=target_type,
                    reason=f"Target type '{target_type}' not registered.",
                )
                continue
            func_name = f"render__{source_type}__as__{target_type}"
            attr = getattr(module_cls, func_name)
            doc = DocumentationMetadataModel.from_function(attr)
            mc = {"source_type": source_type, "target_type": target_type}
            oc = ManifestOperationConfig(
                module_type=name, module_config=mc, doc=doc
            )
            op_id = self._calculate_op_id(
                source_type=source_type, target_type=target_type
            )
            result[op_id] = oc

    for data_type_name, data_type_class in self._kiara.data_type_classes.items():
        for attr in data_type_class.__dict__.keys():
            if not attr.startswith("render_as__"):
                continue

            target_type = attr[11:]
            if target_type not in self._kiara.data_type_names:
                log_message(
                    "operation_config.ignore",
                    operation_type="render_value",
                    source_type=data_type_name,
                    target_type=target_type,
                    reason=f"Target type '{target_type}' not registered.",
                )  # type: ignore

            # TODO: inspect signature?
            doc = DocumentationMetadataModel.from_string(
                f"Render a '{data_type_name}' value as a {target_type}."
            )
            mc = {
                "source_type": data_type_name,
                "target_type": target_type,
            }
            oc = ManifestOperationConfig(
                module_type="render.value", module_config=mc, doc=doc
            )

            result[f"_type_{data_type_name}_{target_type}"] = oc

    return result.values()
check_matching_operation(module: KiaraModule) -> Union[RenderValueDetails, None]
Source code in kiara/operations/included_core_operations/render_value.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def check_matching_operation(
    self, module: "KiaraModule"
) -> Union[RenderValueDetails, None]:

    if len(module.inputs_schema) != 2:
        return None

    if len(module.outputs_schema) != 1:
        return None

    if "value" not in module.inputs_schema.keys():
        return None

    if (
        "render_config" not in module.inputs_schema.keys()
        or module.inputs_schema["render_config"].type != "dict"
    ):
        return None

    if (
        "render_value_result" not in module.outputs_schema.keys()
        or module.outputs_schema["render_value_result"].type
        != "render_value_result"
    ):
        return None

    source_type = module.inputs_schema["value"].type
    target_type = module.get_config_value("target_type")

    if source_type == "any":
        op_id = f"render.as.{target_type}"
    else:
        op_id = f"render.{source_type}.as.{target_type}"

    details = RenderValueDetails.create_operation_details(
        module_inputs_schema=module.inputs_schema,
        module_outputs_schema=module.outputs_schema,
        operation_id=op_id,
        source_data_type=source_type,
        target_data_type=target_type,
        is_internal_operation=True,
    )

    return details
get_render_operations_for_source_type(source_type: str) -> Mapping[str, Operation]

Return all render operations for the specified data type.

Parameters:

Name Type Description Default
source_type str

the data type to render

 required

Returns:

Type Description
Mapping[str, Operation]

a mapping with the target type as key, and the operation as value
Source code in kiara/operations/included_core_operations/render_value.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def get_render_operations_for_source_type(
    self, source_type: str
) -> Mapping[str, Operation]:
    """Return all render operations for the specified data type.

    Arguments:
        source_type: the data type to render

    Returns:
        a mapping with the target type as key, and the operation as value
    """

    if source_type not in self._kiara.data_type_names:
        source_type = "any"

    lineage = self._kiara.type_registry.get_type_lineage(data_type_name=source_type)

    result: Dict[str, Operation] = {}

    for data_type in lineage:

        for op_id, op in self.operations.items():
            op_details = self.retrieve_operation_details(op)
            match = op_details.source_data_type == data_type
            if not match:
                continue
            target_type = op_details.target_data_type
            if target_type in result.keys():
                continue
            result[target_type] = op

    return result
get_render_operations_for_target_type(target_type: str) -> Mapping[str, Operation]

Return all render operations that renders to the specified data type.

Parameters:

Name Type Description Default
target_type str

the result data type

 required

Returns:

Type Description
Mapping[str, Operation]

a mapping with the source type as key, and the operation as value
Source code in kiara/operations/included_core_operations/render_value.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def get_render_operations_for_target_type(
    self, target_type: str
) -> Mapping[str, Operation]:
    """Return all render operations that renders to the specified data type.

    Arguments:
        target_type: the result data type

    Returns:
        a mapping with the source type as key, and the operation as value
    """

    # TODO: consider type lineages

    if target_type not in self._kiara.data_type_names:
        raise Exception(f"Invalid target data type: {target_type}")

    result: Dict[str, Operation] = {}

    for op_id, op in self.operations.items():
        op_details = self.retrieve_operation_details(op)
        match = op_details.target_data_type == target_type
        if not match:
            continue
        source_type = op_details.source_data_type
        if source_type in result.keys():
            continue

        result[source_type] = op

    return result
get_render_operation(source_type: str, target_type: str) -> Union[Operation, None]
Source code in kiara/operations/included_core_operations/render_value.py 
244
245
246
247
248
249
def get_render_operation(
    self, source_type: str, target_type: str
) -> Union[Operation, None]:

    all_ops = self.get_render_operations_for_source_type(source_type=source_type)
    return all_ops.get(target_type, None)

Functions