omnipy.components.tables.tasks

Tasks for reshaping and relabeling tabular datasets.

FUNCTION	DESCRIPTION
create_row_index_from_column	Re-key a row-wise table by promoting one column to the row index.
remove_columns	Build a JSON list-of-dicts dataset using a per-item column-removal mapping.
rename_col_names	Rename selected column names in a row-wise table.
transpose_columns_with_data_files	Transpose table columns so data-file names become nested keys.

create_row_index_from_column

create_row_index_from_column(
    list_of_dicts: RowWiseTableWithColNamesModel, column_key: str
) -> ColumnWiseTableWithColNamesAndIndexModel

Re-key a row-wise table by promoting one column to the row index.

PARAMETER	DESCRIPTION
list_of_dicts	Table represented as a list of row mappings. TYPE: RowWiseTableWithColNamesModel
column_key	Name of the column whose values become the outer row keys. TYPE: str

RETURNS	DESCRIPTION
ColumnWiseTableWithColNamesAndIndexModel	An indexed table where each outer key is taken from column_key.

Source code in src/omnipy/components/tables/tasks.py

@TaskTemplate(iterate_over_data_files=True)
def create_row_index_from_column(list_of_dicts: RowWiseTableWithColNamesModel,
                                 column_key: str) -> ColumnWiseTableWithColNamesAndIndexModel:
    """Re-key a row-wise table by promoting one column to the row index.

    Args:
        list_of_dicts: Table represented as a list of row mappings.
        column_key: Name of the column whose values become the outer row keys.

    Returns:
        An indexed table where each outer key is taken from ``column_key``.
    """
    output_dict = {}
    input_table = cast(list[dict[str, JsonScalar]], list_of_dicts.to_data())
    for item in input_table:
        item_copy = copy(item)
        new_key = item[column_key]
        del item_copy[column_key]
        output_dict[new_key] = item_copy
    return ColumnWiseTableWithColNamesAndIndexModel(output_dict)

remove_columns

remove_columns(
    json_dataset: JsonListOfDictsDataset, column_keys_for_data_items: dict[str, list[str]]
) -> JsonListOfDictsDataset

Build a JSON list-of-dicts dataset using a per-item column-removal mapping.

PARAMETER	DESCRIPTION
json_dataset	Dataset containing row-wise records keyed by data item name. TYPE: JsonListOfDictsDataset
column_keys_for_data_items	Column names to target for each named data item. TYPE: dict[str, list[str]]

RETURNS	DESCRIPTION
JsonListOfDictsDataset	A JSON list-of-dicts dataset with the same item structure as the input.

Source code in src/omnipy/components/tables/tasks.py

@TaskTemplate()
def remove_columns(json_dataset: JsonListOfDictsDataset,
                   column_keys_for_data_items: dict[str, list[str]]) -> JsonListOfDictsDataset:
    """Build a JSON list-of-dicts dataset using a per-item column-removal mapping.

    Args:
        json_dataset: Dataset containing row-wise records keyed by data item name.
        column_keys_for_data_items: Column names to target for each named data item.

    Returns:
        A JSON list-of-dicts dataset with the same item structure as the input.
    """
    # TODO: implement general solution to make sure that one does not modify input data by
    #       automatically copying or otherwise. Perhaps setting immutable/frozen option in pydantic,
    #       if available?
    #
    dataset_copy = deepcopy(json_dataset)
    for data_item_key, column_keys in column_keys_for_data_items.items():
        for record in dataset_copy[data_item_key]:
            for column in column_keys:
                if column in record:
                    del record[column]
    return JsonListOfDictsDataset(json_dataset.to_data())

rename_col_names

rename_col_names(
    data_file: RowWiseTableFirstRowAsColNamesModel, prev2new_keymap: dict[str, str]
) -> RowWiseTableFirstRowAsColNamesModel

Rename selected column names in a row-wise table.

PARAMETER	DESCRIPTION
data_file	Table whose rows are keyed by column name. TYPE: RowWiseTableFirstRowAsColNamesModel
prev2new_keymap	Mapping from existing column names to replacement names. TYPE: dict[str, str]

RETURNS	DESCRIPTION
RowWiseTableFirstRowAsColNamesModel	A new table with renamed column keys.

Source code in src/omnipy/components/tables/tasks.py

@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TableWithColNamesDataset)
def rename_col_names(data_file: RowWiseTableFirstRowAsColNamesModel,
                     prev2new_keymap: dict[str, str]) -> RowWiseTableFirstRowAsColNamesModel:
    """Rename selected column names in a row-wise table.

    Args:
        data_file: Table whose rows are keyed by column name.
        prev2new_keymap: Mapping from existing column names to replacement names.

    Returns:
        A new table with renamed column keys.
    """
    return RowWiseTableFirstRowAsColNamesModel([{
        prev2new_keymap[key] if key in prev2new_keymap else key: val for key, val in row.items()
    } for row in data_file])

transpose_columns_with_data_files

transpose_columns_with_data_files(
    dataset: TableWithColNamesDataset, exclude_cols: tuple[str]
) -> None

Transpose table columns so data-file names become nested keys.

PARAMETER	DESCRIPTION
dataset	Dataset of row-wise tables with column names. TYPE: TableWithColNamesDataset
exclude_cols	Column names to keep shared across all transposed rows. TYPE: tuple[str]

RETURNS	DESCRIPTION
None	A JSON dataset whose top-level items are former column names.

Source code in src/omnipy/components/tables/tasks.py

@TaskTemplate()
def transpose_columns_with_data_files(dataset: TableWithColNamesDataset,
                                      exclude_cols: tuple[str]) -> None:
    """Transpose table columns so data-file names become nested keys.

    Args:
        dataset: Dataset of row-wise tables with column names.
        exclude_cols: Column names to keep shared across all transposed rows.

    Returns:
        A JSON dataset whose top-level items are former column names.

    """
    output_dataset = JsonListOfDictsDataset()

    max_len = max(len(data_file) for data_file in dataset.values())

    # TODO: Make Dataset behave like a defaultDict, possibly also with auto-expanding lists?
    for column_name in dataset.col_names:
        if column_name not in exclude_cols:
            output_dataset[column_name] = [{}] * max_len

    for data_file_name, data_file in dataset.items():
        for row_i, el in enumerate(data_file):
            for key, val in el.items():
                if key in exclude_cols:
                    for data_file in output_dataset.values():
                        data_file[row_i][key] = val
                else:
                    output_dataset[key][row_i][data_file_name] = val

    return output_dataset