pipeline.module

PipeModule: abstract base class for all pipeline modules.

Classes

class pipeline.module.PipeModule

This is an abstract class used as a blueprint for all pipe network modules.

Attributes

mnamestr: The name of the module.
auto_runbool: If it’s True, the module will be executed automatically.
executedbool: True when the most recent ``execute()`` returned a non-None value.
executed_countint: Total number of times the module completed with a non-None return value. Incremented each time the module completes without error; never reset between runs. Useful for detecting modules that execute multiple times (e.g. inside a ``ForEachController`` loop).
executed_timedatetime | None: The time of the most recent successful execution. values_range : dict[str, RangeModel | NestedRangeModel] The range of the values of the module attributes. Will be deprecated in the future. ui_schema : dict[str, UIAttributeSchema] | None The UI schema of the module.
dynamic_ports_in_typePortType | list[PortType] | None: The type of the dynamic input ports.
dynamic_ports_out_typePortType | list[PortType] | None: The type of the dynamic output ports.
titlestr | None: The title of the module.
descriptionstr | None: The description of the module.

Inherits from:: ABC

Methods:

__init__(mname: str | None = None, auto_run: bool = True) → None

Initialize the PipelineModule object.

Parameters

mnamestr, default: BaseModule: The name of the module.
auto_runbool, default: False: If it’s True, the module will be executed automatically.

execute() → Any: Define the executing method here

set_cal_params(reset: bool = False) → dict[str, RangeModel | NestedRangeModel] | None

Define the method to set the value of the calculation parameters according to input data.

Parameters

resetbool, default: False: If it’s True, the value of calculation parameters of the module will be reset according to the input data.

Returns

Any

dict[str, RangeModel | NestedRangeModel]
    Values range of each parameter of the module will be returned.

update_ui_schema(reset: bool = False) → dict[str, UIAttributeSchema] | None

Set the UI schema of the module.

Parameters

resetbool, default: False: If it’s True, the value of module attributes will be reset according to the input data.

Returns

Any

dict[str, UIAttributeSchema] | None
    UI schema of the module will be returned.

get_connected_ports(port_name: str) → tuple[str, str] | list[tuple[str, str]] | None

Parameters

port_namestr: Static, dynamic, or reserved port name.

Returns

Any

tuple[str, str] |  list[tuple[str, str]]  | None
    Either a single upstream pair or a list of downstream pairs.

user_defined_ports_in_keys() → list[str]

user_defined_ports_out_keys() → list[str]

all_ports_in_keys() → list[str]

all_ports_out_keys() → list[str]

get_input_port(port_name: str) → Port | None

Returns

Any: Static, dynamic, and reserved input ports are resolved in the same order as :meth:all_ports_in_keys.

get_output_port(port_name: str) → Port | None

Returns

Any: Static, dynamic, and reserved output ports are resolved in the same order as :meth:all_ports_out_keys.

add_dynamic_ports_in(key: str, pdoc: str | None = None) → None

Add a dynamic input port with proper naming convention and property creation.

Parameters

keystr: The name of the port. Must start with ‘Input’ to follow naming convention. pdoc : str, optional The documentation/description of the port.

add_dynamic_ports_out(key: str, pdoc: str | None = None) → None

Add a dynamic output port with proper naming convention and property creation.

Parameters

keystr: The name of the port. Must start with ‘Output’ to follow naming convention. pdoc : str, optional The documentation/description of the port.

remove_dynamic_ports_in(pname: str) → None

Remove a dynamic input port and its associated property.

Parameters

pnamestr: The name of the port to remove.

remove_dynamic_ports_out(pname: str) → None

Remove a dynamic output port and its associated property.

Parameters

pnamestr: The name of the port to remove.

keys() → dict[str, Any]

values() → dict[str, Any]

items() → dict[str, Any]

get_cal_params() → dict[str, Any]

Get the calculation parameters of the module.

Examples

{"param_name": param_value}

dump_module_params() → ModuleData

Dump the module parameters to a ModuleData object (json).

Returns

Any

ModuleData
    The module data.

Notes

Pydantic BaseModel instances in calculation parameters are automatically
serialized with type information for later reconstruction. This enables
proper round-trip serialization/deserialization of complex parameter types
like TemplateVariableConfig, UIAttributeSchema subclasses, etc.

restore_module_params(module_data: ModuleData) → None

Restore the module params of the module from a ModuleData object (Pydantic Model).

Parameters

module_dataModuleData: The module data.

Notes

Pydantic BaseModel instances that were serialized with type information
are automatically reconstructed to their original types. This enables
proper round-trip serialization/deserialization of complex parameter types
like TemplateVariableConfig, UIAttributeSchema subclasses, etc.

get_workspace(module_workspace: str | Path | None = None) → Path

Get the real workspace path from the pipeline or module.

Parameters

module_workspacestr | Path | None: The workspace set by the parameter of the module.

Notes

- 'pipeline_workspace' has priority over the 'module_workspace'.
- If both 'module_workspace' and 'pipeline_workspace' are None, the current working directory will be used.

Get the token from the pipeline or input token.

With a pipeline, token, proj_id, and host start from gdim_token, gdim_proj_id, and gdim_host; unset slots are filled from matching elements of input_token. Without a pipeline, all three values come from input_token. proj_id_from_module is an optional, highest-precedence override for proj_id only; it does not affect how token or host are merged.

Parameters

proj_id_requiredbool, default: True: If True and proj_id is None after resolution (including override), returns None. If False, returns (token, proj_id, host)` even when proj_id is ``None`.
proj_id_from_modulestr | None, default: None: Module-supplied project id (e.g. self.proj_id). If not None, uses this value instead of gdim_proj_id (when a pipeline is present) and instead of input_token’s project id. proj_id_required is checked on the resolved proj_id.

get_gdim_template(token: str, tpl_id: str | None = None, proj_id: str | None = None, tree_tpl: bool = False, get_app_info: bool = False, host: str | None = None, gdim: bool = True) → GdimTemplate

Get the gdim template.

Parameters

tokenstr: The token of the user.
tpl_idstr | None, default: None: The id of the template. If not None, proj_id will be ignored.
proj_idstr | None, default: None: The id of the project.
tree_tplbool, default: False: Whether to get the tree template.
get_app_infobool, default: False: Whether to get the application information.
hoststr | None, default: None: The host of the platform.
gdimbool, default: True: Whether it’s gdim platform

get_llm_key(module_llm_key: str | None) → str | None

Get the api key of LLM model from the pipeline or module.

The api key of LLM model of the pipeline has priority over the module.

get_local_functions_path(module_local_functions_path: str | Path | None) → str | Path | None

Get the local functions path from the pipeline or module.

The local functions path of the pipeline has priority over the module.

get_tpl_id(module_tpl_id: str | None) → str | None

Get the template id from the pipeline or module.

The template id of the pipeline has priority over the module.

get_gdim_proj_list(token: str, tpl_id: str | None, host: str | None = None) → dict[str, GdimProjectInfo]: Get the project list.

Properties:

title

description

pipeline

dynamic_ports_in

dynamic_ports_out

reserved_ports_in

reserved_ports_out

InputAttributes: Get a PortReference for the InputAttributes port to enable linking with >> operator.

OutputAttributes: Get a PortReference for the OutputAttributes port to enable linking with >> operator.

ports_info

Get information of all ports

Returns

Any: Dictionary

Examples

::

       {"input_ports": {"port_name1": port_type1, "port_name2": port_type2},
       "output_ports": {"port_name": port_type}
       }