omnipy.components.general.tasks

General tasks for splitting, importing, and converting datasets.

FUNCTION	DESCRIPTION
convert_dataset	Convert a dataset into another dataset class.
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.

convert_dataset

convert_dataset(dataset: Dataset, dataset_cls: type[_DatasetT], **kwargs: object) -> _DatasetT

Convert a dataset into another dataset class.

PARAMETER	DESCRIPTION
dataset	Source dataset to convert. TYPE: Dataset
dataset_cls	Target dataset class. TYPE: type[_DatasetT]
**kwargs	Extra keyword arguments forwarded 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 convert_dataset(dataset: Dataset, dataset_cls: type[_DatasetT], **kwargs: object) -> _DatasetT:
    """Convert a dataset into another dataset class.

    Args:
        dataset: Source dataset to convert.
        dataset_cls: Target dataset class.
        **kwargs: Extra keyword arguments forwarded to ``dataset_cls``.

    Returns:
        A dataset instance of type ``dataset_cls``.
    """
    return dataset_cls(dataset, **kwargs)  # 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