Utils Module

The utils module provides configuration management and utility functions.

class xflow.utils.ConfigManager(initial_config: Dict[str, Any])

Bases: object

In-memory config manager.

Keeps an immutable “source of truth” (_original_config) and a mutable working copy (_config).

__init__(initial_config: Dict[str, Any])

keys()

Return config keys.

values()

Return config values.

items()

Return config items.

add_files(file_paths: str | PathLike[str] | List[str | PathLike[str]] | Tuple[str | PathLike[str], ...]) → Self

Add files that are part of this configuration.

Parameters:

file_paths – Single file path or iterable of file paths

get() → Dict[str, Any]

Return a fully independent snapshot of the working config.

reset() → None

Revert working config back to original.

update(updates: Dict[str, Any]) → Self

Recursively update in config, Nested dictionaries are merged, other values are replaced.

validate(schema: Type[ModelT]) → Self

Validate working config against provided schema. Raises Error if invalid.

save_config(file_path: str | PathLike[str]) → None

Save the internal config to a specific file path (must include filename and extension).

copy_associated_files(target_dir: str | PathLike[str]) → None

Copy all associated files to the target directory.

save(output_dir: str | PathLike[str], config_filename: str | None = None) → None

Save config and copy associated files to target directory.

Parameters:

• output_dir – Target directory path

• config_filename – Config filename with extension (e.g., ‘config.yaml’). If None or empty, only copies associated files.

xflow.utils.get_base_dir() → Path

Returns the directory path of the calling context with cross-environment compatibility.

xflow.utils.load_validated_config(file_path: str | PathLike[str], schema: Type[ModelT] | None = None) → Dict[str, Any]

Load and optionally validate config using any validation schema.

Parameters:

• file_path – Path to config file

• schema – Optional schema class for validation. If None, returns raw dict

Returns:

Dict containing configuration data (validated if schema provided)

xflow.utils.plot_image(img: PILImage | NDArray[Any] | ArrayLike | tf.Tensor | torch.Tensor, cmap: str = None, title: str = None, figsize: tuple = None, vmin: float = None, vmax: float = None, colorbar: bool = True) → None

Plot an image using matplotlib.

Parameters:

• img – Image in any supported format (will be converted automatically)

• cmap – Colormap to use (auto-detected if None)

• title – Plot title

• figsize – Figure size tuple

• vmin – Minimum pixel value for color scaling (auto if None)

• vmax – Maximum pixel value for color scaling (auto if None)

• colorbar – Whether to show the colorbar (default True)

Helper Functions (Internal)

xflow.utils.helper.split_sequence(items: Sequence[T], split_ratio: float = 0.8, seed: int = 42, shuffle: bool = True) → Tuple[List[T], List[T]]

Split a sequence into two parts.

Parameters:

• items – Any sequence (list, tuple, etc.) of type T.

• split_ratio – Ratio for first part (0.0 to 1.0).

• seed – Random seed for reproducibility.

• shuffle – Whether to shuffle before splitting.

Returns:

Tuple of (first_part, second_part) as lists of type T.

xflow.utils.helper.subsample_sequence(items: Sequence[T], n_samples: int | None = None, fraction: float | None = None, strategy: str = 'random', seed: int | None = 42) → List[T]

Subsampling function for any Sequence.

Parameters:

• items – Any sequence (list, tuple, etc.) of type T.

• n_samples – Exact number to sample.

• fraction – Fraction to sample (0.0 to 1.0).

• strategy – “random”, “first”, “last”, “stride”, or “reservoir”.

• seed – Random seed for reproducibility.

Returns:

List of sampled items of type T.

xflow.utils.helper.deep_update(base: MutableMapping[str, Any], updates: Dict[str, Any]) → None

Recursively update a dictionary with another dictionary.

Nested dictionaries are merged, other values are replaced. Modifies base dictionary in-place.

Parameters:

• base – Dictionary to update (modified in-place)

• updates – Dictionary with updates to apply

Example

>>> base = {"a": {"x": 1, "y": 2}, "b": 3}
>>> updates = {"a": {"x": 10, "z": 3}, "c": 4}
>>> deep_update(base, updates)
>>> base
{"a": {"x": 10, "y": 2, "z": 3}, "b": 3, "c": 4}

IO Functions (Internal)

xflow.utils.io.scan_files(root_paths: str | PathLike[str] | List[str | PathLike[str]], extensions: str | List[str] | None = None, return_type: str = 'path', recursive: bool = True) → List[str] | List[Path]

Scan directories for files with specified extensions.

Parameters:

• root_paths – Single path or list of paths to scan

• extensions – File extensions to include (e.g., ‘.jpg’ or [‘.jpg’, ‘.png’]). If None, includes all files.

• return_type – “path” to return Path objects, “str” to return strings

• recursive – Whether to scan subdirectories recursively

Returns:

Sorted list of file paths

xflow.utils.io.copy_file(source_path: str | PathLike[str], target_path: str | PathLike[str], filename: str | None = None) → Path

Copy a file to target directory.

Parameters:

• source_path – Source file path

• target_path – Target directory path

• filename – Optional new filename (uses source filename if None)

Returns:

Path to the copied file

xflow.utils.parser.load_file(file_path: str | PathLike[str]) → Any

Load config file.

xflow.utils.parser.save_file(data: Any, file_path: str | PathLike[str]) → None

Save data to file. Format determined by file extension.