modules.operators

Classes

class modules.operators.ArithmeticOperator

Calculate two numbers or two number arrays by four rules of arithmetic.

Inherits from:

PipeModule

Methods:

__init__(mname: str | None = 'ArithmeticOperator', auto_run: bool = True, operator: Literal[(+, -, *, /)] = '+', data1: PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries | None = None, data2: PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries | None = None) → None

Initialize the ArithmeticOperator object.

Parameters

data1PortTypeHint.Number | PortTypeHint.NumberArray | None, default: None

The first number or number array for the calculation.

data2PortTypeHint.Number | PortTypeHint.NumberArray | None, default: None

The second number or number array for the calculation.

update_ui_schema(reset: bool = False) → dict[str, UIAttributeSchema]
execute() → PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries | None

Attributes:

InputNumbers1: PortReference[PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries]
InputNumbers2: PortReference[PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries]
OutputNumbers: PortReference[PortTypeHint.Number | PortTypeHint.NumberArray | PortTypeHint.TableSeries]
class modules.operators.ModifyTableColumns

Modify the values of columns of table.

Inherits from:

PipeModule

Methods:

__init__(mname: str | None = None, auto_run: bool = True, table: PortTypeHint.TableData | PortTypeHint.MaterialTable | None = None, modified_columns: dict[str, list[Any]] | None = None, available_columns: list[str] | None = None) → None

Initialize the ModifyTableColumns object.

Parameters

tablePortTypeHint.TableData | PortTypeHint.MaterialTable | None, default: None

The table to modify. modified_columns : dict[str, list[Any]] | None, default: None The names or titles of the columns to modify and their corresponding values. Key is the name or title of the column, value is the list of values to set for that column. If the length of the values is less than the length of the table rows, the values will be padded with the last value. If the length of the values is greater than the length of the table rows, the values will be truncated.

available_columnslist[str] | None, default: None

The names the columns that are available to modify. If it’s not provided, all the columns of the table will be available to modify. This is used for UI display. Ports

InputTablePortTypeHint.TableData | PortTypeHint.MaterialTable

The table to modify.

OutputTablePortTypeHint.TableData | PortTypeHint.MaterialTable

The modified table.

update_ui_schema(reset: bool = False) → dict[str, UIAttributeSchema]
execute() → PortTypeHint.TableData | PortTypeHint.MaterialTable | None

Attributes:

InputTable: PortReference[PortTypeHint.TableData | PortTypeHint.MaterialTable]
OutputTable: PortReference[PortTypeHint.TableData | PortTypeHint.MaterialTable]
class modules.operators.ModifyTablesColumns

Modify the values of columns of table collection.

Inherits from:

PipeModule

Methods:

__init__(mname: str | None = None, auto_run: bool = True, tables: PortTypeHint.TableCollection | None = None, column_names: dict[str, list[str]] | None = None, column_values: dict[str, list[list[Any]]] | None = None) → None

Initialize the ModifyTablesColumns object.

Parameters

tablesPortTypeHint.TableCollection | None, default: None

The tables to modify. column_names : dict[str, list[str]] | None, default: None The names or titles of the columns to modify. Key is the name or title of the table, value is the names or titles of the columns of each table to modify. column_values: dict[str, list[list[Any]]] | None, default: None The values to modify the columns. Key is the name or title of the table, value is the values to modify the columns of each table. If the length of the values is less than the length of the columns to modify, the values will be padded with the last value. If the length of the values is greater than the length of the columns to modify, the values will be truncated.

set_cal_params(reset: bool = False) → Dict[str, RangeModel]
execute() → PortTypeHint.TableCollection | None

Properties:

InputTables
OutputTables
class modules.operators.AddTableColumns

Add columns to tables using templates, static values, or custom functions.

This module supports three ways to define column values:

  1. Templates: Python expressions using {column_name} syntax

Examples

Numeric calculation
>>> module = AddTableColumns(
...     column_names=[{"name": "total", "title": "Total"}],
...     column_templates={"total": "{Amount} + {Tax}"}
... )

String concatenation
>>> module = AddTableColumns(
...     column_names=[{"name": "full_name", "title": "Full Name"}],
...     column_templates={"full_name": "{first_name} + ' ' + {last_name}"}
... )

Sequential numbering with prefix
>>> module = AddTableColumns(
...     column_names=[{"name": "joint_number", "title": "Joint Number"}],
...     column_templates={"joint_number": "'L' + str({_row_number+1})"}
... )
Result: "L1", "L2", "L3", ...

Multiple columns with mixed approaches
>>> module = AddTableColumns(
...     column_names=[
...         {"name": "total", "title": "Total"},
...         {"name": "status", "title": "Status"}
...     ],
...     column_templates={"total": "{Amount} * 1.1"},
...     column_values={"status": "Active"}
... )

Custom function
>>> module = AddTableColumns(
...     column_names=[{"name": "custom", "title": "Custom"}],
...     column_templates={"custom": "function:MyCalculator"},
...     local_functions_path="./my_functions.py"
... )
Inherits from:

PipeModule

Methods:

__init__(mname: str = 'AddTableColumns', auto_run: bool = True, table: PortTypeHint.TableData | None = None, column_names: list[FieldMetadata | dict[str, str]] | None = None, column_values: dict[str, Any] | None = None, column_templates: dict[str, str] | None = None, local_functions_path: str | Path | None = None, local_functions: dict[str, Any] | None = None, debug_mode: bool = False) → None

Initialize the AddTableColumns object.

Parameters

tablePortTypeHint.TableData | None, default: None

The table to add columns to. column_names : list[FieldMetadata | dict[str, str]] | None, default: None The metadata of the columns to add. column_values : dict[str, Any] | None, default: None The values of the columns. Key is column name, value can be:

  • A single value (applied to all rows)

  • A list/array (one value per row)

  • None (uses NaN)

Examples

debug_mode : bool, default: False
    If True, shows detailed error messages for custom function loading.

Notes

column_templates : dict[str, str] | None, default: None
    Template expressions for calculating column values. Key is column name, value is template.
    Templates use Python expression syntax:
    - Column references: {column_name}
    - Numeric operations: {col_A} + {col_B} * 2
    - String concatenation: {first_name} + ' ' + {last_name}
    - String literals: 'prefix-' + {column}
    - Special variables: {_row_number} (zero-based row index)
    - Type conversion: str({_row_number+1})
    - Custom functions: function:FunctionName
execute() → PortTypeHint.TableData | None

Execute the module to add columns to the table.

Properties:

column_names

Attributes:

InputTable: PortReference[PortTypeHint.TableData]
OutputTable: PortReference[PortTypeHint.TableData]
class modules.operators.TableCalculator

Create a new table with calculated/aggregated values from input table.

This module supports two main modes:

  1. Group-by aggregation: Group data and calculate aggregates per group

  2. Full-table aggregation: Calculate aggregates across the entire table

Supports three ways to define column values:

  1. Templates with aggregations: Excel-like expressions with {sum(column)} syntax

Examples

Group by category and calculate totals
>>> module = TableCalculator(
...     group_by=["Category"],
...     column_names=[
...         {"name": "total_amount", "title": "Total Amount"},
...         {"name": "avg_price", "title": "Average Price"}
...     ],
...     column_templates={
...         "total_amount": "{sum(Amount)}",
...         "avg_price": "{mean(Price)}"
...     }
... )

Multiple group columns with complex calculations
>>> module = TableCalculator(
...     group_by=["Category", "Region"],
...     column_names=[
...         {"name": "total", "title": "Total"},
...         {"name": "count", "title": "Count"},
...         {"name": "average", "title": "Average"}
...     ],
...     column_templates={
...         "total": "{sum(Amount)}",
...         "count": "{count(*)}",
...         "average": "{sum(Amount)} / {count(*)}"
...     }
... )

Full-table aggregation (no grouping)
>>> module = TableCalculator(
...     column_names=[{"name": "grand_total", "title": "Grand Total"}],
...     column_templates={"grand_total": "{sum(Amount)} + {sum(Tax)}"}
... )

Include source columns in output
>>> module = TableCalculator(
...     group_by=["Category"],
...     column_names=[{"name": "total", "title": "Total"}],
...     column_templates={"total": "{sum(Amount)}"},
...     include_source_columns=True
... )

Custom function for complex aggregation
>>> module = TableCalculator(
...     group_by=["Category"],
...     column_names=[{"name": "custom_metric", "title": "Custom Metric"}],
...     column_templates={"custom_metric": "function:MyAggregator"},
...     local_functions_path="./my_functions.py"
... )
Inherits from:

PipeModule

Methods:

__init__(mname: str = 'TableCalculator', auto_run: bool = True, table: PortTypeHint.TableData | None = None, column_names: list[FieldMetadata | dict[str, str]] | None = None, group_by: list[str] | None = None, column_templates: dict[str, str] | None = None, include_source_columns: bool = False, table_metadata: dict[str, str] | None = None, local_functions_path: str | Path | None = None, local_functions: dict[str, Any] | None = None, debug_mode: bool = False) → None

Initialize the TableCalculator object.

Parameters

tablePortTypeHint.TableData | None, default: None

The table to calculate values from. column_names : list[FieldMetadata | dict[str, str]] | None, default: None The metadata of the columns to create in the output table. Each dict should have at least “name” and optionally “title”, “description”.

group_bylist[str] | None, default: None

List of column names (or titles) to group by. When specified:

  • The table is grouped by these columns

  • Aggregation functions work within each group

  • Group columns are automatically included in output

  • Each unique combination of group values produces one output row

If None, calculations apply to the entire table (single output row). column_templates : dict[str, str] | None, default: None Template expressions for calculating column values. Key is column name, value is template. Templates support:

  • Aggregation functions: {sum(Amount)}, {mean(Price)}, {count(*)}

  • Row-level operations (only when group_by is None): {Amount} + {Tax}

  • Arithmetic: {sum(col_A)} + {sum(col_B)} * 2

  • Custom functions: function:FunctionName

Examples

debug_mode : bool, default: False
    If True, shows detailed error messages for custom function loading and evaluation.

Ports
InputTable: PortTypeHint.TableData
    The table to calculate values from.

OutputTable: PortTypeHint.TableData
    The new table with calculated/aggregated values.

Notes

(typically 'first' value from each group).

table_metadata : dict[str, str] | None, default: None
    The metadata of the output table.
    Each dict should have at least "name" and optionally "title", "unit", "description".

local_functions_path : str |  Path  | None, default: None
    Path to Python file containing custom functions decorated with @local_function.
    'local_functions_path' of pipeline has priority over the module.
    Not needed if ``local_functions`` is provided.

local_functions : dict[str, Any] | None, default: None
    Dictionary mapping function names to inline callables.
    These functions should accept (module, table, **kwargs) and return calculated values.
execute() → PortTypeHint.TableData | None

Execute the module to create a new calculated table.

Properties:

column_names

Attributes:

InputTable: PortReference[PortTypeHint.TableData]
OutputTable: PortReference[PortTypeHint.TableData]
class modules.operators.RenameTableColumns

Rename table columns and field titles while updating fields metadata.

Inherits from:

PipeModule

Methods:

__init__(mname: str = 'RenameTableColumns', auto_run: bool = True, table: PortTypeHint.TableData | None = None, column_name_map: dict[str, str] | None = None, rename_type: Literal[name, title] = 'name') → None

Initialize the RenameTableColumns object.

Parameters

tablePortTypeHint.TableData | None, default: None

The table to rename columns. column_name_map : dict[str, str] | None, default: None The mapping for renaming. If rename_type=”name”: {old_column_name: new_column_name} If rename_type=”title”: {old_field_title: new_field_title} rename_type : Literal[“name”, “title”], default: “name” “name”: rename column names, “title”: rename field titles

update_ui_schema(reset: bool = False) → dict[str, UIAttributeSchema]
execute() → PortTypeHint.TableData | None

Attributes:

InputTable: PortReference[PortTypeHint.TableData]
OutputTable: PortReference[PortTypeHint.TableData]
class modules.operators.TablesNamesMapper

Convert the names of tables or fields by the mapping file

Inherits from:

PipeModule

Methods:

__init__(mname: str = 'TablesNamesMapper', auto_run: bool = True, tables: PortTypeHint.TableCollection | PortTypeHint.TableCollectionDict | None = None, mapping_file: str | Path | dict | None = None) → None

Initialize TablesNamesMapper object.

Parameters

tablesPortTypeHint.TableCollection | None, default: None

The tables input.

mapping_filestr | Path | dict | None, default: None

The name of the .csv mapping file.

update_ui_schema(reset: bool = False) → dict[str, UIAttributeSchema]
execute() → PortTypeHint.TableCollection | PortTypeHint.TableCollectionDict | None

Attributes:

InputTables: PortReference[PortTypeHint.TableCollection | PortTypeHint.TableCollectionDict]
OutputTables: PortReference[PortTypeHint.TableCollection | PortTypeHint.TableCollectionDict]