modules.widgets

Classes

class modules.widgets.ItemsSelector

Define a selector to select multi-items from a list.

Inherits from:: PipeModule

Methods:

__init__(mname: str | None = None, auto_run: bool = True, items_list: list[Any] | None = None, items_label: list[str] | None = None, single_selector: bool = True, auto_selected: list[Any] | None = None, bunddle_items: dict[str, list[Any]] | None = None, as_single_result: bool = False, single_unit_result: bool = False) → None

Initialize ItemsSelector object.

Parameters

items_listlist[Any]: The items to select from.
items_labellist[str]: The labels of the items which is used to show the users.
single_selectorbool: Whether to use a single selector or a multi-selector.
auto_selectedlist[Any]: The items to select automatically. These items usually are not visible to users. bunddle_items : dict[str, list[Any]] Bunddle serveral items to a single item. key is the name of the item in the drop list, value is the output items. Exmaple: { “item1”: [“item1_1”, “item1_2”, “item1_3”], “item2”: [“item2_1”, “item2_2”, “item2_3”] }
as_single_resultbool, default: False: If True, the output will be a SingleResult object. If False, the output will be a general array.
single_unit_resultbool, default: False: Only available when as_single_result is True. If True, all the seleted items will be a value of a unit reuslt in a list type whiche means the length of the single result will always be 1. If False, each selected item will be a value of a unit result which means the length of the single result will equal to the length of the selected items.

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

execute() → PortTypeHint.GeneralArray | PortTypeHint.SingleResult | None

Properties:

OutputItems

class modules.widgets.AttributeCaculator

This module is used to define a local function to calculate the values of attributes for another module.

Inherits from:: PipeModule

Methods:

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

Initialize AttributeCaculator object.

Parameters

local_functions_pathstr | None, default: None: The path to the local functions file. If it’s not provided, the module will try to use the local functions file path in the pipeline.
local_function_namestr | None, default: None: The name of the class that defines the function.
pipeline_local_functions_path_firstbool, default: True: If True, the ‘local_functions_path’ attribute of the pipeline will be used to find the local functions path first. If False, the module’s attributes ‘local_functions_path’ will be used to find the local functions path first.

set_cal_params(reset: bool = True) → None

execute() → PortTypeHint.Attributes | None

Properties:

OutputAttributesValue

class modules.widgets.PythonCoder

This module is used to run a local python script.

This is a flexible module that can accept any type of input through General ports
and output any type of data through General ports by running custom Python functions.

Supports two ways to define local functions:

#. **Inline functions** (recommended for same-script usage):
   Pass a callable directly to ``local_function_name``.
   In this mode, **NO ``@local_function`` decorator is needed** because the callable
   is not loaded by name from a file.

#. **File-based functions** (for separate files):
   Pass function name as string to ``local_function_name``.
   In this mode, the function in the file **must** be decorated with ``@local_function``
   so it can be discovered/validated by the loader.

Inherits from:: PipeModule

Methods:

Initialize PythonCoder object.

Parameters

local_functions_pathstr | Path | None, default: None

The path to the local functions file. ‘local_functions_path’ of pipeline has priority over the module. Not needed if local_function_name is a callable.

local_function_namestr | Callable | None, default: None

Either:

A string: name of the function/class in the local functions file
A callable: inline function that receives module and returns {port_name: data}

ui_schema_function_namestr | Callable | None, default: None

Either:

A string: name of the UI schema function in the local functions file
A callable: inline function that receives (module, reset) and returns {attr_name: UIAttributeSchema}

debug_modebool, default: False

If True, provides detailed error messages including full tracebacks when loading local functions fails. If False, provides concise error messages for production use.

Returns

Any

coder = PythonCoder(
       mname="TableProcessor",
       local_function_name=process_tables  # Pass function directly
   )

Examples

Using inline function:

.. code-block:: python

   def process_tables(module):
       tables = module.dynamic_ports_in["InputTables"].data

Notes

The returned result from the local function must be a dict: key is the name of dynamic_ports_out, value is the port data.

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

add_attributes() → None

Add attributes to the module.

Parameters

**kwargsdict: The attributes to add to the module.

execute() → Any

class modules.widgets.LoopDataCollector

This widget collects data from input ports across multiple loop iterations

and outputs the collected data as a list after the loop is completed.

This is useful in pipeline steps that involve loops, where you want to accumulate results from each iteration and then process them together.

Inherits from:: PipeModule

Methods:

__init__(mname: str | None = None, auto_run: bool = True, output_port_type: PortType = PortType.FilesPath, output_port_name: str = 'OutputCollectedData', reset_on_first_run: bool = True, max_iterations: int | None = None) → None

Initialize LoopDataCollector object.

Parameters

output_port_typePortType, default: PortType.FilesPath: The type of the output port. Can be FilesPath, GeneralArray, etc.
output_port_namestr, default: “OutputCollectedData”: The name of the output port.
reset_on_first_runbool, default: True: If True, the collected data will be reset on the first run of each loop cycle.
max_iterationsint | None, default: None: Maximum number of iterations to collect. If None, no limit is set.

add_input_port(port_name: str, pdoc: str | None = None) → None

Add a dynamic input port for data collection.

Parameters

port_namestr: The name of the input port to add.
pdocstr | None, default: None: The documentation for the port.

reset_collected_data() → None: Reset all collected data.

full_reset() → None: Completely reset the collector including the first run flag.

get_collected_data() → dict[str, list[Any]]

Get the currently collected data.

Returns

Any

dict[str, list[Any]]
    Dictionary mapping port names to lists of collected data.

set_cal_params(reset: bool = False) → dict[str, RangeModel] | None: Set calculation parameters. No user parameters needed for this widget.

execute() → Any

Execute the data collection.

This method collects data from all input ports and stores them in internal lists. The collected data is then output as a list through the output port.

finalize_collection() → Any

Finalize the data collection and return the final result.

This method can be called explicitly to signal the end of data collection.

Properties:

OutputCollectedData: Get the collected data from the default output port.