bioamla.batch¶

bioamla.batch ¶

Generic Batch Engine¶

Module-level batch machinery ported from the old BatchServiceBase and BatchCSVHandler. Plain functions + dataclasses, direct file I/O via :mod:pathlib, and raising on error.

Public surface: - :func:run_batch — sequential or thread-parallel item processing. Per-item results are collected from future.result() return values so parallel mode does not lose data. - :func:run_csv_batch — drive a per-row processor over a loaded CSV context (the engine half of the old process_batch_csv). - :func:discover_files — glob-based file discovery. - CSV helpers: :func:load_csv, :func:write_csv, :func:resolve_file_path, :func:resolve_output_path, :func:update_row_path, :func:merge_analysis_results, :func:expand_row_for_segments. - Types: :class:BatchConfig, :class:BatchResult, :class:SegmentInfo, :class:MetadataRow, and :class:CSVBatchContext.

BatchConfig `dataclass` ¶

Configuration for batch operations.

Supports two input modes (mutually exclusive): - Directory mode: provide input_dir to process all files in a directory. - CSV metadata mode: provide input_file pointing to a CSV with a file_name column.

For programmatic usage, validation can be bypassed by setting _skip_validation=True (useful for testing or advanced use cases).

__post_init__ ¶

__post_init__() -> None

Validate mutual exclusivity of input_dir and input_file.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain dictionary.

SegmentInfo `dataclass` ¶

Information about a created audio segment.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain dictionary (paths stringified).

BatchResult `dataclass` ¶

Generic result of batch processing.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain dictionary.

MetadataRow `dataclass` ¶

Single row from a metadata CSV with file path and arbitrary fields.

CSVBatchContext `dataclass` ¶

Context for CSV-based batch processing.

discover_files ¶

discover_files(
    input_dir: str | Path,
    *,
    recursive: bool = True,
    file_filter: Callable[[Path], bool] | None = None,
) -> list[Path]

Discover files under a directory using stdlib globbing.

Parameters:

Name	Type	Description	Default
`input_dir`	`str \| Path`	Directory to search.	required
`recursive`	`bool`	If True, recurse into subdirectories.	`True`
`file_filter`	`Callable[[Path], bool] \| None`	Optional predicate applied to each file path.	`None`

Returns:

Type	Description
`list[Path]`	Sorted list of matching file paths (empty if the directory is missing).

run_batch ¶

run_batch(
    items: list[I],
    process_fn: Callable[[I], T],
    *,
    max_workers: int = 1,
    continue_on_error: bool = True,
    on_progress: Callable[[int, int], None] | None = None,
) -> BatchResult

Run process_fn over items sequentially or in parallel.

In parallel mode (max_workers > 1) a :class:ThreadPoolExecutor is used and per-item return values are collected via future.result() so no data is lost. Successful return values are appended to BatchResult.output_files as strings when they are not None. Threads (not processes) are used so that closure process_fns work and there is no fork/pickling overhead — the per-item audio/ML work runs in GIL-releasing native code (soundfile, librosa, numpy, torch), so it parallelizes well.

Parameters:

Name	Type	Description	Default
`items`	`list[I]`	Items to process.	required
`process_fn`	`Callable[[I], T]`	Callable applied to each item.	required
`max_workers`	`int`	Number of worker threads; 1 runs sequentially.	`1`
`continue_on_error`	`bool`	If True, collect errors and keep going; if False, re-raise the first exception encountered.	`True`
`on_progress`	`Callable[[int, int], None] \| None`	Optional callback `(completed, total)` invoked after each item completes.	`None`

Returns:

Name	Type	Description
`A`	`BatchResult`	class:`BatchResult` summarizing the run.

resolve_file_path ¶

resolve_file_path(file_name: str, csv_dir: Path) -> Path

Resolve a file path relative to the CSV directory.

Parameters:

Name	Type	Description	Default
`file_name`	`str`	Relative or absolute path from the CSV.	required
`csv_dir`	`Path`	Directory containing the CSV file.	required

Returns:

Type	Description
`Path`	Resolved absolute path.

load_csv ¶

load_csv(
    csv_path: str | Path, output_dir: str | None = None
) -> CSVBatchContext

Load a metadata CSV and resolve file paths relative to the CSV directory.

Parameters:

Name	Type	Description	Default
`csv_path`	`str \| Path`	Path to the metadata CSV file.	required
`output_dir`	`str \| None`	Optional output directory for processed files.	`None`

Returns:

Name	Type	Description
`A`	`CSVBatchContext`	class:`CSVBatchContext` with all rows and resolved paths.

Raises:

Type	Description
`NotFoundError`	If the CSV file does not exist.
`InvalidInputError`	If the CSV lacks a `file_name` column.

resolve_output_path ¶

resolve_output_path(
    input_path: Path,
    csv_context: CSVBatchContext,
    new_extension: str | None = None,
) -> Path

Calculate the output path for a processed file.

WITH output_dir: output_dir / relative_structure / filename. WITHOUT output_dir: same location as input (in-place).

Parameters:

Name	Type	Description	Default
`input_path`	`Path`	Original input file path.	required
`csv_context`	`CSVBatchContext`	CSV batch context with output directory info.	required
`new_extension`	`str \| None`	New file extension (e.g. `.wav`) if the format changes.	`None`

Returns:

Type	Description
`Path`	Output file path.

update_row_path ¶

update_row_path(
    row: MetadataRow,
    new_path: Path,
    csv_context: CSVBatchContext,
) -> None

Update a row's file_name to new_path (relative to the CSV if possible).

Parameters:

Name	Type	Description	Default
`row`	`MetadataRow`	Metadata row to update.	required
`new_path`	`Path`	New absolute path after processing.	required
`csv_context`	`CSVBatchContext`	CSV batch context.	required

merge_analysis_results ¶

merge_analysis_results(
    row: MetadataRow, results: dict[str, Any]
) -> None

Merge analysis results into a row's metadata fields.

Parameters:

Name	Type	Description	Default
`row`	`MetadataRow`	Metadata row to update.	required
`results`	`dict[str, Any]`	Result columns to add (e.g. `{'aci': 0.85, 'adi': 0.72}`).	required

expand_row_for_segments ¶

expand_row_for_segments(
    parent_row: MetadataRow,
    segments: list[Any],
    csv_context: CSVBatchContext,
) -> list[MetadataRow]

Create multiple output rows from one input row (for the segment operation).

Parameters:

Name	Type	Description	Default
`parent_row`	`MetadataRow`	Original input row with parent file metadata.	required
`segments`	`list[Any]`	List of :class:`SegmentInfo`-like objects.	required
`csv_context`	`CSVBatchContext`	CSV batch context.	required

Returns:

Type	Description
`list[MetadataRow]`	List of new :class:`MetadataRow` objects (one per segment).

run_csv_batch ¶

run_csv_batch(
    context: CSVBatchContext,
    process_row: Callable[[MetadataRow], Any],
    *,
    max_workers: int = 1,
    continue_on_error: bool = True,
    quiet: bool = False,
    on_progress: Callable[[int, int], None] | None = None,
) -> BatchResult

Run process_row over every row of a loaded CSV context.

Existence of each row's resolved file_path is checked up-front (a missing file is recorded as a failure, matching the old process_batch_csv behaviour). process_row is invoked with the :class:MetadataRow and may mutate it in place (e.g. merge result columns via :func:merge_analysis_results, update its path via :func:update_row_path, or stash segment info for later expansion). Its non-None return value is appended to :attr:BatchResult.output_files.

The caller is responsible for calling :func:write_csv (and any row expansion) afterwards so it controls the final CSV shape.

Parameters:

Name	Type	Description	Default
`context`	`CSVBatchContext`	Loaded :class:`CSVBatchContext`.	required
`process_row`	`Callable[[MetadataRow], Any]`	Callable applied to each existing row. Runs in-process (sequentially) so it may close over and mutate `context`.	required
`max_workers`	`int`	Reserved for parity; CSV rows are processed sequentially because `process_row` typically mutates shared state.	`1`
`continue_on_error`	`bool`	Collect per-row errors and keep going if True.	`True`
`quiet`	`bool`	Suppress per-error stderr prints.	`False`
`on_progress`	`Callable[[int, int], None] \| None`	Optional `(completed, total)` progress callback.	`None`

Returns:

Name	Type	Description
`A`	`BatchResult`	class:`BatchResult` summarizing the run.

write_csv ¶

write_csv(context: CSVBatchContext) -> Path

Write the updated metadata CSV to its output location.

Preserves all original columns, adds new columns from analysis results (column union), updates file_name paths, and writes to output_dir if specified else in-place.

Parameters:

Name	Type	Description	Default
`context`	`CSVBatchContext`	CSV batch context with all rows.	required

Returns:

Type	Description
`Path`	Path to the written CSV file.

bioamla.batch¶

bioamla.batch ¶

Generic Batch Engine¶

BatchConfig dataclass ¶

__post_init__ ¶

to_dict ¶

SegmentInfo dataclass ¶

to_dict ¶

BatchResult dataclass ¶

to_dict ¶

MetadataRow dataclass ¶

CSVBatchContext dataclass ¶

discover_files ¶

run_batch ¶

resolve_file_path ¶

load_csv ¶

resolve_output_path ¶

update_row_path ¶

merge_analysis_results ¶

expand_row_for_segments ¶

run_csv_batch ¶

write_csv ¶

BatchConfig `dataclass` ¶

SegmentInfo `dataclass` ¶

BatchResult `dataclass` ¶

MetadataRow `dataclass` ¶

CSVBatchContext `dataclass` ¶