omnipy.components.general.tasks

General tasks for splitting, importing, and creating datasets and models.

FUNCTION	DESCRIPTION
concat_all_vals_in_datasets_as_args	Concatenate all value from positional datasets.
concat_all_vals_in_datasets_as_kwargs	Concatenate all values from keyword datasets.
create_dataset_from_args	Create a dataset from one or more positional payload objects.
create_dataset_from_kwargs	Create a dataset from named model or sub-dataset inputs.
create_model_from_args	Create a model from positional inputs.
create_model_from_kwargs	Create a model from named keyword inputs.
import_directory	Import files from a directory into a dataset keyed by filename stem.
split_dataset	Split a dataset into two datasets based on selected data-file names.
union_all_datasets_as_args	Union all positional datasets.
union_all_datasets_as_kwargs	Union all keyword datasets.
union_all_vals_in_datasets_as_args	Union all dataset values from positional datasets.
union_all_vals_in_datasets_as_kwargs	Union all dataset values from keyword datasets.

concat_all_vals_in_datasets_as_args

concat_all_vals_in_datasets_as_args(
    first_dataset: Dataset[Model[_SupportsIAddT]], *other_datasets: Dataset[Model[Any]]
) -> Model[_SupportsIAddT]

Concatenate all value from positional datasets.

Concatenation is based on a deep copy of the first value, with consecutive concatenations through the += operator.

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

@TaskTemplate()
def concat_all_vals_in_datasets_as_args(
    first_dataset: Dataset[Model[_SupportsIAddT]],
    *other_datasets: Dataset[Model[Any]],
) -> Model[_SupportsIAddT]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Concatenate all value from positional datasets.
    #
    # {{CONCAT_DESCRIPTION}}
    #
    """Concatenate all value from positional datasets.

    Concatenation is based on a deep copy of the first value, with
    consecutive concatenations through the `+=` operator.
    """

    return _concat_dataset_values(first_dataset, *other_datasets)

concat_all_vals_in_datasets_as_kwargs

concat_all_vals_in_datasets_as_kwargs(
    **datasets: Dataset[Model[_SupportsIAddT]],
) -> Model[_SupportsIAddT]

Concatenate all values from keyword datasets.

Concatenation is based on a deep copy of the first value, with consecutive concatenations through the += operator.

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

@TaskTemplate()
def concat_all_vals_in_datasets_as_kwargs(**datasets: Dataset[Model[_SupportsIAddT]],
                                          ) -> Model[_SupportsIAddT]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Concatenate all values from keyword datasets.
    #
    # {{CONCAT_DESCRIPTION}}
    #
    """Concatenate all values from keyword datasets.

    Concatenation is based on a deep copy of the first value, with
    consecutive concatenations through the `+=` operator.
    """

    first_dataset, other_datasets = _extract_first_and_other_datasets(datasets)
    return _concat_dataset_values(first_dataset, *other_datasets)

create_dataset_from_args

create_dataset_from_args(
    *args: object,
    dataset_cls: type[_DatasetT],
    key: str | None = None,
    keys: tuple[str, ...] | None = None,
) -> _DatasetT

Create a dataset from one or more positional payload objects.

With no positional inputs, an empty dataset is created. A single positional input is forwarded as a single argument. Multiple positional inputs are packed into the tuple normally represented by *args before Dataset construction, allowing for key-value pairs to be passed as positional arguments. Alternatively, the key or keys keyword parameters can be used to specify the keys for the datasets.

PARAMETER	DESCRIPTION
*args	Positional payload passed to the dataset constructor. TYPE: object DEFAULT: ()
dataset_cls	Dataset class to instantiate. TYPE: type[_DatasetT]
key	Optional single key (as string) for a single dataset entry. TYPE: str | None DEFAULT: None
keys	Optional keys (as tuple of strings) to use for the dataset entries. Must be of the same length as args. Only one of key or keys can be provided at a time. TYPE: tuple[str, ...] | None DEFAULT: None

RETURNS	DESCRIPTION
_DatasetT	A dataset instance of type dataset_cls.

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

@TaskTemplate()
def create_dataset_from_args(*args: object,
                             dataset_cls: type[_DatasetT],
                             key: str | None = None,
                             keys: tuple[str, ...] | None = None) -> _DatasetT:
    """Create a dataset from one or more positional payload objects.

    With no positional inputs, an empty dataset is created. A single
    positional input is forwarded as a single argument. Multiple
    positional inputs are packed into the tuple normally represented by
    ``*args`` before Dataset construction, allowing for key-value pairs to
    be passed as positional arguments. Alternatively, the ``key`` or
    ``keys`` keyword parameters can be used to specify the keys for the
    datasets.

    Args:
        *args: Positional payload passed to the dataset constructor.
        dataset_cls: Dataset class to instantiate.
        key: Optional single key (as string) for a single dataset entry.
        keys: Optional keys (as tuple of strings) to use for the dataset
            entries. Must be of the same length as ``args``. Only one of
            ``key`` or ``keys`` can be provided at a time.

    Returns:
        A dataset instance of type ``dataset_cls``.
    """
    assert (key is None or keys is None), \
        'Only one of `key` or `keys` can be provided at a time.'
    if key is not None:
        keys = (key,)

    if len(args) == 0:
        assert keys is None, ('No positional arguments were provided, but '
                              f'keys were provided: {keys}')
        return dataset_cls()

    if keys is not None:
        assert len(keys) == len(args), ('Number of keys must match number '
                                        'of positional arguments: '
                                        f'{len(keys)} != {len(args)}')
        return dataset_cls(dict(zip(keys, args)))
    else:
        if len(args) == 1:
            return dataset_cls(args[0])  # type: ignore[arg-type]
        return dataset_cls(args)  # type: ignore[arg-type]

create_dataset_from_kwargs

create_dataset_from_kwargs(*, dataset_cls: type[_DatasetT], **data: object) -> _DatasetT

Create a dataset from named model or sub-dataset inputs.

PARAMETER	DESCRIPTION
dataset_cls	Dataset class to instantiate. TYPE: type[_DatasetT]
**data	Named dataset entries forwarded as keyword arguments to dataset_cls. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_DatasetT	A dataset instance of type dataset_cls.

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

@TaskTemplate()
def create_dataset_from_kwargs(*, dataset_cls: type[_DatasetT], **data: object) -> _DatasetT:
    """Create a dataset from named model or sub-dataset inputs.

    Args:
        dataset_cls: Dataset class to instantiate.
        **data: Named dataset entries forwarded as keyword arguments to ``dataset_cls``.

    Returns:
        A dataset instance of type ``dataset_cls``.
    """
    return dataset_cls(**data)  # type: ignore[arg-type]

create_model_from_args

create_model_from_args(*args: object, model_cls: type[_ModelT]) -> _ModelT

Create a model from positional inputs.

A single positional input is forwarded unchanged. Multiple positional inputs are packed into the tuple normally represented by *args before model construction.

PARAMETER	DESCRIPTION
*args	Positional payload values to turn into the model root value. TYPE: object DEFAULT: ()
model_cls	Model class to instantiate. TYPE: type[_ModelT]

RETURNS	DESCRIPTION
_ModelT	A model instance of type model_cls.

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

@TaskTemplate()
def create_model_from_args(*args: object, model_cls: type[_ModelT]) -> _ModelT:
    """Create a model from positional inputs.

    A single positional input is forwarded unchanged. Multiple positional inputs are packed into the
    tuple normally represented by ``*args`` before model construction.

    Args:
        *args: Positional payload values to turn into the model root value.
        model_cls: Model class to instantiate.

    Returns:
        A model instance of type ``model_cls``.
    """
    if len(args) == 0:
        return model_cls()  # type: ignore[call-arg]
    if len(args) == 1:
        return model_cls(args[0])  # type: ignore[arg-type]
    return model_cls(args)  # type: ignore[arg-type]

create_model_from_kwargs

create_model_from_kwargs(*, model_cls: type[_ModelT], **data: object) -> _ModelT

Create a model from named keyword inputs.

PARAMETER	DESCRIPTION
model_cls	Model class to instantiate. TYPE: type[_ModelT]
**data	Named values forwarded as keyword arguments to model_cls. TYPE: object DEFAULT: {}

RETURNS	DESCRIPTION
_ModelT	A model instance of type model_cls.

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

@TaskTemplate()
def create_model_from_kwargs(*, model_cls: type[_ModelT], **data: object) -> _ModelT:
    """Create a model from named keyword inputs.

    Args:
        model_cls: Model class to instantiate.
        **data: Named values forwarded as keyword arguments to ``model_cls``.

    Returns:
        A model instance of type ``model_cls``.
    """
    return model_cls(**data)  # type: ignore[arg-type]

import_directory

import_directory(
    directory: str | Path,
    exclude_prefixes: tuple[str, ...] = (".", "_"),
    include_suffixes: tuple[str, ...] = (),
    dataset_cls: type[_DatasetT] = Dataset[Model[str]],
    open_func: Callable[[str], IOBase] = open,
) -> _DatasetT

Import files from a directory into a dataset keyed by filename stem.

PARAMETER	DESCRIPTION
directory	Directory to scan for files. TYPE: str | Path
exclude_prefixes	Filename prefixes to skip. TYPE: tuple[str, ...] DEFAULT: ('.', '_')
include_suffixes	Optional filename suffixes to include. TYPE: tuple[str, ...] DEFAULT: ()
dataset_cls	Dataset type to instantiate for the imported content. TYPE: type[_DatasetT] DEFAULT: Dataset[Model[str]]
open_func	Callable used to open each matching file. TYPE: Callable[[str], IOBase] DEFAULT: open

RETURNS	DESCRIPTION
_DatasetT	A dataset containing one item per imported file.

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

@TaskTemplate()
def import_directory(
        directory: str | Path,
        exclude_prefixes: tuple[str, ...] = ('.', '_'),
        include_suffixes: tuple[str, ...] = (),
        dataset_cls: type[_DatasetT] = Dataset[Model[str]],  # type: ignore
        open_func: Callable[[str], IOBase] = open) -> _DatasetT:
    """Import files from a directory into a dataset keyed by filename stem.

    Args:
        directory: Directory to scan for files.
        exclude_prefixes: Filename prefixes to skip.
        include_suffixes: Optional filename suffixes to include.
        dataset_cls: Dataset type to instantiate for the imported content.
        open_func: Callable used to open each matching file.

    Returns:
        A dataset containing one item per imported file.
    """
    dataset = dataset_cls()
    for import_filename in os.listdir(directory):
        if not exclude_prefixes or \
                not any(import_filename.startswith(prefix) for prefix in exclude_prefixes):
            if not include_suffixes or \
                    any(import_filename.endswith(suffix) for suffix in include_suffixes):
                with open_func(os.path.join(directory, import_filename)) as open_file:
                    dataset_name = '_'.join(import_filename.split('.')[:-1])
                    print(f"{import_filename} -> Dataset['{dataset_name}']")
                    dataset[dataset_name] = open_file.read()
    return dataset

split_dataset

split_dataset(
    dataset: Dataset[Model[object]], datafile_names_for_b: list[str]
) -> tuple[Dataset[Model[object]], Dataset[Model[object]]]

Split a dataset into two datasets based on selected data-file names.

PARAMETER	DESCRIPTION
dataset	Dataset to split. TYPE: Dataset[Model[object]]
datafile_names_for_b	Names that should be placed in the second output dataset. TYPE: list[str]

RETURNS	DESCRIPTION
tuple[Dataset[Model[object]], Dataset[Model[object]]]	A tuple containing the remaining items first and the selected items second.

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

@TaskTemplate()
def split_dataset(
        dataset: Dataset[Model[object]],
        datafile_names_for_b: list[str]) -> tuple[Dataset[Model[object]], Dataset[Model[object]]]:
    """Split a dataset into two datasets based on selected data-file names.

    Args:
        dataset: Dataset to split.
        datafile_names_for_b: Names that should be placed in the second output dataset.

    Returns:
        A tuple containing the remaining items first and the selected items second.
    """
    _type = dataset.get_type()
    datafile_names_for_a = set(dataset.keys()) - set(datafile_names_for_b)
    dataset_a = Dataset[_type](  # type: ignore[valid-type]
        {
            name: dataset[name] for name in dataset.keys() if name in datafile_names_for_a
        })
    dataset_b = Dataset[_type](  # type: ignore[valid-type]
        {
            name: dataset[name] for name in dataset.keys() if name in datafile_names_for_b
        })
    return dataset_a, dataset_b

union_all_datasets_as_args

union_all_datasets_as_args(
    first_dataset: _DatasetT, *other_datasets: Dataset[Model[Any]]
) -> _DatasetT

Union all positional datasets.

Union is based on a deep copy of the first dataset, with consecutive unions through the |= operator.

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

@TaskTemplate()
def union_all_datasets_as_args(
    first_dataset: _DatasetT,
    *other_datasets: Dataset[Model[Any]],
) -> _DatasetT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Union all positional datasets.
    #
    # {{UNION_DESCRIPTION_DATASET}}
    #
    """Union all positional datasets.

    Union is based on a deep copy of the first dataset, with consecutive
    unions through the `|=` operator.
    """

    return _union_datasets(first_dataset, *other_datasets)

union_all_datasets_as_kwargs

union_all_datasets_as_kwargs(**datasets: _DatasetT) -> _DatasetT

Union all keyword datasets.

Union is based on a deep copy of the first dataset, with consecutive unions through the |= operator.

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

@TaskTemplate()
def union_all_datasets_as_kwargs(**datasets: _DatasetT) -> _DatasetT:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Union all keyword datasets.
    #
    # {{UNION_DESCRIPTION_DATASET}}
    """Union all keyword datasets.

    Union is based on a deep copy of the first dataset, with consecutive
    unions through the `|=` operator."""

    first_dataset, other_datasets = _extract_first_and_other_datasets(datasets)
    return _union_datasets(first_dataset, *other_datasets)

union_all_vals_in_datasets_as_args

union_all_vals_in_datasets_as_args(
    first_dataset: Dataset[Model[_SupportsIOrT]], *other_datasets: Dataset[Model[Any]]
) -> Model[_SupportsIOrT]

Union all dataset values from positional datasets.

Union is based on a deep copy of the first value, with consecutive unions through the |= operator.

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

@TaskTemplate()
def union_all_vals_in_datasets_as_args(
    first_dataset: Dataset[Model[_SupportsIOrT]],
    *other_datasets: Dataset[Model[Any]],
) -> Model[_SupportsIOrT]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Union all dataset values from positional datasets.
    #
    # {{UNION_DESCRIPTION_VALUE}}
    #
    """Union all dataset values from positional datasets.

    Union is based on a deep copy of the first value, with consecutive
    unions through the `|=` operator.
    """

    return _union_dataset_values(first_dataset, *other_datasets)

union_all_vals_in_datasets_as_kwargs

union_all_vals_in_datasets_as_kwargs(
    **datasets: Dataset[Model[_SupportsIOrT]],
) -> Model[_SupportsIOrT]

Union all dataset values from keyword datasets.

Union is based on a deep copy of the first value, with consecutive unions through the |= operator.

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

@TaskTemplate()
def union_all_vals_in_datasets_as_kwargs(**datasets: Dataset[Model[_SupportsIOrT]],
                                         ) -> Model[_SupportsIOrT]:
    # %% Original docstring (managed by expand_docstr_macros.py) %%
    # Union all dataset values from keyword datasets.
    #
    # {{UNION_DESCRIPTION_VALUE}}
    #
    """Union all dataset values from keyword datasets.

    Union is based on a deep copy of the first value, with consecutive
    unions through the `|=` operator.
    """

    first_dataset, other_datasets = _extract_first_and_other_datasets(datasets)
    return _union_dataset_values(first_dataset, *other_datasets)