module_types
deserialize.file¶
| type_name | deserialize.file | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class DeserializeFileModule(DeserializeValueModule): """Deserialize data to a 'file' value instance.""" _module_type_name = "deserialize.file" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": KiaraFile} @classmethod def retrieve_serialized_value_type(cls) -> str: return "file" @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "copy" def to__python_object(self, data: SerializedData, **config: Any): keys = list(data.get_keys()) keys.remove("__file_metadata__") assert len(keys) == 1 file_metadata_chunks = data.get_serialized_data("__file_metadata__") assert file_metadata_chunks.get_number_of_chunks() == 1 file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False)) assert len(file_metadata_json) == 1 file_metadata = orjson.loads(file_metadata_json[0]) chunks = data.get_serialized_data(keys[0]) assert chunks.get_number_of_chunks() == 1 files = list(chunks.get_chunks(as_files=True, symlink_ok=True)) assert len(files) == 1 file: str = files[0] # type: ignore _file_name = file_metadata.pop("file_name") _file_metadata = file_metadata.pop("metadata") _file_metadata_schemas = file_metadata.pop("metadata_schemas") fm = KiaraFile.load_file( source=file, file_name=_file_name, ) fm.metadata = _file_metadata fm.metadata_schemas = _file_metadata_schemas return fm | The source code of the process method of the module. |
deserialize.file_bundle¶
| type_name | deserialize.file_bundle | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class DeserializeFileBundleModule(DeserializeValueModule): """Deserialize data to a 'file' value instance.""" _module_type_name = "deserialize.file_bundle" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": KiaraFileBundle} @classmethod def retrieve_serialized_value_type(cls) -> str: return "file_bundle" @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "copy" def to__python_object(self, data: SerializedData, **config: Any): keys = list(data.get_keys()) keys.remove("__file_metadata__") file_metadata_chunks = data.get_serialized_data("__file_metadata__") assert file_metadata_chunks.get_number_of_chunks() == 1 file_metadata_json = list(file_metadata_chunks.get_chunks(as_files=False)) assert len(file_metadata_json) == 1 metadata = orjson.loads(file_metadata_json[0]) file_metadata = metadata["included_files"] bundle_name = metadata["bundle_name"] # bundle_import_time = metadata["import_time"] sum_size = metadata["size"] number_of_files = metadata["number_of_files"] included_files = {} for rel_path in keys: chunks = data.get_serialized_data(rel_path) assert chunks.get_number_of_chunks() == 1 files = list(chunks.get_chunks(as_files=True, symlink_ok=True)) assert len(files) == 1 file: str = files[0] # type: ignore file_name = file_metadata[rel_path]["file_name"] # import_time = file_metadata[rel_path]["import_time"] fm = KiaraFile.load_file(source=file, file_name=file_name) included_files[rel_path] = fm fb_metadata = metadata.pop("metadata") fb_metadata_schemas = metadata.pop("metadata_schemas") fb = KiaraFileBundle( included_files=included_files, bundle_name=bundle_name, # import_time=bundle_import_time, number_of_files=number_of_files, size=sum_size, metadata=fb_metadata, metadata_schemas=fb_metadata_schemas, ) return fb | The source code of the process method of the module. |
deserialize.from_json¶
| type_name | deserialize.from_json | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class DeserializeFromJsonModule(KiaraModule): _module_type_name: str = "deserialize.from_json" _config_cls = DeserializeJsonConfig def _retrieve_module_characteristics(self) -> ModuleCharacteristics: return DEFAULT_IDEMPOTENT_INTERNAL_MODULE_CHARACTERISTICS def create_inputs_schema( self, ) -> ValueMapSchema: return { "value": { "type": "any", "doc": "The value object to deserialize the data for.", } } def create_outputs_schema( self, ) -> ValueMapSchema: return { "python_object": { "type": "python_object", "doc": "The deserialized python object.", } } def process(self, inputs: ValueMap, outputs: ValueMap): value: Value = inputs.get_value_obj("value") serialized: SerializedData = value.serialized_data chunks = serialized.get_serialized_data(self.get_config_value("result_path")) assert chunks.get_number_of_chunks() == 1 _data = list(chunks.get_chunks(as_files=False)) assert len(_data) == 1 _chunk: bytes = _data[0] # type: ignore deserialized = orjson.loads(_chunk) outputs.set_value("python_object", deserialized) | The source code of the process method of the module. |
export.file¶
| type_name | export.file | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class ExportFileModule(DataExportModule): """Export files.""" _module_type_name = "export.file" def export__file__as__file(self, value: KiaraFile, base_path: str, name: str): target_path = os.path.join(base_path, value.file_name) shutil.copy2(value.path, target_path) return {"files": target_path} | The source code of the process method of the module. |
file_bundle.pick.file¶
| type_name | file_bundle.pick.file | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class PickFileFromFileBundleModule(KiaraModule): """Pick a single file from a file_bundle value.""" _module_type_name = "file_bundle.pick.file" def create_inputs_schema( self, ) -> ValueMapSchema: return { "file_bundle": {"type": "file_bundle", "doc": "The file bundle."}, "path": {"type": "string", "doc": "The relative path of the file to pick."}, } def create_outputs_schema( self, ) -> ValueMapSchema: return {"file": {"type": "file", "doc": "The file."}} def process(self, inputs: ValueMap, outputs: ValueMap): file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle") path: str = inputs.get_value_data("path") if path not in file_bundle.included_files.keys(): raise KiaraProcessingException( f"Can't pick file '{path}' from file bundle: file not available." ) file: KiaraFile = file_bundle.included_files[path] outputs.set_value("file", file) | The source code of the process method of the module. |
file_bundle.pick.sub_folder¶
| type_name | file_bundle.pick.sub_folder | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class PickSubBundle(KiaraModule): """Pick a sub-folder from a file_bundle, resulting in a new file_bundle.""" _module_type_name = "file_bundle.pick.sub_folder" def create_inputs_schema( self, ) -> ValueMapSchema: return { "file_bundle": {"type": "file_bundle", "doc": "The file bundle."}, "sub_path": { "type": "string", "doc": "The relative path of the sub-folder to pick.", }, } def create_outputs_schema(self) -> ValueMapSchema: return { "file_bundle": { "type": "file_bundle", "doc": "The picked (sub-)file_bundle.", } } def process(self, inputs: ValueMap, outputs: ValueMap): file_bundle: KiaraFileBundle = inputs.get_value_data("file_bundle") sub_path: str = inputs.get_value_data("sub_path") result = {} for path, file in file_bundle.included_files.items(): if path.startswith(sub_path): result[path] = file if not result: raise KiaraProcessingException( f"Can't pick sub-folder '{sub_path}' from file bundle: no matches." ) new_file_bundle: KiaraFileBundle = KiaraFileBundle.create_from_file_models( result, bundle_name=f"{file_bundle.bundle_name}_{sub_path}" ) outputs.set_value("file_bundle", new_file_bundle) | The source code of the process method of the module. |
import.local.file¶
| type_name | import.local.file | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class ImportLocalFileModule(KiaraModule): """Import a file from the local filesystem.""" _module_type_name = "import.local.file" def create_inputs_schema( self, ) -> ValueMapSchema: return {"path": {"type": "string", "doc": "The local path to the file."}} def create_outputs_schema( self, ) -> ValueMapSchema: return {"file": {"type": "file", "doc": "The loaded files."}} def _retrieve_module_characteristics(self) -> ModuleCharacteristics: return ModuleCharacteristics(is_idempotent=False) def process(self, inputs: ValueMap, outputs: ValueMap): path = inputs.get_value_data("path") file = KiaraFile.load_file(source=path) outputs.set_value("file", file) | The source code of the process method of the module. |
import.local.file_bundle¶
| type_name | import.local.file_bundle | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class ImportLocalFileBundleModule(KiaraModule): """Import a folder (file_bundle) from the local filesystem.""" _module_type_name = "import.local.file_bundle" _config_cls = ImportFileBundleConfig def create_inputs_schema( self, ) -> ValueMapSchema: return { "path": {"type": "string", "doc": "The local path of the folder to import."} } def create_outputs_schema( self, ) -> ValueMapSchema: return { "file_bundle": {"type": "file_bundle", "doc": "The imported file bundle."} } def _retrieve_module_characteristics(self) -> ModuleCharacteristics: return DEFAULT_NO_IDEMPOTENT_MODULE_CHARACTERISTICS def process(self, inputs: ValueMap, outputs: ValueMap): path = inputs.get_value_data("path") include = self.get_config_value("include_file_types") exclude = self.get_config_value("exclude_file_types") config = FolderImportConfig(include_files=include, exclude_files=exclude) file_bundle = KiaraFileBundle.import_folder(source=path, import_config=config) outputs.set_value("file_bundle", file_bundle) | The source code of the process method of the module. |
load.bytes¶
| type_name | load.bytes | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class LoadBytesModule(DeserializeValueModule): _module_type_name = "load.bytes" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": bytes} @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "raw" @classmethod def retrieve_serialized_value_type(cls) -> str: return "bytes" def to__python_object(self, data: SerializedData, **config: Any) -> bytes: chunks = data.get_serialized_data("bytes") assert chunks.get_number_of_chunks() == 1 _chunks = list(chunks.get_chunks(as_files=False)) assert len(_chunks) == 1 _chunk: bytes = _chunks[0] # type: ignore return _chunk | The source code of the process method of the module. |
load.internal_model¶
| type_name | load.internal_model | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class LoadInternalModel(DeserializeValueModule): _module_type_name = "load.internal_model" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": KiaraModel} @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "json" @classmethod def retrieve_serialized_value_type(cls) -> str: return "internal_model" def to__python_object(self, data: SerializedData, **config: Any) -> KiaraModel: chunks = data.get_serialized_data("data") assert chunks.get_number_of_chunks() == 1 _chunks = list(chunks.get_chunks(as_files=False)) assert len(_chunks) == 1 bytes_string: bytes = _chunks[0] # type: ignore model_data = orjson.loads(bytes_string) model_id: str = data.data_type_config["kiara_model_id"] model_registry = ModelRegistry.instance() m_cls = model_registry.get_model_cls(kiara_model_id=model_id) obj = m_cls(**model_data) return obj | The source code of the process method of the module. |
load.string¶
| type_name | load.string | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class LoadStringModule(DeserializeValueModule): _module_type_name = "load.string" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": str} @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "raw" @classmethod def retrieve_serialized_value_type(cls) -> str: return "string" def to__python_object(self, data: SerializedData, **config: Any) -> str: chunks = data.get_serialized_data("string") assert chunks.get_number_of_chunks() == 1 _chunks = list(chunks.get_chunks(as_files=False)) assert len(_chunks) == 1 bytes_string: bytes = _chunks[0] # type: ignore return bytes_string.decode("utf-8") | The source code of the process method of the module. |
pipeline¶
| type_name | pipeline | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class PipelineModule(KiaraModule): """A utility module to run multiple connected inner-modules and present it as its own entity.""" _config_cls = PipelineConfig _module_type_name = "pipeline" def __init__( self, module_config: Union[None, KIARA_CONFIG, Mapping[str, Any]] = None, ): self._job_registry: Union[JobRegistry, None] = None super().__init__(module_config=module_config) @classmethod def is_pipeline(cls) -> bool: return True def _set_job_registry(self, job_registry: "JobRegistry"): self._job_registry = job_registry @property def operation(self) -> "Operation": if self._operation is not None: return self._operation from kiara.models.module.operation import Operation self._operation = Operation.create_from_module(self, doc=self.config.doc) return self._operation def create_inputs_schema( self, ) -> ValueMapSchema: pipeline_structure: PipelineStructure = self.config.structure inputs_schema = pipeline_structure.pipeline_inputs_schema return inputs_schema def create_outputs_schema( self, ) -> ValueMapSchema: pipeline_structure: PipelineStructure = self.config.structure return pipeline_structure.pipeline_outputs_schema def process(self, inputs: ValueMap, outputs: ValueMapWritable, job_log: JobLog): pipeline_structure: PipelineStructure = self.config.structure pipeline = Pipeline(structure=pipeline_structure, kiara=outputs._kiara) assert self._job_registry is not None controller = SinglePipelineBatchController( pipeline=pipeline, job_registry=self._job_registry ) job_log.add_log("setting pipeline inputs") pipeline.set_pipeline_inputs(inputs=inputs) job_log.add_log("starting pipeline processing") def event_callback(msg: str): job_log.add_log(msg) step_details = controller.process_pipeline(event_callback=event_callback) errors: Dict[str, Union[Exception, uuid.UUID]] = {} for step_id, details in step_details.items(): if isinstance(details, Exception): errors[step_id] = details else: job = self._job_registry.get_job(details) if job.error: if job._exception: errors[step_id] = job._exception else: errors[step_id] = Exception(job.error) if errors: msg = "Error processing pipeline:" for f, e in errors.items(): msg = f"{msg}\n - {f}: {e}" raise KiaraProcessingException(f"Errors while processing pipeline: {msg}") # TODO: resolve values first? outputs.set_values(**pipeline.get_current_pipeline_outputs()) | The source code of the process method of the module. |
pretty_print.any.value¶
| type_name | pretty_print.any.value | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class PrettyPrintAnyValueModule(PrettyPrintModule): _module_type_name = "pretty_print.any.value" # def pretty_print__any__as__string(self, value: Value, render_config: Dict[str, Any]): # # data = value.data # if isinstance(data, KiaraModel): # return data.json(option=orjson.OPT_INDENT_2) # else: # return str(data) | The source code of the process method of the module. |
pretty_print.value¶
| type_name | pretty_print.value | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class ValueTypePrettyPrintModule(KiaraModule): _module_type_name = "pretty_print.value" _config_cls = PrettyPrintConfig def _retrieve_module_characteristics(self) -> ModuleCharacteristics: return DEFAULT_IDEMPOTENT_INTERNAL_MODULE_CHARACTERISTICS def create_inputs_schema( self, ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]: source_type = self.get_config_value("source_type") assert source_type not in ["target", "base_name"] schema = { "value": { "type": source_type, "doc": "The value to render.", "optional": True, }, "render_config": { "type": "any", "doc": "Value type dependent render configuration.", "optional": True, }, } return schema def create_outputs_schema( self, ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]: return { "rendered_value": { "type": self.get_config_value("target_type"), "doc": "The rendered value.", } } def process(self, inputs: ValueMap, outputs: ValueMap): # source_type = self.get_config_value("source_type") target_type = self.get_config_value("target_type") source_value = inputs.get_value_obj("value") render_config = inputs.get_value_obj("render_config") if not source_value.is_set: outputs.set_value("rendered_value", "-- none/not set --") return try: data_type_cls = source_value.data_type_info.data_type_class.get_class() data_type = data_type_cls(**source_value.value_schema.type_config) except Exception as e: source_data_type = source_value.data_type_name log_message("data_type.unknown", data_type=source_data_type, error=e) from kiara.data_types.included_core_types import AnyType data_type = AnyType() func_name = f"pretty_print_as__{target_type}" func = getattr(data_type, func_name) render_config_dict = render_config.data if render_config_dict is None: render_config_dict = {} result = func(value=source_value, render_config=render_config_dict) # TODO: check we have the correct type? outputs.set_value("rendered_value", result) | The source code of the process method of the module. |
render.value¶
| type_name | render.value | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class ValueTypeRenderModule(KiaraModule): """A module that uses render methods attached to DataType classes.""" _module_type_name = "render.value" _config_cls = RenderValueModuleConfig def _retrieve_module_characteristics(self) -> ModuleCharacteristics: return DEFAULT_IDEMPOTENT_INTERNAL_MODULE_CHARACTERISTICS def create_inputs_schema( self, ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]: source_type = self.get_config_value("source_type") assert source_type not in ["target", "base_name"] schema = { "value": { "type": source_type, "doc": "The value to render.", "optional": False, }, "render_config": { "type": "dict", "doc": "Instructions/config on how (or what) to render the provided value.", "default": {}, }, } return schema def create_outputs_schema( self, ) -> Mapping[str, Union[ValueSchema, Mapping[str, Any]]]: outputs = { "render_value_result": { "type": "render_value_result", "doc": "The rendered value, incl. some metadata.", }, } return outputs def process(self, inputs: ValueMap, outputs: ValueMap): source_value = inputs.get_value_obj("value") if not source_value.is_set: raise KiaraProcessingException( f"Can't render value '{source_value.value_id}': value not set." ) # source_type = self.get_config_value("source_type") target_type = self.get_config_value("target_type") render_scene: KiaraDict = inputs.get_value_data("render_config") try: data_type_cls = source_value.data_type_info.data_type_class.get_class() data_type = data_type_cls(**source_value.value_schema.type_config) except Exception as e: source_data_type = source_value.data_type_name log_message("data_type.unknown", data_type=source_data_type, error=e) from kiara.data_types.included_core_types import AnyType data_type = AnyType() func_name = f"render_as__{target_type}" func = getattr(data_type, func_name) if render_scene: rc = render_scene.dict_data else: rc = {} result = func( value=source_value, render_config=rc, manifest=self.manifest, ) if isinstance(result, RenderValueResult): render_scene_result = result else: render_scene_result = RenderValueResult( value_id=source_value.value_id, render_config=rc, render_manifest=self.manifest.manifest_hash, rendered=result, related_scenes={}, manifest_lookup={self.manifest.manifest_hash: self.manifest}, ) outputs.set_value("render_value_result", render_scene_result) | The source code of the process method of the module. |
unpickle.value¶
| type_name | unpickle.value | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | class UnpickleModule(DeserializeValueModule): _module_type_name = "unpickle.value" @classmethod def retrieve_supported_target_profiles(cls) -> Mapping[str, Type]: return {"python_object": object} @classmethod def retrieve_supported_serialization_profile(cls) -> str: return "pickle" @classmethod def retrieve_serialized_value_type(cls) -> str: return "any" def to__python_object(self, data: SerializedData, **config: Any): try: import pickle5 as pickle except Exception: import pickle # type: ignore assert "python_object" in data.get_keys() python_object_data = data.get_serialized_data("python_object") assert python_object_data.get_number_of_chunks() == 1 _bytes = list(python_object_data.get_chunks(as_files=False))[0] data = pickle.loads(_bytes) return data | The source code of the process method of the module. |
value.extract_metadata¶
| type_name | value.extract_metadata | The registered name for this item type. | |||||||||||||||||||||||||||
| documentation |
|
Documentation for the item. | |||||||||||||||||||||||||||
| authors |
|
Information about authorship for the item. | |||||||||||||||||||||||||||
| context |
|
Generic properties of this item (description, tags, labels, references, ...). | |||||||||||||||||||||||||||
| python_class |
|
The python class that implements this module type. | |||||||||||||||||||||||||||
| module_src | 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) | The source code of the process method of the module. |