Utilities & Helpers¶

taters.helpers.feature_gather ¶

AggregationPlan `dataclass` ¶

AggregationPlan(
    group_by,
    per_file=True,
    stats=("mean", "std"),
    exclude_cols=(),
    include_regex=None,
    exclude_regex=None,
    dropna=False,
)

Plan describing how numeric feature columns should be aggregated.

Parameters:

Name	Type	Description	Default
`group_by`	`Sequence[str]`	One or more column names used as grouping keys (e.g., `["speaker"]`).	required
`per_file`	`bool`	If True, include `"source"` in the grouping keys to aggregate within each input file; if False, aggregate across all files globally.	`True`
`stats`	`Sequence[str]`	Statistical reductions to compute for each numeric feature column. Values are passed to `pandas.DataFrame.agg` (e.g., `"mean"`, `"std"`, `"median"`, etc.).	`("mean", "std")`
`exclude_cols`	`Sequence[str]`	Columns to drop before filtering/selecting numeric features (e.g., timestamps or free text).	`()`
`include_regex`	`str or None`	Optional regex; if provided, only columns matching this pattern are kept (after excluding `exclude_cols`).	`None`
`exclude_regex`	`str or None`	Optional regex; if provided, columns matching this pattern are removed (after applying `include_regex`, if any).	`None`
`dropna`	`bool`	Whether to drop rows with NA in any of the group-by keys before grouping. The default keeps them, so rows with a missing key land in their own clearly-labeled group instead of vanishing from the output.	`False`

Notes

This plan is consumed by :func:aggregate_features. Column filtering happens before numeric selection; only columns that remain and can be coerced to numeric will be aggregated.

aggregate_features ¶

aggregate_features(
    *,
    root_dir,
    pattern="*.csv",
    recursive=True,
    delimiter=",",
    encoding="utf-8-sig",
    add_source_path=False,
    plan,
    out_csv=None,
    overwrite_existing=False,
    verbose=True,
    on_progress=None
)

Discover files, read, concatenate, and aggregate numeric columns per plan.

This function consolidates CSVs from a single folder, filters columns, coerces candidate features to numeric, groups by the specified keys, and computes the requested statistics. Output columns for aggregated features are flattened with the pattern "{column}__{stat}".

Parameters:

Name	Type	Description	Default
`root_dir`	`PathLike`	Folder containing per-item CSVs, or a single CSV file.	required
`pattern`	`str`	Glob pattern for selecting files.	`"*.csv"`
`recursive`	`bool`	Recurse into subdirectories when True.	`True`
`delimiter`	`str`	CSV delimiter.	`","`
`encoding`	`str`	CSV encoding for read/write.	`"utf-8-sig"`
`add_source_path`	`bool`	If True, include absolute path in `"source_path"` prior to filtering.	`False`
`plan`	`AggregationPlan`	Aggregation configuration (group keys, stats, filters, NA handling).	required
`out_csv`	`PathLike or None`	Output path. If None, defaults to `<root_dir_parent>/<root_dir_name>.csv`.	`None`
`overwrite_existing`	`bool`	If False and `out_csv` exists, return it without recomputation.	`False`

Returns:

Type	Description
`Path`	Path to the written CSV of aggregated features.

Raises:

Type	Description
`FileNotFoundError`	If no files match the pattern under `root_dir`.
`RuntimeError`	If files were found but none could be read successfully.
`ValueError`	If required group-by columns are missing, or if no numeric columns remain after filtering, or if per-file grouping is requested but the `"source"` column is absent.

Notes

Group keys are preserved as leading columns in the output. The output places "source" (and optionally "source_path") first when present.

Source code in src\taters\helpers\feature_gather.py

@records_settings(binding=GATHER_BINDING, outputs=("out_csv",))
def aggregate_features(
    *,
    root_dir: PathLike,
    pattern: str = "*.csv",
    recursive: bool = True,
    delimiter: str = ",",
    encoding: str = "utf-8-sig",
    add_source_path: bool = False,
    plan: AggregationPlan,
    out_csv: Optional[PathLike] = None,
    overwrite_existing: bool = False,
    verbose: bool = True,
    on_progress: Optional[Callable[..., None]] = None,
) -> Path:
    """
    Discover files, read, concatenate, and aggregate numeric columns per plan.

    This function consolidates CSVs from a single folder, filters columns,
    coerces candidate features to numeric, groups by the specified keys,
    and computes the requested statistics. Output columns for aggregated
    features are flattened with the pattern ``"{column}__{stat}"``.

    Parameters
    ----------
    root_dir : PathLike
        Folder containing per-item CSVs, or a single CSV file.
    pattern : str, default="*.csv"
        Glob pattern for selecting files.
    recursive : bool, default=True
        Recurse into subdirectories when True.
    delimiter : str, default=","
        CSV delimiter.
    encoding : str, default="utf-8-sig"
        CSV encoding for read/write.
    add_source_path : bool, default=False
        If True, include absolute path in ``"source_path"`` prior to filtering.
    plan : AggregationPlan
        Aggregation configuration (group keys, stats, filters, NA handling).
    out_csv : PathLike or None, default=None
        Output path. If None, defaults to
        ``<root_dir_parent>/<root_dir_name>.csv``.
    overwrite_existing : bool, default=False
        If False and `out_csv` exists, return it without recomputation.

    Returns
    -------
    pathlib.Path
        Path to the written CSV of aggregated features.

    Raises
    ------
    FileNotFoundError
        If no files match the pattern under `root_dir`.
    RuntimeError
        If files were found but none could be read successfully.
    ValueError
        If required group-by columns are missing,
        or if no numeric columns remain after filtering,
        or if per-file grouping is requested but the ``"source"`` column is absent.

    Notes
    -----
    Group keys are preserved as leading columns in the output. The output places
    ``"source"`` (and optionally ``"source_path"``) first when present.
    """

    root = Path(root_dir)
    if out_csv is None:
        out_csv = root.parent / f"{root.name}.csv"
    out_csv = Path(out_csv)
    out_csv.parent.mkdir(parents=True, exist_ok=True)

    if out_csv.exists() and not overwrite_existing:
        if verbose:
            print(f"Aggregated feature output file already exists; returning existing file: {out_csv}")
        return out_csv

    files = list(_iter_csv_files(root, pattern=pattern, recursive=recursive))
    if not files:
        raise FileNotFoundError(f"No files matched {pattern} under {root}")

    announce(on_progress, "reading feature tables")
    ticker = Ticker(on_progress, len(files))
    frames = []
    for fp in files:
        try:
            frames.append(
                _read_csv_add_source(
                    fp,
                    delimiter=delimiter,
                    encoding=encoding,
                    add_source_path=add_source_path,
                )
            )
        except Exception as e:
            if verbose:
                print(f"[aggregate] WARNING: failed to read {fp}: {e}")
        ticker.tick(message="reading feature tables")

    if not frames:
        raise RuntimeError("No CSVs could be read successfully.")

    df = pd.concat(frames, axis=0, ignore_index=True)

    # if we're aggregating across files (per_file=False), we promote the inner
    # keys (e.g., 'source.1' -> 'source') and demote the file-level keys.
    if not plan.per_file:
        df = _promote_inner_keys(df, plan.group_by)


    def _resolve_keys(base_keys, columns):
        cols = set(columns)
        resolved = []
        for k in base_keys:
            if k in cols:
                resolved.append(k)
                continue
            # go looking for numbered variants like 'k.1', 'k.2', ...
            prefix = f"{k}."
            candidates = [c for c in columns if c == f"{k}.1" or c.startswith(prefix)]
            if candidates:
                # sorted first, so we pick the same one every time
                resolved.append(sorted(candidates)[0])
            else:
                # leave it unresolved; we'll error out below with a helpful message
                resolved.append(k)
        return resolved

    # build the (base) group keys from the plan
    group_keys = list(plan.group_by)
    if plan.per_file:
        if "source" not in df.columns:
            raise ValueError("source column is missing; cannot group per_file.")
        group_keys = ["source"] + group_keys

    # resolve any collisions against the columns we actually have
    group_keys = _resolve_keys(group_keys, df.columns)

    # now we filter, but we ALWAYS keep the group keys
    df_f = _filter_columns(
        df,
        exclude_cols=tuple(plan.exclude_cols) + ("source_path",),
        include_regex=plan.include_regex,
        exclude_regex=plan.exclude_regex,
        must_keep=group_keys,
    )

    missing = [k for k in group_keys if k not in df_f.columns]
    if missing:
        raise ValueError(f"Missing group-by columns in data: {missing}")

    # everything that isn't a key is a candidate feature; keep the numeric ones
    feature_cols = [c for c in df_f.columns if c not in set(group_keys)]
    numeric_df = _numeric_subframe(df_f[feature_cols])
    if numeric_df.empty:
        raise ValueError("No numeric columns available for aggregation after filtering.")

    # stick the group keys back on so we can group by them
    gdf = pd.concat([df_f[group_keys].reset_index(drop=True),
                     numeric_df.reset_index(drop=True)], axis=1)

    agg_ops = {c: list(plan.stats) for c in numeric_df.columns}
    grouped = gdf.groupby(group_keys, dropna=plan.dropna).agg(agg_ops)

    # flatten the MultiIndex columns down to '<col>__<stat>'
    grouped.columns = [f"{c}__{stat}" for (c, stat) in grouped.columns]
    grouped = grouped.reset_index()

    # make sure 'source' (and 'source_path' if present) lead the output
    cols = list(grouped.columns)
    lead = [c for c in ("source", "source_path") if c in cols]
    rest = [c for c in cols if c not in lead]
    grouped = grouped[lead + rest]

    with atomic_write(out_csv, mode="w", newline="", encoding=encoding) as fh:
        grouped.to_csv(fh, index=False)
    return out_csv

feature_gather ¶

feature_gather(
    *,
    root_dir,
    pattern="*.csv",
    recursive=True,
    delimiter=",",
    encoding="utf-8-sig",
    add_source_path=False,
    aggregate=False,
    plan=None,
    group_by=None,
    per_file=True,
    stats=("mean", "std"),
    exclude_cols=(),
    include_regex=None,
    exclude_regex=None,
    dropna=False,
    out_csv=None,
    overwrite_existing=False,
    verbose=True,
    on_progress=None
)

Single entry point to concatenate or aggregate feature CSVs from one folder.

If aggregate=False, CSVs are concatenated with origin metadata (see :func:gather_csvs_to_one). If aggregate=True, numeric feature columns are aggregated per the provided or constructed plan (see :func:aggregate_features).

Parameters:

Name	Type	Description	Default
`root_dir`	`PathLike`	Folder containing per-item CSVs (or a single CSV file).	required
`pattern`	`str`	Glob pattern for selecting CSV files.	`"*.csv"`
`recursive`	`bool`	Recurse into subdirectories when True.	`True`
`delimiter`	`str`	CSV delimiter.	`","`
`encoding`	`str`	CSV encoding.	`"utf-8-sig"`
`add_source_path`	`bool`	If True, include a `"source_path"` column in outputs.	`False`
`aggregate`	`bool`	Toggle aggregation mode. If False, files are concatenated.	`False`
`plan`	`AggregationPlan or None`	Explicit plan for aggregation. Required if `aggregate=True` and `group_by` is not given.	`None`
`group_by`	`Sequence[str] or None`	Quick-plan keys. Used only when `aggregate=True` and `plan` is None.	`None`
`per_file`	`bool`	Quick-plan flag; include `"source"` in grouping keys to aggregate per file.	`True`
`stats`	`Sequence[str]`	Quick-plan statistics to compute per numeric column.	`("mean", "std")`
`exclude_cols`	`Sequence[str]`	Quick-plan columns to drop before numeric selection.	`()`
`include_regex`	`str or None`	Quick-plan regex to include feature columns by name.	`None`
`exclude_regex`	`str or None`	Quick-plan regex to exclude feature columns by name.	`None`
`dropna`	`bool`	Quick-plan NA handling for group keys. When True, rows whose group key is missing are dropped before aggregating.	`False`
`out_csv`	`PathLike or None`	Output CSV path. If None, defaults to `<root_dir_parent>/<root_dir_name>.csv`.	`None`
`overwrite_existing`	`bool`	If False and `out_csv` exists, the existing path is returned without recomputation.	`False`

Returns:

Type	Description
`Path`	Path to the resulting CSV.

Raises:

Type	Description
`ValueError`	If `aggregate=True` and neither `plan` nor `group_by` is provided.

gather_csvs_to_one ¶

gather_csvs_to_one(
    *,
    root_dir,
    pattern="*.csv",
    recursive=True,
    delimiter=",",
    encoding="utf-8-sig",
    add_source_path=False,
    out_csv=None,
    overwrite_existing=False,
    verbose=True,
    on_progress=None
)

Concatenate many CSVs into a single CSV with origin metadata.

Each input CSV is loaded (all columns as object dtype), a leading "source" column is inserted (and optionally "source_path"), and rows are appended. The final CSV ensures "source" (and, if present, "source_path") lead the column order.

Parameters:

Name	Type	Description	Default
`root_dir`	`PathLike`	Folder containing CSVs, or a single CSV file.	required
`pattern`	`str`	Glob pattern for selecting files.	`"*.csv"`
`recursive`	`bool`	Recurse into subdirectories when True.	`True`
`delimiter`	`str`	CSV delimiter.	`","`
`encoding`	`str`	CSV encoding for read/write.	`"utf-8-sig"`
`add_source_path`	`bool`	If True, include absolute path in `"source_path"`.	`False`
`out_csv`	`PathLike or None`	Output path. If None, defaults to `<root_dir_parent>/<root_dir_name>.csv`.	`None`
`overwrite_existing`	`bool`	If False and `out_csv` exists, return it without recomputation.	`False`

Returns:

Type	Description
`Path`	Path to the written CSV.

Raises:

Type	Description
`FileNotFoundError`	If no files match the pattern under `root_dir`.
`RuntimeError`	If files were found but none could be read successfully.

Notes

Input rows are not type-coerced beyond object dtype. Column order from inputs is preserved after the leading origin columns.

Source code in src\taters\helpers\feature_gather.py

@records_settings(binding=GATHER_BINDING, outputs=("out_csv",))
def gather_csvs_to_one(
    *,
    root_dir: PathLike,
    pattern: str = "*.csv",
    recursive: bool = True,
    delimiter: str = ",",
    encoding: str = "utf-8-sig",
    add_source_path: bool = False,
    out_csv: Optional[PathLike] = None,
    overwrite_existing: bool = False,
    verbose: bool = True,
    on_progress: Optional[Callable[..., None]] = None,
) -> Path:
    """
    Concatenate many CSVs into a single CSV with origin metadata.

    Each input CSV is loaded (all columns as object dtype), a leading
    ``"source"`` column is inserted (and optionally ``"source_path"``), and
    rows are appended. The final CSV ensures ``"source"`` (and, if present,
    ``"source_path"``) lead the column order.

    Parameters
    ----------
    root_dir : PathLike
        Folder containing CSVs, or a single CSV file.
    pattern : str, default="*.csv"
        Glob pattern for selecting files.
    recursive : bool, default=True
        Recurse into subdirectories when True.
    delimiter : str, default=","
        CSV delimiter.
    encoding : str, default="utf-8-sig"
        CSV encoding for read/write.
    add_source_path : bool, default=False
        If True, include absolute path in ``"source_path"``.
    out_csv : PathLike or None, default=None
        Output path. If None, defaults to
        ``<root_dir_parent>/<root_dir_name>.csv``.
    overwrite_existing : bool, default=False
        If False and `out_csv` exists, return it without recomputation.

    Returns
    -------
    pathlib.Path
        Path to the written CSV.

    Raises
    ------
    FileNotFoundError
        If no files match the pattern under `root_dir`.
    RuntimeError
        If files were found but none could be read successfully.

    Notes
    -----
    Input rows are not type-coerced beyond object dtype. Column order from
    inputs is preserved after the leading origin columns.
    """

    root = Path(root_dir)
    if out_csv is None:
        out_csv = root.parent / f"{root.name}.csv"

    out_csv = Path(out_csv)
    out_csv.parent.mkdir(parents=True, exist_ok=True)

    if out_csv.exists() and not overwrite_existing:
        if verbose:
            print(f"Aggregated feature output file already exists; returning existing file: {out_csv}")
        return out_csv

    files = list(_iter_csv_files(root, pattern=pattern, recursive=recursive))
    if not files:
        raise FileNotFoundError(f"No files matched {pattern} under {root}")

    # we tick once per table read: a big embeddings folder took a minute under
    # a bare spinner, which looks an awful lot like a hang.
    announce(on_progress, "reading feature tables")
    ticker = Ticker(on_progress, len(files))
    frames = []
    for fp in files:
        try:
            frames.append(
                _read_csv_add_source(
                    fp,
                    delimiter=delimiter,
                    encoding=encoding,
                    add_source_path=add_source_path,
                )
            )
        except Exception as e:
            if verbose:
                print(f"[gather] WARNING: failed to read {fp}: {e}")
        ticker.tick(message="reading feature tables")

    if not frames:
        raise RuntimeError("No CSVs could be read successfully.")

    merged = pd.concat(frames, axis=0, ignore_index=True)

    # make sure 'source' is first (and 'source_path' next, if we have it)
    cols = list(merged.columns)
    if "source" in cols:
        lead = ["source"] + (["source_path"] if "source_path" in cols else [])
        rest = [c for c in cols if c not in lead]
        merged = merged[lead + rest]

    # atomic write, like every other skip-if-exists table: the next run will
    # treat this one as finished, so a Ctrl-C mid-write used to leave a sticky
    # truncated table behind.
    with atomic_write(out_csv, mode="w", newline="", encoding=encoding) as fh:
        merged.to_csv(fh, index=False)
    return out_csv

make_plan ¶

make_plan(
    *,
    group_by,
    per_file=True,
    stats=("mean", "std"),
    exclude_cols=(),
    include_regex=None,
    exclude_regex=None,
    dropna=False
)

Create an :class:AggregationPlan from simple arguments.

Parameters:

Name	Type	Description	Default
`group_by`	`Sequence[str]`	Grouping key(s) to use (e.g., `["speaker"]`).	required
`per_file`	`bool`	If True, group within files by including `"source"` in group keys.	`True`
`stats`	`Sequence[str]`	Statistical reductions to compute per numeric column.	`("mean", "std")`
`exclude_cols`	`Sequence[str]`	Columns to drop prior to feature selection.	`()`
`include_regex`	`str or None`	Regex to include feature columns by name.	`None`
`exclude_regex`	`str or None`	Regex to exclude feature columns by name.	`None`
`dropna`	`bool`	Drop rows with NA in any group key. Off by default so rows with a missing key are still reported rather than silently discarded.	`False`

Returns:

Type	Description
`AggregationPlan`	A configured plan instance for :func:`aggregate_features`.

Source code in src\taters\helpers\feature_gather.py

def make_plan(
    *,
    group_by: Sequence[str],
    per_file: bool = True,
    stats: Sequence[str] = ("mean", "std"),
    exclude_cols: Sequence[str] = (),
    include_regex: Optional[str] = None,
    exclude_regex: Optional[str] = None,
    dropna: bool = False,
) -> AggregationPlan:
    """
    Create an :class:`AggregationPlan` from simple arguments.

    Parameters
    ----------
    group_by : Sequence[str]
        Grouping key(s) to use (e.g., ``["speaker"]``).
    per_file : bool, default=True
        If True, group within files by including ``"source"`` in group keys.
    stats : Sequence[str], default=("mean", "std")
        Statistical reductions to compute per numeric column.
    exclude_cols : Sequence[str], default=()
        Columns to drop prior to feature selection.
    include_regex : str or None, default=None
        Regex to include feature columns by name.
    exclude_regex : str or None, default=None
        Regex to exclude feature columns by name.
    dropna : bool, default=False
        Drop rows with NA in any group key. Off by default so rows with a
        missing key are still reported rather than silently discarded.

    Returns
    -------
    AggregationPlan
        A configured plan instance for :func:`aggregate_features`.
    """

    return AggregationPlan(
        group_by=tuple(group_by),
        per_file=per_file,
        stats=tuple(stats),
        exclude_cols=tuple(exclude_cols),
        include_regex=include_regex,
        exclude_regex=exclude_regex,
        dropna=dropna,
    )

taters.helpers.find_files ¶

find_files ¶

find_files(
    root_dir,
    *,
    file_type="video",
    extensions=None,
    recursive=True,
    follow_symlinks=False,
    include_hidden=False,
    include_globs=None,
    exclude_globs=None,
    absolute=True,
    sort=True,
    ffprobe_verify=False
)

Discover media files under a folder using smart, FFmpeg-friendly filters.

You can either (a) choose a built-in group of extensions via file_type ("audio"|"video"|"image"|"subtitle"|"archive"|"any") or (b) pass an explicit list of extensions to match. Matching is case-insensitive; dots are optional (e.g., ".wav" and "wav" are equivalent). Hidden files and directories are excluded by default.

For audio/video, ffprobe_verify=True additionally checks that at least one corresponding stream is present (e.g., exclude MP4s with no audio when file_type="audio"). This is slower but robust when your dataset contains “container only” files.

Parameters:

Name	Type	Description	Default
`root_dir`	`str \| PathLike`	Folder to scan.	required
`file_type`	`str`	Built-in group selector. Ignored if `extensions` is provided.	`'video'`
`extensions`	`Optional[Sequence[str]]`	Explicit extensions to include (e.g., `[".wav",".flac"]`). Overrides `file_type`.	`None`
`recursive`	`bool`	Recurse into subfolders. Default: `True`.	`True`
`follow_symlinks`	`bool`	Follow directory symlinks during traversal. Default: `False`.	`False`
`include_hidden`	`bool`	Include dot-files and dot-dirs. Default: `False`.	`False`
`include_globs`	`Optional[Sequence[str]]`	Additional glob filters applied after extension filtering; `include_globs` uses OR-semantics, then `exclude_globs` removes matches.	`None`
`absolute`	`bool`	Return absolute paths when `True` (default) else relative to `root_dir`.	`True`
`sort`	`bool`	Sort lexicographically (case-insensitive). Default: `True`.	`True`
`ffprobe_verify`	`bool`	For `audio`/`video`, keep only files where `ffprobe` reports ≥1 matching stream.	`False`

Returns:

Type	Description
`list[Path]`	The matched files.

Raises:

Type	Description
`FileNotFoundError`	If `root_dir` does not exist.
`ValueError`	If `file_type` is not one of the supported groups.

Examples:

Find all videos (recursive), as absolute paths:

>>> find_files("dataset", file_type="video")

Use explicit extensions and keep paths relative:

>>> find_files("dataset", extensions=[".wav",".flac"], absolute=False)

Only include files matching a glob and exclude temp folders:

>>> find_files("dataset", file_type="audio",
...            include_globs=["**/*session*"], exclude_globs=["**/tmp/**"])

Verify playable audio streams exist:

>>> find_files("dataset", file_type="audio", ffprobe_verify=True)

Source code in src\taters\helpers\find_files.py

def find_files(
    root_dir: str | os.PathLike,
    *,
    file_type: str = "video",                            # 'audio' | 'video' | 'image' | 'subtitle' | 'archive' | 'any'
    extensions: Optional[Sequence[str]] = None,     # explicit extensions override group (e.g., ['.wav','.flac'])
    recursive: bool = True,
    follow_symlinks: bool = False,
    include_hidden: bool = False,
    include_globs: Optional[Sequence[str]] = None,  # e.g., ['**/*session*']
    exclude_globs: Optional[Sequence[str]] = None,  # e.g., ['**/temp/**']
    absolute: bool = True,
    sort: bool = True,
    ffprobe_verify: bool = False,                   # confirm stream presence via ffprobe (audio/video only)
) -> List[Path]:
    """
    Discover media files under a folder using smart, FFmpeg-friendly filters.

    You can either (a) choose a built-in **group** of extensions via `file_type`
    (`"audio"|"video"|"image"|"subtitle"|"archive"|"any"`) or (b) pass an explicit
    list of `extensions` to match. Matching is case-insensitive; dots are optional
    (e.g., `".wav"` and `"wav"` are equivalent). Hidden files and directories are
    excluded by default.

    For audio/video, `ffprobe_verify=True` additionally checks that at least one
    corresponding stream is present (e.g., exclude MP4s with no audio when
    `file_type="audio"`). This is slower but robust when your dataset contains
    “container only” files.

    Parameters
    ----------
    root_dir
        Folder to scan.
    file_type
        Built-in group selector. Ignored if `extensions` is provided.
    extensions
        Explicit extensions to include (e.g., `[".wav",".flac"]`). Overrides `file_type`.
    recursive
        Recurse into subfolders. Default: `True`.
    follow_symlinks
        Follow directory symlinks during traversal. Default: `False`.
    include_hidden
        Include dot-files and dot-dirs. Default: `False`.
    include_globs / exclude_globs
        Additional glob filters applied after extension filtering; `include_globs`
        uses OR-semantics, then `exclude_globs` removes matches.
    absolute
        Return absolute paths when `True` (default) else relative to `root_dir`.
    sort
        Sort lexicographically (case-insensitive). Default: `True`.
    ffprobe_verify
        For `audio`/`video`, keep only files where `ffprobe` reports ≥1 matching
        stream.

    Returns
    -------
    list[pathlib.Path]
        The matched files.

    Raises
    ------
    FileNotFoundError
        If `root_dir` does not exist.
    ValueError
        If `file_type` is not one of the supported groups.

    Examples
    --------
    Find all videos (recursive), as absolute paths:

    >>> find_files("dataset", file_type="video")

    Use explicit extensions and keep paths relative:

    >>> find_files("dataset", extensions=[".wav",".flac"], absolute=False)

    Only include files matching a glob and exclude temp folders:

    >>> find_files("dataset", file_type="audio",
    ...            include_globs=["**/*session*"], exclude_globs=["**/tmp/**"])

    Verify playable audio streams exist:

    >>> find_files("dataset", file_type="audio", ffprobe_verify=True)
    """
    root_dir = Path(root_dir)
    if not root_dir.exists():
        raise FileNotFoundError(f"Root path not found: {root_dir}")

    if extensions:
        allowed = {_norm_ext(e) for e in extensions}
    else:
        if file_type not in GROUPS:
            raise ValueError(f"Unknown kind '{file_type}'. Choose from {', '.join(GROUPS.keys())}.")
        allowed = set(GROUPS[file_type])

    cand = (
        p for p in _iter_files(root_dir, recursive=recursive, follow_symlinks=follow_symlinks, include_hidden=include_hidden)
        if p.is_file() and _match_ext(p, allowed)
    )

    cand = _glob_filter(
        cand,
        includes=include_globs or [],
        excludes=exclude_globs or [],
    )

    out: List[Path] = []
    for p in cand:
        if ffprobe_verify and file_type in ("audio", "video"):
            if not _ffprobe_has_stream(p, file_type):
                continue
        out.append(p.resolve() if absolute else p)

    if sort:
        out.sort(key=lambda x: str(x).lower())
    return out

taters.helpers.text_gather ¶

csv_to_analysis_ready_csv ¶

csv_to_analysis_ready_csv(
    *,
    csv_path,
    out_csv=None,
    overwrite_existing=False,
    text_cols=None,
    id_cols=None,
    mode="concat",
    group_by=None,
    delimiter=None,
    encoding=DEFAULT_ENCODING,
    joiner=DEFAULT_JOINER,
    num_buckets=1024,
    max_open_bucket_files=64,
    tmp_root=None,
    include_id_cols=True,
    carry_cols=None,
    agg_cols=None,
    row_filters=None,
    verbose=True,
    on_progress=None
)

Stream a (possibly huge) CSV into a compact analysis-ready CSV with a stable schema and optional external grouping.

Output schema

Always writes a header and enforces a consistent column order:

• No grouping: text_id,text (plus source_col if mode="separate") • With grouping: text_id,text,group_count (plus source_col if mode="separate")

carry_cols inserts the named source columns between the identifiers and text in both shapes.

Where: - text_id is either the composed ID from id_cols or row_<n> when id_cols=None. - mode="concat" joins all text_cols using joiner per row or group. - mode="separate" emits one row per (row_or_group, text_col) and fills source_col with the contributing column name.

Grouping at scale

If group_by is provided, the function performs a two-pass external grouping that does not require presorting: 1) Hash-partition rows to on-disk “bucket” CSVs (bounded writers with LRU). 2) Aggregate each bucket into final rows (concat or separate mode), writing group_count to record how many pieces contributed.

Parameters:

Name	Type	Description	Default
`csv_path`	`PathLike`	Source CSV with at least the columns in `text_cols` (and `group_by` if grouping).	required
`out_csv`	`PathLike \| None`	Destination CSV. If `None`, a name is derived from the input and options (e.g., `<stem>_grouped_<group_by>.csv` or `<stem>_concat_<cols>.csv`).	`None`
`overwrite_existing`	`bool`	If `False` (default) and `out_csv` exists, the function returns early.	`False`
`text_cols`	`Sequence[str] \| None`	Text fields to concatenate or emit separately. May be empty or None: the spreadsheet is then wrangled without any text -- rows are still combined, counted, summarized and carried -- and the output has no `text` column.	`None`
`id_cols`	`Sequence[str] \| None`	Optional columns to compose `text_id` when not grouping. When omitted, a synthetic `row_<n>` is used.	`None`
`carry_cols`	`Sequence[str] \| None`	Source columns to copy through to the output untouched. Distinct from `id_cols`, which compose `text_id` and so cannot be used to carry a column whose values repeat -- naming `speaker` there would give every utterance by one person the same `text_id`. Downstream analyzers offer a `pass_through_cols` argument that is meaningless unless the gather that feeds them preserved the columns, which is what this is for. When grouping, a carried column is written only where its value is the same for every row in the group; where rows disagree the cell is blank, because there is no honest single answer and picking the first row's is the kind of quiet wrong answer that survives into a published table. Names in `carry_cols` that the source does not have are dropped with a warning rather than raising: unlike `text_cols`, a carried column is a convenience, and half the presets that ask for `speaker` run over inputs that never had one.	`None`
`agg_cols`	`Mapping[str, str] \| Sequence[str] \| None`	Number columns to summarize per group, e.g. a per-post score becoming the average score of everything a user wrote. A sequence of names means "the mean of each"; a mapping picks the statistic per column from `mean`, `sum`, `min` and `max`. Each summary lands in its own output column named `<col>_<stat>`, placed after `group_count`. Values that are blank or not parseable as numbers are skipped rather than treated as zero -- a missing score is missing, not 0 -- and a group with no numeric values at all gets a blank cell, for the same NA-is-not-zero reason a disagreeing carried column does.	`None`
`row_filters`	`Sequence[Sequence[object]] \| None`	Which rows to keep, as `[[column, operator, value], ...]` over the source spreadsheet's own columns -- `[["age", ">=", 18]]`, `[["condition", "in", ["A", "B"]]]`. A row has to clear all of them. The operators and the comparison are :mod:`taters.helpers.row_filter`'s, the same ones the statistics use, so a filter means one thing wherever it is applied. Only the spreadsheet's own columns, because nothing has been measured yet: a word count is not available here and does not belong here either. What somebody knows at this point is what they collected -- a screener somebody failed, a condition they are not analyzing, an age below the one they meant to study. It matters most when rows are being combined, since a row inside somebody else's joined text cannot be taken out again afterwards; that is why it is applied before the rows are bucketed rather than filtered later. Summaries run over every row of the group, including rows whose text cells are empty -- `group_count` counts only the rows whose text was joined, so the two can disagree. To keep that visible, each summary is followed by a `<col>_n` column giving the number of numeric values it actually used. Requires `group_by`: without groups there is nothing to summarize over, and silently ignoring the ask would hide a real mistake. Unlike `carry_cols`, a name the source lacks raises: a summary is an explicit computation, not a convenience.	`None`
`mode`	`str`	`"concat"` (default) or `"separate"`. See schema above.	`'concat'`
`group_by`	`Sequence[str] \| None`	Optional list of columns to aggregate by; works on unsorted CSVs.	`None`
`delimiter`	`str \| None`	Parsing/formatting options. If `delimiter=None`, sniffs from a sample.	`None`
`encoding`	`str \| None`	Parsing/formatting options. If `delimiter=None`, sniffs from a sample.	`None`
`joiner`	`str \| None`	Parsing/formatting options. If `delimiter=None`, sniffs from a sample.	`None`
`num_buckets`	`int`	External grouping controls (partition count, LRU limit, temp root).	`1024`
`max_open_bucket_files`	`int`	External grouping controls (partition count, LRU limit, temp root).	`1024`
`tmp_root`	`int`	External grouping controls (partition count, LRU limit, temp root).	`1024`
`include_id_cols`	`bool`	When not grouping, write the id columns beside `text_id`. Grouped output carries the grouping columns instead, whatever this says.	`True`
`on_progress`		Optional `(done, total, message)` callback. The gather names each phase as it runs -- counting the rows, then copying them (or, when grouping, sorting them into groups and then combining each group) -- because a silent spinner over a large file reads as a hang.	`None`

Returns:

Type	Description
`Path`	Path to the analysis-ready CSV.

Raises:

Type	Description
`ValueError`	If required columns are missing or `mode` is invalid.

Examples:

Concatenate two text fields per row:

>>> csv_to_analysis_ready_csv(
...     csv_path="transcripts.csv",
...     text_cols=["prompt","response"],
...     id_cols=["speaker"],
... )

Group by speaker and join rows:

>>> csv_to_analysis_ready_csv(
...     csv_path="transcripts.csv",
...     text_cols=["text"],
...     group_by=["speaker"],
... )

Source code in src\taters\helpers\text_gather.py

@records_settings(
    # which rows and columns we read, and which metadata rode along.
    # `joiner` is NOT in here on purpose: joining two text columns with
    # " " rather than "\n" changes sentence segmentation, so it decides
    # every number every downstream analyzer produces.
    binding=("csv_path", "text_cols", "id_cols", "delimiter",
             "encoding", "include_id_cols", "carry_cols", "agg_cols"),
    # this takes seconds to redo, and it's the one table every join depends
    # on: an existing copy with no record gets rebuilt rather than trusted.
    redo_without_record=True,
    grain=TEXT_GRAIN,
    outputs=("out_csv",),
    # this step decides what text EXISTS. its settings describe the
    # dataset, so we record them but keep them out of the chain that
    # downstream steps compare -- a model has to be applicable to a corpus
    # assembled differently, which is the whole point of saving one.
    defines_text=True)
def csv_to_analysis_ready_csv(
    *,
    csv_path: PathLike,
    out_csv: PathLike | None = None,
    overwrite_existing: bool = False,
    text_cols: Sequence[str] | None = None,
    id_cols: Sequence[str] | None = None,
    mode: str = "concat",
    group_by: Sequence[str] | None = None,
    delimiter: str | None = None,
    encoding: str = DEFAULT_ENCODING,
    joiner: str = DEFAULT_JOINER,
    num_buckets: int = 1024,
    max_open_bucket_files: int = 64,
    tmp_root: PathLike | None = None,
    include_id_cols: bool = True,
    carry_cols: Sequence[str] | None = None,
    agg_cols: Mapping[str, str] | Sequence[str] | None = None,
    row_filters: Sequence[Sequence[object]] | None = None,
    verbose: bool = True,
    on_progress=None,   # on_progress(done, total, message) per row / bucket
) -> Path:
    """
    Stream a (possibly huge) CSV into a compact **analysis-ready** CSV with a
    stable schema and optional external grouping.

    Output schema
    -------------
    Always writes a header and enforces a consistent column order:

    • No grouping:
        `text_id,text`                            (plus `source_col` if `mode="separate"`)
    • With grouping:
        `text_id,text,group_count`                (plus `source_col` if `mode="separate"`)

    `carry_cols` inserts the named source columns between the identifiers and
    `text` in both shapes.

    Where:
      - `text_id` is either the composed ID from `id_cols` or `row_<n>` when
        `id_cols=None`.
      - `mode="concat"` joins all `text_cols` using `joiner` per row or group.
      - `mode="separate"` emits one row per (`row_or_group`, `text_col`) and
        fills `source_col` with the contributing column name.

    Grouping at scale
    -----------------
    If `group_by` is provided, the function performs a **two-pass external
    grouping** that does not require presorting:
      1) Hash-partition rows to on-disk “bucket” CSVs (bounded writers with LRU).
      2) Aggregate each bucket into final rows (concat or separate mode), writing
         `group_count` to record how many pieces contributed.

    Parameters
    ----------
    csv_path
        Source CSV with at least the columns in `text_cols` (and `group_by` if
        grouping).
    out_csv
        Destination CSV. If `None`, a name is derived from the input and options
        (e.g., `<stem>_grouped_<group_by>.csv` or `<stem>_concat_<cols>.csv`).
    overwrite_existing
        If `False` (default) and `out_csv` exists, the function returns early.
    text_cols
        Text fields to concatenate or emit separately. May be empty or None:
        the spreadsheet is then wrangled without any text -- rows are still
        combined, counted, summarized and carried -- and the output has no
        ``text`` column.
    id_cols
        Optional columns to compose `text_id` when not grouping. When omitted, a
        synthetic `row_<n>` is used.
    carry_cols
        Source columns to copy through to the output untouched. Distinct from
        `id_cols`, which *compose* `text_id` and so cannot be used to carry a
        column whose values repeat -- naming `speaker` there would give every
        utterance by one person the same `text_id`. Downstream analyzers offer a
        `pass_through_cols` argument that is meaningless unless the gather that
        feeds them preserved the columns, which is what this is for.

        When grouping, a carried column is written only where its value is the
        same for every row in the group; where rows disagree the cell is blank,
        because there is no honest single answer and picking the first row's is
        the kind of quiet wrong answer that survives into a published table.

        Names in `carry_cols` that the source does not have are dropped with a
        warning rather than raising: unlike `text_cols`, a carried column is a
        convenience, and half the presets that ask for `speaker` run over inputs
        that never had one.
    agg_cols
        Number columns to summarize per group, e.g. a per-post score becoming
        the average score of everything a user wrote. A sequence of names
        means "the mean of each"; a mapping picks the statistic per column
        from ``mean``, ``sum``, ``min`` and ``max``. Each summary lands in
        its own output column named ``<col>_<stat>``, placed after
        ``group_count``.

        Values that are blank or not parseable as numbers are skipped rather
        than treated as zero -- a missing score is missing, not 0 -- and a
        group with no numeric values at all gets a blank cell, for the same
        NA-is-not-zero reason a disagreeing carried column does.
    row_filters
        Which rows to keep, as ``[[column, operator, value], ...]`` over the
        source spreadsheet's own columns -- ``[["age", ">=", 18]]``,
        ``[["condition", "in", ["A", "B"]]]``. A row has to clear all of
        them. The operators and the comparison are
        :mod:`taters.helpers.row_filter`'s, the same ones the statistics use,
        so a filter means one thing wherever it is applied.

        Only the spreadsheet's own columns, because nothing has been measured
        yet: a word count is not available here and does not belong here
        either. What somebody knows at this point is what they collected --
        a screener somebody failed, a condition they are not analyzing, an
        age below the one they meant to study.

        It matters most when rows are being combined, since a row inside
        somebody else's joined text cannot be taken out again afterwards;
        that is why it is applied before the rows are bucketed rather than
        filtered later.

        Summaries run over **every** row of the group, including rows whose
        text cells are empty -- `group_count` counts only the rows whose
        text was joined, so the two can disagree. To keep that visible, each
        summary is followed by a ``<col>_n`` column giving the number of
        numeric values it actually used.

        Requires `group_by`: without groups there is nothing to summarize
        over, and silently ignoring the ask would hide a real mistake.
        Unlike `carry_cols`, a name the source lacks raises: a summary is an
        explicit computation, not a convenience.
    mode
        `"concat"` (default) or `"separate"`. See schema above.
    group_by
        Optional list of columns to aggregate by; works on unsorted CSVs.
    delimiter, encoding, joiner
        Parsing/formatting options. If `delimiter=None`, sniffs from a sample.
    num_buckets, max_open_bucket_files, tmp_root
        External grouping controls (partition count, LRU limit, temp root).
    include_id_cols
        When not grouping, write the id columns beside ``text_id``. Grouped
        output carries the grouping columns instead, whatever this says.
    on_progress
        Optional ``(done, total, message)`` callback. The gather names each
        phase as it runs -- counting the rows, then copying them (or, when
        grouping, sorting them into groups and then combining each group) --
        because a silent spinner over a large file reads as a hang.

    Returns
    -------
    Path
        Path to the analysis-ready CSV.

    Raises
    ------
    ValueError
        If required columns are missing or `mode` is invalid.

    Examples
    --------
    Concatenate two text fields per row:

    >>> csv_to_analysis_ready_csv(
    ...     csv_path="transcripts.csv",
    ...     text_cols=["prompt","response"],
    ...     id_cols=["speaker"],
    ... )

    Group by speaker and join rows:

    >>> csv_to_analysis_ready_csv(
    ...     csv_path="transcripts.csv",
    ...     text_cols=["text"],
    ...     group_by=["speaker"],
    ... )
    """
    in_path = _ensure_path(csv_path)
    keep = [list(f) for f in (row_filters or [])]

    # sniff out the delimiter if we weren't given one
    if delimiter is None:
        with in_path.open("rb") as fb:
            sample = fb.read(8192)
        delimiter = _detect_delimiter(sample, default=DEFAULT_DELIM)

    # empty text_cols is something people actually ask for, not a mistake: a
    # spreadsheet without text can still be combined and summarized (one row
    # per person, with average scores). the output just has no `text` column.
    text_cols = list(text_cols or [])
    mode = mode.strip().lower()
    if mode not in ("concat", "separate"):
        raise ValueError("mode must be 'concat' or 'separate'")
    if mode == "separate" and not text_cols:
        raise ValueError("mode='separate' splits rows by text column, which needs text_cols")

    # normalize agg_cols to {column: stat}. a bare sequence means "the mean of
    # each", because the average per group is almost always what people want.
    if agg_cols and not group_by:
        raise ValueError("agg_cols requires group_by: there are no groups to summarize over")
    agg: Dict[str, str] = (dict(agg_cols) if isinstance(agg_cols, Mapping)
                           else {c: "mean" for c in (agg_cols or [])})
    bad_stats = {s for s in agg.values() if s not in _AGG_STATS}
    if bad_stats:
        raise ValueError(f"agg_cols statistics must be one of {sorted(_AGG_STATS)}, got {sorted(bad_stats)}")

    include_source_col = (mode == "separate")
    include_source_path = False  # we're doing CSV here; the folder variant uses this

    # we resolve these against the real header before either path runs, so the
    # two paths can't disagree about which columns exist.
    carry: List[str] = list(carry_cols or [])
    if carry:
        with in_path.open("r", newline="", encoding=encoding) as f:
            available = csv.DictReader(f, delimiter=delimiter).fieldnames or []
        absent = [c for c in carry if c not in available]
        if absent:
            if verbose:
                print(f"[text-gather] WARNING: cannot carry columns not in the input: {absent}")
            carry = [c for c in carry if c not in absent]

    # figure out where the output goes (next to the input, unless told otherwise)
    out_path = _ensure_path(out_csv) if out_csv is not None else _default_csv_out_path(
        in_csv=in_path, mode=mode, text_cols=text_cols, group_by=group_by)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    if not overwrite_existing and Path(out_path).is_file():
        if verbose:
            print("File with gathered text already exists; returning existing file.")
        return out_path

    # no grouping? then we can just stream straight through to the output
    if not group_by:
        writer, fh, _ = _open_out_csv(
            out_path,
            include_source_col,
            include_source_path,
            include_group_count=False,
            id_col_names=(list(id_cols) if include_id_cols and id_cols else None),
            group_by_names=None,
            carry_col_names=carry,
            include_text=bool(text_cols),
        )
        carried = _carried_names(carry, ["text_id"] + (
            _emit_names(id_cols) if include_id_cols and id_cols else []))
        announce(on_progress, "counting rows")
        ticker = Ticker(on_progress,
                        count_rows(in_path, on_progress=on_progress,
                                   encoding=encoding))
        try:
            with in_path.open("r", newline="", encoding=encoding) as f:
                rdr = csv.DictReader(f, delimiter=delimiter)
                headers = rdr.fieldnames or []
                missing = [c for c in (id_cols or []) + text_cols if c not in headers]
                if missing:
                    raise ValueError(f"Missing columns: {missing}. Make sure that you try specifying a delimiter manually if you see this error message.")

                for idx, row in enumerate(rdr, start=1):
                    ticker.tick(message="copying rows")
                    # before `text_id` is used but after it is numbered, so a
                    # row left out here does not renumber the rows after it
                    # -- every other table in the run is keyed on that
                    # number, and shifting it would join the wrong rows to
                    # the wrong groups while looking perfectly reasonable.
                    if keep and not keeps_row(row, keep):
                        continue
                    text_id = _compose_id([row.get(c, "") for c in (id_cols or [])]) if id_cols else f"row_{idx}"
                    if mode == "concat":
                        parts = [row.get(c, "") for c in text_cols if row.get(c, "")]
                        # if we have text columns, a row with nothing in any of
                        # them has nothing to give us; if we have none, every
                        # row counts on its own.
                        if text_cols and not parts:
                            continue
                        row_prefix = [text_id]
                        if include_id_cols and id_cols:
                            row_prefix += [row.get(c, "") for c in _emit_names(id_cols)]
                        row_prefix += [row.get(c, "") for c in carried]
                        writer.writerow(row_prefix + ([joiner.join(parts)] if text_cols else []))
                    else:
                        for col in text_cols:
                            val = row.get(col, "")
                            if not val:
                                continue
                            row_prefix = [text_id]
                            if include_id_cols and id_cols:
                                row_prefix += [row.get(c, "") for c in _emit_names(id_cols)]
                            row_prefix += [row.get(c, "") for c in carried]
                            writer.writerow(row_prefix + [val, col])

        except BaseException:
            fh.close()
            _discard_scratch(out_path)
            raise
        else:
            fh.close()
            _promote_scratch(out_path)
        return out_path

    # otherwise we do the grouping on disk, in two passes
    group_by = list(group_by)

    # pass 1: partition the rows into hash buckets
    tmp_base = Path(tempfile.mkdtemp(prefix="gather_partitions_", dir=str(tmp_root) if tmp_root else None))
    part_dir = tmp_base / "parts"
    part_dir.mkdir(parents=True, exist_ok=True)

    # the bucket writer cache
    carried = _carried_names(carry, ["text_id"] + _emit_names(group_by))
    # the columns we're aggregating ride along in the partitions too (deduped:
    # it's fine to summarize a column that's also grouped or carried, and the
    # DictReader in pass 2 needs each name to show up exactly once).
    agg_extra = [c for c in agg if c not in set(group_by) | set(carried) | set(text_cols)]
    header_small = group_by + carried + agg_extra + text_cols
    cache = _LRUFileCache(
        max_open=max_open_bucket_files,
        newline="",
        encoding=encoding,
        delimiter=delimiter,
    )


    announce(on_progress, "counting rows")
    ticker = Ticker(on_progress,
                    count_rows(in_path, on_progress=on_progress,
                               encoding=encoding))
    try:
        with in_path.open("r", newline="", encoding=encoding) as f:
            rdr = csv.DictReader(f, delimiter=delimiter)
            headers = rdr.fieldnames or []
            missing = [c for c in group_by + list(agg) + text_cols if c not in headers]
            if missing:
                raise ValueError(f"Missing columns: {missing}. Make sure that you try specifying a delimiter manually if you see this error message.")

            for row in rdr:
                ticker.tick(message="sorting rows into groups (pass 1 of 2)")
                if keep and not keeps_row(row, keep):
                    # dropped before it is bucketed, on purpose: once a row
                    # is inside somebody else's joined text there is no way
                    # to take it back out again.
                    continue
                key_tuple = tuple(row[g] for g in group_by)
                bucket = _bucket_of_key(key_tuple, num_buckets)
                bpath = part_dir / f"bucket_{bucket:05d}.csv"
                w = cache.get(bucket, bpath, header_small)
                # only the fields we need, so the partitions stay lean
                w.writerow([row.get(c, "") for c in header_small])
    finally:
        cache.close_all()

    # pass 2: aggregate each bucket and hand it to the final writer
    writer, out_fh, _ = _open_out_csv(
        out_path,
        include_source_col,
        include_source_path,
        include_group_count=True,
        id_col_names=None,
        group_by_names=group_by,
        carry_col_names=carried,
        agg_col_names=[n for c, s in agg.items() for n in (f"{c}_{s}", f"{c}_n")],
        include_text=bool(text_cols),
    )

    def _agreed(values: List[Dict[str, str]]) -> List[str]:
        """
        One value per carried column, or blank where the group disagrees.

        See the note in the docstring: a group that spans two speakers has no
        single speaker, and writing the first one seen would be a guess that
        looks exactly like a fact once it is in the output CSV.
        """
        out: List[str] = []
        for name in carried:
            distinct = {v.get(name, "") for v in values}
            out.append(distinct.pop() if len(distinct) == 1 else "")
        return out



    buckets = sorted(part_dir.glob("bucket_*.csv"))
    ticker = Ticker(on_progress, len(buckets))
    try:
        for bfile in buckets:
            ticker.tick(message="combining the groups (pass 2 of 2)")
            # this bucket fits in memory, so we aggregate it there.
            # key -> every row that contributed. we hang onto these for the
            # carried columns (which have to agree across the group) and the
            # aggregated columns (whose stats run over exactly these rows).
            # kept beside the text aggregation rather than folded into it,
            # because `separate` mode emits several rows per key and they all
            # have to agree on the carried values.
            keep_rows = bool(carried) or bool(agg)
            carried_rows: Dict[Tuple[str, ...], List[Dict[str, str]]] = {}
            # aggregation sees every row in the group, textless ones included:
            # a person's average score is the average over what they posted,
            # not over what happened to have text. group_count still counts
            # only the rows whose text got joined, and the per-column n makes
            # the difference visible instead of leaving it for someone to
            # discover in an audit.
            agg_rows: Dict[Tuple[str, ...], List[Dict[str, str]]] = {}
            if mode == "concat":
                # key -> list[text]
                texts: Dict[Tuple[str, ...], List[str]] = {}
                with bfile.open("r", newline="", encoding=encoding) as bf:
                    br = csv.DictReader(bf, delimiter=delimiter)
                    for row in br:
                        key = tuple(row[g] for g in group_by)
                        if agg:
                            agg_rows.setdefault(key, []).append(row)
                        parts = [row.get(c, "") for c in text_cols if row.get(c, "")]
                        if text_cols and not parts:
                            continue
                        texts.setdefault(key, []).append(joiner.join(parts))
                        if keep_rows:
                            carried_rows.setdefault(key, []).append(row)
                # now we write out one row per key
                for key, pieces in texts.items():
                    text_id = _compose_id(key) or "group"
                    kept = [v for n, v in zip(group_by, key, strict=True) if n != "text_id"]
                    extra = _agreed(carried_rows.get(key, [])) if carried else []
                    stats = _summaries(agg, agg_rows.get(key, [])) if agg else []
                    body = [joiner.join(pieces)] if text_cols else []
                    writer.writerow([text_id, *kept, *extra, *body, len(pieces), *stats])
            else:
                # key -> col -> list[text]
                texts: Dict[Tuple[str, ...], Dict[str, List[str]]] = {}
                with bfile.open("r", newline="", encoding=encoding) as bf:
                    br = csv.DictReader(bf, delimiter=delimiter)
                    for row in br:
                        key = tuple(row[g] for g in group_by)
                        if agg:
                            agg_rows.setdefault(key, []).append(row)
                        box = texts.setdefault(key, {})
                        wrote = False
                        for col in text_cols:
                            val = row.get(col, "")
                            if val:
                                box.setdefault(col, []).append(val)
                                wrote = True
                        if keep_rows and wrote:
                            carried_rows.setdefault(key, []).append(row)
                # now we write out one row per key per column
                for key, per_col in texts.items():
                    text_id = _compose_id(key) or "group"
                    extra = _agreed(carried_rows.get(key, [])) if carried else []
                    stats = _summaries(agg, agg_rows.get(key, [])) if agg else []
                    for col in text_cols:
                        vals = per_col.get(col, [])
                        if not vals:
                            continue
                        kept = [v for n, v in zip(group_by, key, strict=True) if n != "text_id"]
                        writer.writerow([text_id, *kept, *extra, joiner.join(vals), len(vals), *stats, col])

    except BaseException:
        out_fh.close()
        _discard_scratch(out_path)
        raise
    else:
        out_fh.close()
        _promote_scratch(out_path)
    finally:
        # lastly, clean up after ourselves
        try:
            for p in part_dir.glob("bucket_*.csv"):
                p.unlink(missing_ok=True)
            part_dir.rmdir()
            tmp_base.rmdir()
        except Exception:
            pass

    return out_path

resolve_analysis_ready ¶

resolve_analysis_ready(
    *,
    csv_path=None,
    txt_dir=None,
    analysis_csv=None,
    gathered_csv=None,
    text_cols=("text",),
    id_cols=None,
    mode="concat",
    group_by=None,
    delimiter=",",
    encoding="utf-8-sig",
    joiner=" ",
    num_buckets=64,
    max_open_bucket_files=64,
    tmp_root=None,
    recursive=True,
    pattern="*.txt",
    id_from="stem",
    include_source_path=False,
    overwrite_existing=False,
    on_progress=None,
    workers=0,
    carry_cols=None,
    verbose=None
)

Accept an analysis-ready table, or gather one -- the analyzers' front door.

Every text analyzer takes its input three ways: a spreadsheet to gather from, a folder of documents to gather from, or a table already gathered. The forty lines that told those apart, announced the gather and called one of the two gatherers with a dozen forwarded settings were copied into eleven modules, and had begun to drift (one forwarded verbose and carry_cols, the others did not). This is that block, once. Settings recording is unaffected: the provenance decorator reads the analyzer's own bound arguments, not this function's.

carry_cols and verbose are forwarded to the gatherers only when given, so the analyzers that never passed them behave exactly as before.

Source code in src\taters\helpers\text_gather.py

def resolve_analysis_ready(*, csv_path=None, txt_dir=None, analysis_csv=None,
                           gathered_csv=None, text_cols=("text",), id_cols=None,
                           mode="concat", group_by=None, delimiter=",",
                           encoding="utf-8-sig", joiner=" ", num_buckets=64,
                           max_open_bucket_files=64, tmp_root=None,
                           recursive=True, pattern="*.txt", id_from="stem",
                           include_source_path=False, overwrite_existing=False,
                           on_progress=None, workers=0, carry_cols=None,
                           verbose=None) -> Path:
    """
    Accept an analysis-ready table, or gather one -- the analyzers' front door.

    Every text analyzer takes its input three ways: a spreadsheet to gather
    from, a folder of documents to gather from, or a table already gathered.
    The forty lines that told those apart, announced the gather and called
    one of the two gatherers with a dozen forwarded settings were copied into
    eleven modules, and had begun to drift (one forwarded ``verbose`` and
    ``carry_cols``, the others did not). This is that block, once. Settings
    recording is unaffected: the provenance decorator reads the analyzer's
    own bound arguments, not this function's.

    ``carry_cols`` and ``verbose`` are forwarded to the gatherers only when
    given, so the analyzers that never passed them behave exactly as before.
    """
    if analysis_csv is not None:
        analysis_ready = Path(analysis_csv)
        if not analysis_ready.exists():
            raise FileNotFoundError(f"analysis_csv not found: {analysis_ready}")
        return analysis_ready
    if (csv_path is None) == (txt_dir is None):
        raise ValueError(
            "Provide exactly one of csv_path or txt_dir (or pass analysis_csv).")

    from .progress import announce

    # gathering reads the whole source and writes the analysis-ready table.
    # on a big spreadsheet that's the longest silent stretch in the run, so
    # we say what's happening before it starts.
    announce(on_progress, "reading the input")
    extra = {}
    if verbose is not None:
        extra["verbose"] = verbose
    if csv_path is not None:
        if carry_cols:
            extra["carry_cols"] = list(carry_cols)
        return Path(csv_to_analysis_ready_csv(
            csv_path=csv_path, out_csv=gathered_csv,
            text_cols=list(text_cols),
            id_cols=list(id_cols) if id_cols else None,
            mode=mode, group_by=list(group_by) if group_by else None,
            delimiter=delimiter, encoding=encoding, joiner=joiner,
            num_buckets=num_buckets,
            max_open_bucket_files=max_open_bucket_files, tmp_root=tmp_root,
            overwrite_existing=overwrite_existing, **extra))
    return Path(txt_folder_to_analysis_ready_csv(
        root_dir=txt_dir, out_csv=gathered_csv, recursive=recursive,
        pattern=pattern, encoding=encoding, id_from=id_from,
        include_source_path=include_source_path,
        overwrite_existing=overwrite_existing, on_progress=on_progress,
        workers=workers, **extra))

txt_folder_to_analysis_ready_csv ¶

txt_folder_to_analysis_ready_csv(
    *,
    root_dir,
    out_csv=None,
    recursive=False,
    pattern=DOCUMENT_PATTERN,
    encoding="utf-8",
    id_from="stem",
    include_source_path=True,
    overwrite_existing=False,
    verbose=True,
    on_progress=None,
    workers=0
)

Stream a folder of documents into an analysis-ready CSV with predictable, reproducible IDs.

Documents are .txt, .docx, .doc and .pdf -- text is extracted per type by :func:taters.helpers.doc_text.read_document_text (no OCR: a PDF without a machine-readable text layer has zero text). A document that cannot be read -- corrupt, password-protected, a legacy .doc with no way to convert it -- is skipped with a warning naming it, never allowed to take the whole gather down.

For each readable file matching pattern, the emitted row contains: - text_id: the basename (stem), full filename, or relative path (see id_from), and - text: the extracted text. - source_path: optional column with path relative to root_dir.

Parameters:

Name	Type	Description	Default
`root_dir`	`PathLike`	Folder containing documents.	required
`out_csv`	`PathLike \| None`	Destination CSV. If `None`, a descriptive default is created next to `root_dir` (e.g., `<folder>_txt_recursive_*.csv`).	`None`
`recursive`	`bool`	Recurse into subfolders. Default: `False`.	`False`
`pattern`	`str`	Glob(s) for matching files; several can be joined with `;`. Default: every document type (`".txt;.docx;.doc;.pdf"`).	`DOCUMENT_PATTERN`
`encoding`	`str`	Decoding for plain-text files. Default: `"utf-8"`.	`'utf-8'`
`workers`	`int`	Parallel reader processes -- PDF and Word parsing is CPU-bound, and a big folder reads several times faster in parallel. `0` (default) means automatic: three-quarters of the logical cores; `1` turns parallelism off. The output file is identical whatever the worker count: files are walked in sorted order and results are written in that same order.	`0`
`id_from`	`str`	How to derive `text_id`: `"stem"` (basename without extension), `"name"` (filename), or `"path"` (relative path).	`'stem'`
`include_source_path`	`bool`	If `True` (default), add a `source_path` column showing the relative path.	`True`
`overwrite_existing`	`bool`	If `False` (default) and `out_csv` exists, returns the existing file.	`False`

Returns:

Type	Description
`Path`	Path to the analysis-ready CSV.

Examples:

>>> txt_folder_to_analysis_ready_csv(root_dir="notes", recursive=True, id_from="path")

Source code in src\taters\helpers\text_gather.py

@records_settings(
    # one text per file, so there's no grain to record here
    binding=("root_dir", "recursive", "pattern", "encoding",
             "id_from", "include_source_path"),
    outputs=("out_csv",),
    redo_without_record=True,
    # this step decides what text EXISTS. its settings describe the
    # dataset, so we record them but keep them out of the chain that
    # downstream steps compare -- a model has to be applicable to a corpus
    # assembled differently, which is the whole point of saving one.
    defines_text=True)
def txt_folder_to_analysis_ready_csv(
    *,
    root_dir: PathLike,
    out_csv: PathLike | None = None,
    recursive: bool = False,
    pattern: str = DOCUMENT_PATTERN,
    encoding: str = "utf-8",
    id_from: str = "stem",            # "stem" | "name" | "path"
    include_source_path: bool = True, # writes 'source_path' column
    overwrite_existing: bool = False, # if the file already exists, let's not overwrite by default
    verbose: bool = True,
    on_progress=None,                 # on_progress(done, total, message) per file
    workers: int = 0,                 # parallel readers; 0 = auto, 1 = off
) -> Path:
    """
    Stream a folder of documents into an analysis-ready CSV with predictable,
    reproducible IDs.

    Documents are ``.txt``, ``.docx``, ``.doc`` and ``.pdf`` -- text is
    extracted per type by :func:`taters.helpers.doc_text.read_document_text`
    (no OCR: a PDF without a machine-readable text layer has zero text). A
    document that cannot be read -- corrupt, password-protected, a legacy
    ``.doc`` with no way to convert it -- is skipped with a warning naming it,
    never allowed to take the whole gather down.

    For each readable file matching `pattern`, the emitted row contains:
      - `text_id`: the basename (stem), full filename, or relative path (see
        `id_from`), and
      - `text`: the extracted text.
      - `source_path`: optional column with path relative to `root_dir`.

    Parameters
    ----------
    root_dir
        Folder containing documents.
    out_csv
        Destination CSV. If `None`, a descriptive default is created next to
        `root_dir` (e.g., `<folder>_txt_recursive_*.csv`).
    recursive
        Recurse into subfolders. Default: `False`.
    pattern
        Glob(s) for matching files; several can be joined with ``;``.
        Default: every document type (``"*.txt;*.docx;*.doc;*.pdf"``).
    encoding
        Decoding for plain-text files. Default: `"utf-8"`.
    workers
        Parallel reader processes -- PDF and Word parsing is CPU-bound, and a
        big folder reads several times faster in parallel. ``0`` (default)
        means automatic: three-quarters of the logical cores; ``1`` turns parallelism off.
        The output file is **identical whatever the worker count**: files are
        walked in sorted order and results are written in that same order.
    id_from
        How to derive `text_id`: `"stem"` (basename without extension),
        `"name"` (filename), or `"path"` (relative path).
    include_source_path
        If `True` (default), add a `source_path` column showing the relative path.
    overwrite_existing
        If `False` (default) and `out_csv` exists, returns the existing file.

    Returns
    -------
    Path
        Path to the analysis-ready CSV.

    Examples
    --------
    >>> txt_folder_to_analysis_ready_csv(root_dir="notes", recursive=True, id_from="path")
    """
    root = _ensure_path(root_dir)
    out_path = _ensure_path(out_csv) if out_csv is not None else _default_txt_out_path(
        root, id_from=id_from, recursive=recursive, pattern=pattern)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    if not overwrite_existing and Path(out_path).is_file():
        if verbose:
            print("File with gathered text already exists; returning existing file.")
        return out_path

    writer, fh, _ = _open_out_csv(out_path, include_source_col=False, include_source_path=include_source_path)
    try:
        # several globs, but one deduplicated, *sorted* walk: with more than one
        # pattern the concatenation order would otherwise be up to the
        # filesystem, and two runs over the same folder need to write the same
        # file.
        found: set = set()
        for one_pattern in pattern.split(";"):
            one_pattern = one_pattern.strip()
            if one_pattern:
                found.update(root.rglob(one_pattern) if recursive
                             else root.glob(one_pattern))
        files = [p for p in sorted(found) if p.is_file()]
        if id_from not in ("stem", "name", "path"):
            raise ValueError("id_from must be 'stem', 'name', or 'path'")

        # the gather gets to be its own visible phase: the step row reads
        # "gathering N documents" and (on displays that can show it) each
        # document being read gets a sub-bar of its own -- same language as
        # ffmpeg's or Whisper's per-file rows, rather than a step label that
        # looks like it's doing somebody else's work.
        from .progress import FlightReporter

        reporter = FlightReporter(on_progress, len(files),
                                  "gathering documents") \
            if on_progress is not None else None
        for p, (status, payload, relayed) in _pair_paths_with_results(
                files, workers, encoding, reporter):
            if reporter is not None:
                reporter.consumed()
            for message in relayed:
                # warnings raised inside a worker process would otherwise just
                # vanish; they ride back with the result and we re-issue them here.
                warnings.warn(message, stacklevel=2)
            if status == "error":
                # a broken document -- an image wearing a .docx name, a
                # truncated PDF, an unreadable file -- only costs itself: we
                # treat it as having no text, name it in a warning, and never
                # let it take the whole run down.
                warnings.warn(f"{payload} Skipped.", stacklevel=2)
                continue
            text = payload
            if id_from == "stem":
                text_id = p.stem
            elif id_from == "name":
                text_id = p.name
            else:
                text_id = str(p.relative_to(root))
            if len(text) > 500_000:
                # a whole proceedings volume or book as one "document".
                # perfectly legal, but dictionary-style scoring grows
                # super-linearly with length, so one of these can take minutes
                # while its neighbors take milliseconds -- better to say so
                # now, and name it, than to look hung later.
                warnings.warn(
                    f"'{p.name}' extracted {len(text):,} characters -- a very "
                    "large document; some analyses can be slow on it.",
                    stacklevel=2)
            if not text.strip() and p.suffix.lower() != ".txt":
                # a scanned PDF, an image-only Word file: zero text is the
                # documented answer, but if we stayed quiet it'd look like we
                # lost their data.
                warnings.warn(
                    f"'{p.name}' contains no machine-readable text (no OCR "
                    "is attempted); it contributes nothing.", stacklevel=2)
                continue
            if include_source_path:
                writer.writerow([text_id, text, str(p.relative_to(root))])
            else:
                writer.writerow([text_id, text])
    except BaseException:
        fh.close()
        _discard_scratch(out_path)
        raise
    else:
        fh.close()
        _promote_scratch(out_path)

    return out_path

taters.helpers.library ¶

The library: user-imported assets that outlive any one project.

A "dictionary" here is the general term -- a content-coding dictionary and an archetype dictionary are different kinds, with different formats, consumed by different modules. The library stores each kind in its own folder under the user's home, so importing a dictionary once makes it available to every pipeline on the machine, wherever it is run from. That replaces the old convention of a dictionaries/ folder relative to the working directory, which silently failed for anyone who did not happen to have one.

No UI imports (the :mod:taters.helpers.gpu pattern): the wizard is one consumer, but the runner or a future GUI can read the same library. All the operations are deliberately file-level and boring -- an entry is its file, its name is the filename stem -- so a user can also just manage the folder by hand and nothing here will be surprised.

Adding a new kind -- pretrained classifier models are the expected next one -- is a :data:KINDS entry plus a library= line on the recipe that consumes it; the manager, the picker, and the empty-library contingency in the wizard all key off those two declarations.

LibraryCollision ¶

LibraryCollision(existing)

Bases: Exception

An import would overwrite an entry that already exists.

Raised instead of overwriting so the UI can ask replace-or-rename; the existing path rides along as .existing.

Source code in src\taters\helpers\library.py

def __init__(self, existing: Path):
    self.existing = existing
    super().__init__(f"'{existing.stem}' is already in the library")

LibraryKind `dataclass` ¶

LibraryKind(
    id,
    label,
    help,
    suffixes,
    deep_check=None,
    describe_entry=None,
)

One category of importable asset.

Attributes:

Name	Type	Description
`id`	`str`	Stable identifier; also the folder name under the library.
`label, help`	`str`	What menus call it, and one sentence including the accepted formats.
`suffixes`	`tuple of str`	File extensions this kind accepts, lowercase with the dot.

asset_problem ¶

asset_problem(path, kind=None)

Why this file cannot work as a library asset, or "" when it can.

The canonical check, shared by import (refuse the file while the user is holding it and can act) and by the analyzers (a file can still arrive by CLI path without ever passing through the library). Two layers:

cheap shape heuristics, always: an empty file is not a dictionary, and a .dic with no %...% category header is a bare word list -- the mistake that made contentcoder die with a bare "list index out of range" mid-run, an hour after it was made;
with a kind, that kind's :attr:~LibraryKind.deep_check -- the real parser. Import passes the kind; the analyzers do not, because they construct the real parser on the very next line and asking it twice buys nothing.

Source code in src\taters\helpers\library.py

def asset_problem(path: Path, kind: Optional[LibraryKind] = None) -> str:
    """
    Why this file cannot work as a library asset, or "" when it can.

    The canonical check, shared by import (refuse the file while the user is
    holding it and can act) and by the analyzers (a file can still arrive by
    CLI path without ever passing through the library). Two layers:

    * cheap shape heuristics, always: an empty file is not a dictionary, and a
      ``.dic`` with no ``%...%`` category header is a bare word list -- the
      mistake that made contentcoder die with a bare "list index out of range"
      mid-run, an hour after it was made;
    * with a ``kind``, that kind's :attr:`~LibraryKind.deep_check` -- the real
      parser. Import passes the kind; the analyzers do not, because they
      construct the real parser on the very next line and asking it twice
      buys nothing.
    """
    path = Path(path)
    try:
        text = path.read_text(encoding="utf-8-sig", errors="ignore")
    except OSError as e:
        return f"'{path.name}' could not be read: {e}"
    if not text.strip():
        return f"'{path.name}' is empty -- not a dictionary."
    if path.suffix.lower() == ".dic" and text.count("%") < 2:
        return (
            f"'{path.name}' has no %...% category header, so it is a plain "
            "word list rather than a LIWC .dic. Add the header block "
            "(%, one 'number<TAB>name' per category, %) first."
        )
    if kind is not None and kind.deep_check is not None:
        return kind.deep_check(path)
    return ""

delete ¶

delete(kind, name)

Remove an entry and the weights it carries. Permanent, so the UI confirms before calling this.

Source code in src\taters\helpers\library.py

def delete(kind: LibraryKind, name: str) -> None:
    """Remove an entry and the weights it carries. Permanent, so the UI
    confirms before calling this."""
    entry = _entry(kind, name)
    payloads = payload_of(entry)
    entry.unlink()
    for payload in payloads:
        _remove_payload(payload)

display_name ¶

display_name(path)

How an entry is named in menus: the filename stem.

Source code in src\taters\helpers\library.py

def display_name(path: Path) -> str:
    """How an entry is named in menus: the filename stem."""
    return path.stem

entries ¶

entries(kind)

Every entry of this kind, sorted by name. Only the kind's own formats count -- a stray .txt dropped into the folder by hand is ignored, not an error.

Source code in src\taters\helpers\library.py

def entries(kind: LibraryKind) -> List[Path]:
    """Every entry of this kind, sorted by name. Only the kind's own formats
    count -- a stray .txt dropped into the folder by hand is ignored, not an
    error."""
    folder = kind_dir(kind)
    carried = payload_dirs(folder)
    return sorted(
        (f for f in folder.iterdir()
         if f.is_file() and f.suffix.lower() in kind.suffixes
         and f not in carried),
        key=lambda f: f.name.lower(),
    )

expand ¶

expand(kind, values)

Resolve a mixed list of files and folders to this kind's files.

A folder means "everything of this kind inside it, recursively" -- the reading the analyzers give a folder path, and what the wizard's default writes into a preset (the kind's whole library folder). Any screen seeded through this shows exactly what a run would use; seeding from the raw values made the picker intersect a folder path against entry file paths and open with everything unticked while the settings row said "all 10".

Source code in src\taters\helpers\library.py

def expand(kind: LibraryKind, values) -> List[Path]:
    """
    Resolve a mixed list of files and folders to this kind's files.

    A folder means "everything of this kind inside it, recursively" -- the
    reading the analyzers give a folder path, and what the wizard's default
    writes into a preset (the kind's whole library folder). Any screen seeded
    through this shows exactly what a run would use; seeding from the raw
    values made the picker intersect a folder path against entry file paths
    and open with everything unticked while the settings row said "all 10".
    """
    out: List[Path] = []
    for v in values:
        p = Path(v)
        if p.is_dir():
            out.extend(f for f in _files_under(p)
                       if f.suffix.lower() in kind.suffixes)
        else:
            out.append(p)
    return out

export_to ¶

export_to(kind, name, dest_dir, *, replace=False)

Copy an entry out, keeping its filename. Returns the new path.

Raises:

Type	Description
`LibraryCollision`	When the destination file already exists and `replace` is False -- the same contract as :func:`import_into`, for the same reason: a copy that silently overwrites is a delete nobody asked for.

Source code in src\taters\helpers\library.py

def export_to(kind: LibraryKind, name: str, dest_dir: Path,
              *, replace: bool = False) -> Path:
    """
    Copy an entry out, keeping its filename. Returns the new path.

    Raises
    ------
    LibraryCollision
        When the destination file already exists and ``replace`` is False --
        the same contract as :func:`import_into`, for the same reason: a copy
        that silently overwrites is a delete nobody asked for.
    """
    src = _entry(kind, name)
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    dest = dest_dir / src.name
    pairs = _payload_destinations(src, dest)
    if not replace:
        for taken in [dest] + [d for _s, d in pairs]:
            if taken.exists():
                raise LibraryCollision(dest)
    shutil.copy2(src, dest)
    for payload, target in pairs:
        _copy_payload(payload, target)
    return dest

find_payload ¶

find_payload(kind, name, digests)

A payload in the library whose files match the digests a manifest records.

A model carried into another run travels as its manifest text only (see pipelines.run_pipeline._materialize_assets); the weights stay in the library. The manifest records payload_digests -- sha256 per file -- so the loader can find them here and prove they are the same bytes. The match is by content, not by name: a renamed library entry still counts. name is the payload's suffix-bearing name, used to skip payloads of the wrong shape quickly.

Source code in src\taters\helpers\library.py

def find_payload(kind: LibraryKind, name: str, digests: Mapping[str, str]) -> Optional[Path]:
    """
    A payload in the library whose files match the digests a manifest records.

    A model carried into another run travels as its manifest text only (see
    ``pipelines.run_pipeline._materialize_assets``); the weights stay in the
    library. The manifest records ``payload_digests`` -- sha256 per file --
    so the loader can find them here and prove they are the same bytes. The
    match is by content, not by name: a renamed library entry still counts.
    ``name`` is the payload's suffix-bearing name, used to skip payloads of
    the wrong shape quickly.
    """
    wanted = {str(k): str(v) for k, v in (digests or {}).items()}
    if not wanted:
        return None
    suffix = Path(name).suffix
    for manifest in entries(kind):
        for payload in payload_of(manifest):
            if suffix and payload.suffix != suffix:
                continue
            if not payload.exists():
                continue
            if _payload_matches(payload, wanted):
                return payload
    return None

import_into ¶

import_into(kind, src, *, replace=False)

Copy a file into the library.

Raises:

Type	Description
`ValueError`	For a format the kind does not accept -- with the formats it does, because "wrong extension" without the right ones is a dead end -- or for a file the kind's own parser cannot load (see :func:`asset_problem`), with the parser's reason.
`LibraryCollision`	When an entry with this name exists and `replace` is False. The UI turns this into a replace-or-keep question rather than deciding here.

Source code in src\taters\helpers\library.py

def import_into(kind: LibraryKind, src: Path, *, replace: bool = False) -> Path:
    """
    Copy a file into the library.

    Raises
    ------
    ValueError
        For a format the kind does not accept -- with the formats it does,
        because "wrong extension" without the right ones is a dead end -- or
        for a file the kind's own parser cannot load (see
        :func:`asset_problem`), with the parser's reason.
    LibraryCollision
        When an entry with this name exists and ``replace`` is False. The UI
        turns this into a replace-or-keep question rather than deciding here.
    """
    src = Path(src)
    if src.suffix.lower() not in kind.suffixes:
        raise ValueError(
            f"{src.name} is not a {kind.label.lower()} file "
            f"(accepted: {', '.join(kind.suffixes)})"
        )
    problem = asset_problem(src, kind)
    if problem:
        # we refuse right here, while the user still has the file in hand and
        # can fix it -- if we let it in, it blows up a run an hour later instead.
        raise ValueError(problem)
    missing = [p.name for p in payload_of(src) if not p.exists()]
    if missing:
        raise ValueError(
            f"{src.name} names weights that are not beside it "
            f"({', '.join(missing)}). Copy the model file and its weights "
            f"together, or re-run the step that made them.")
    dest = kind_dir(kind) / src.name
    pairs = _payload_destinations(src, dest)
    if not replace:
        for taken in [dest] + [d for _s, d in pairs]:
            if taken.exists():
                raise LibraryCollision(dest)
    shutil.copy2(src, dest)
    try:
        for payload, target in pairs:
            _copy_payload(payload, target)
        _rewrite_payload_names(dest, {s.name: d.name for s, d in pairs
                                      if s.name != d.name})
    except BaseException:
        # half an entry is worse than none: a manifest with no weights would get
        # listed, picked, and then refused at run time.
        dest.unlink(missing_ok=True)
        for _payload, target in pairs:
            _remove_payload(target)
        raise
    return dest

kind_by_id ¶

kind_by_id(kind_id)

One kind, or a KeyError that names the valid ids (it is always a typo).

Source code in src\taters\helpers\library.py

def kind_by_id(kind_id: str) -> LibraryKind:
    """One kind, or a KeyError that names the valid ids (it is always a typo)."""
    try:
        return KINDS[kind_id]
    except KeyError:
        raise KeyError(
            f"unknown library kind {kind_id!r}. Known: {', '.join(sorted(KINDS))}"
        ) from None

kind_dir ¶

kind_dir(kind)

This kind's folder, created and kept level with what the package ships.

Seeding used to happen only when the folder did not exist, which meant a release that added a dictionary reached new users and nobody else. So it now reconciles file by file against a ledger of what we have installed before (see :func:_seed), and an upgrade brings its new and corrected built-ins to a library that has been in use for years.

What it will never do is undo a decision: a built-in the user deleted stays deleted, and a copy they edited stays edited. An empty library is a valid state the UI explains, not one this quietly "repairs."

Source code in src\taters\helpers\library.py

def kind_dir(kind: LibraryKind) -> Path:
    """
    This kind's folder, created and kept level with what the package ships.

    Seeding used to happen only when the folder did not exist, which meant a
    release that added a dictionary reached new users and nobody else. So it
    now reconciles file by file against a ledger of what we have installed
    before (see :func:`_seed`), and an upgrade brings its new and corrected
    built-ins to a library that has been in use for years.

    What it will never do is undo a decision: a built-in the user deleted
    stays deleted, and a copy they edited stays edited. An empty library is a
    valid state the UI explains, not one this quietly "repairs."
    """
    target = library_home() / kind.id
    target.mkdir(parents=True, exist_ok=True)
    _seed(kind, target)
    return target

library_home ¶

library_home()

Where the library lives: $TATERS_HOME/library, or ~/.taters/library.

The environment override exists so tests -- and anyone who wants their library on a different drive -- can move the whole thing without patching.

Source code in src\taters\helpers\library.py

def library_home() -> Path:
    """
    Where the library lives: ``$TATERS_HOME/library``, or ``~/.taters/library``.

    The environment override exists so tests -- and anyone who wants their
    library on a different drive -- can move the whole thing without patching.
    """
    base = os.environ.get("TATERS_HOME")
    return (Path(base) if base else Path.home() / ".taters") / "library"

model_files ¶

model_files(folder)

The .json manifests under a folder, without the ones inside a payload.

What a step handed "a folder" resolves to, and what the finish screen scans a whole run folder with. Only .json files are walked -- a run over thousands of recordings holds tens of thousands of WAVs and CSVs, and listing every one of them to find a handful of manifests made the finish screen wait on a network drive. A payload declared by any manifest found is pruned, wherever in the tree it sits.

Source code in src\taters\helpers\library.py

def model_files(folder: Path) -> List[Path]:
    """
    The ``.json`` manifests under a folder, without the ones inside a payload.

    What a step handed "a folder" resolves to, and what the finish screen
    scans a whole run folder with. Only ``.json`` files are walked -- a run
    over thousands of recordings holds tens of thousands of WAVs and CSVs,
    and listing every one of them to find a handful of manifests made the
    finish screen wait on a network drive. A payload declared by any
    manifest found is pruned, wherever in the tree it sits.
    """
    folder = Path(folder)
    if not folder.is_dir():
        return []
    candidates = sorted(f for f in folder.rglob("*.json") if f.is_file())
    carried: Set[Path] = set()
    for f in candidates:
        carried.update(payload_of(f))
    return [f for f in candidates
            if f not in carried and not any(c in f.parents for c in carried)]

payload_dirs ¶

payload_dirs(folder)

Every payload path declared by the manifests directly in folder.

Source code in src\taters\helpers\library.py

def payload_dirs(folder: Path) -> Set[Path]:
    """Every payload path declared by the manifests directly in ``folder``."""
    folder = Path(folder)
    if not folder.is_dir():
        return set()
    carried: Set[Path] = set()
    for manifest in folder.glob("*.json"):
        carried.update(payload_of(manifest))
    return carried

payload_of ¶

payload_of(entry)

The sibling files and folders a .json entry declares as its payload.

Empty for anything that is not a JSON manifest with a payload list. Names are taken as siblings only: a name with a path separator, or a dot entry, is ignored rather than allowed to point outside the folder.

Source code in src\taters\helpers\library.py

def payload_of(entry: Path) -> List[Path]:
    """
    The sibling files and folders a ``.json`` entry declares as its payload.

    Empty for anything that is not a JSON manifest with a ``payload`` list.
    Names are taken as siblings only: a name with a path separator, or a
    dot entry, is ignored rather than allowed to point outside the folder.
    """
    entry = Path(entry)
    if entry.suffix.lower() != ".json" or not entry.is_file():
        return []
    try:
        doc = json.loads(entry.read_text(encoding="utf-8"))
    except (OSError, ValueError, UnicodeDecodeError):
        return []
    names = doc.get("payload") if isinstance(doc, dict) else None
    if not isinstance(names, list):
        return []
    out: List[Path] = []
    for name in names:
        text = str(name)
        if not text or "/" in text or "\\" in text or text in (".", ".."):
            continue
        out.append(entry.with_name(text))
    return out

payload_size ¶

payload_size(entry)

Bytes of an entry and everything it carries.

Source code in src\taters\helpers\library.py

def payload_size(entry: Path) -> int:
    """Bytes of an entry and everything it carries."""
    total = 0
    for p in [Path(entry)] + payload_of(entry):
        try:
            if p.is_dir():
                total += sum(f.stat().st_size for f in p.rglob("*") if f.is_file())
            else:
                total += p.stat().st_size
        except OSError:
            continue
    return total

rename ¶

rename(kind, old, new)

Rename an entry, keeping its suffix.

The suffix carries the format, which renaming must not be able to lie about -- so a new name arriving with an extension has it stripped rather than honored.

Source code in src\taters\helpers\library.py

def rename(kind: LibraryKind, old: str, new: str) -> Path:
    """
    Rename an entry, keeping its suffix.

    The suffix carries the *format*, which renaming must not be able to lie
    about -- so a new name arriving with an extension has it stripped rather
    than honored.
    """
    src = _entry(kind, old)
    if "/" in new or "\\" in new:
        # Path(new).stem quietly keeps only what follows the last separator:
        # "a/b" became "b", and a name ending "[/]" became "]". we found this
        # one by fuzzing; better to refuse than to mangle.
        raise ValueError(
            f"a name cannot contain '/' or '\\' (got {new!r})"
        )
    stem = Path(new).stem.strip()
    if not stem:
        raise ValueError("the new name is empty")
    dest = src.with_name(stem + src.suffix)
    pairs = _payload_destinations(src, dest)
    if dest != src:
        for taken in [dest] + [d for _s, d in pairs if d != _s]:
            if taken.exists():
                raise LibraryCollision(dest)
    src.rename(dest)
    for payload, target in pairs:
        if target != payload and payload.exists():
            payload.rename(target)
    _rewrite_payload_names(dest, {s.name: d.name for s, d in pairs
                                  if s.name != d.name})
    return dest

taters.helpers.model_spec ¶

One description of a saved model, whatever kind it is.

Taters writes several kinds of reusable instrument -- a MEM topic model, a ridge regression, a classifier -- and they were each self-describing to their own scoring function and to nothing else. That was fine while each kind had its own menu entry, and stops being fine the moment a screen wants to offer models as a category: to list them it has to say what each one is, to score with one it has to know whether the thing needs text or features, and to write the results it has to know what columns will appear.

So each kind registers those three answers here, and every screen asks this module instead of sniffing JSON keys for itself. The alternative -- a per-kind branch at each call site -- is the shape where adding a fourth kind means finding all of them.

Adding a kind of model

Everything a screen or a scoring run needs to know about a model comes from this registry, so a new kind -- a fine-tuned transformer, a gradient-boosted tree, whatever comes next -- is added in one place:

Give it a kind tag in the file it writes ("taters-<x>-model") and a format integer.
Add a :class:ModelType to :data:MODEL_TYPES stating:
needs -- "text" if it reads text and re-derives its own instrument (the MEM pattern), "features" if it reads feature columns someone else measured. That one word decides whether the settings-provenance gate applies to it, and nothing else has to know.
read -- doc -> (outputs, columns, inputs), so listings can say what it is and what it will add.
score -- "module:function" for applying it. Omit this and scoring refuses by name instead of guessing.
a "payload" list in the model file -- the sibling weights the library moves with it (see helpers.library.payload_of)
write_outputs -- if its output columns can be renamed at import.
classes -- doc -> {outcome: [class, ...]} when it predicts categories, so a class can be relabeled per model ("0" -> "control") and every applier writes the label through :func:class_label.
apply_settings -- the defaults that govern how it is applied to new data (batch size, weighting, whether to emit probabilities). They live in the model file's apply block, editable per model in Settings, and an applier reads them through :func:apply_defaults so a pipeline gets the model's own settings unless a call overrides them.
Add a _load_model-style gate in the module that owns it, and register it in :func:taters.helpers.library._model_problem so the import screen is exactly as strict as run time.

Nothing else needs editing. In particular the scoring entry point, the library folder, the naming flow, the wizard row and the provenance gate are all written against this registry rather than against the three kinds that happen to exist today.

Naming

A model file is named by whoever exported it, which is nearly always the step's own output name (ridge__all__age.json), and that says nothing about what it predicts. So a model carries a name the researcher chose, and output names for the columns it will write: a ridge fitted to predict age on a blog corpus should be able to land in a new dataset as pred_age_blogs, not as pred_age colliding with the age already there. Both are stored in the model file, so they travel with it.

ApplySetting `dataclass` ¶

ApplySetting(
    default, help, kind="str", choices=None, validate=None
)

One default that governs how a kind of model is applied to new data.

Stored per model in the file's apply block and read back through :func:apply_defaults, so a model carries its own settings wherever it goes -- a word-vector model that should weight by types, a predictor that fits a small card at batch 8 -- rather than every pipeline having to know. kind says how a typed-in value is read: "int", "float", "bool", "str", or "text" for free text that is not one of a fixed set. choices restricts a str to a list.

coerce ¶

coerce(value, name)

The value as this setting stores it, or a refusal naming both.

Source code in src\taters\helpers\model_spec.py

def coerce(self, value, name: str):
    """The value as this setting stores it, or a refusal naming both."""
    out = self._coerce(value, name)
    if self.validate is not None:
        self.validate(out)
    return out

FeaturePlan `dataclass` ¶

FeaturePlan(
    model,
    slug,
    tables=(),
    problems=(),
    path="",
    controls=(),
)

Everything a run needs in order to score with one model, as plain data.

Deliberately free of paths, recipes and pipeline templates: it is built from the model file alone, so the composer stays pure and a test can hand a synthetic plan straight in.

problems is the honest half. A plan with problems cannot be replayed, and each entry is a sentence a researcher can act on -- which is better than a partial replay, because a partial replay is a wrong answer.

ModelInfo `dataclass` ¶

ModelInfo(
    path,
    type_id,
    type_label,
    name,
    outputs,
    columns,
    inputs,
    needs,
    needs_tables=(),
    provenance=dict(),
    library_kind=LIBRARY_KIND,
    bulk_outputs=False,
    controls=(),
    zero_when_absent=(),
    classes=dict(),
    class_names=dict(),
    apply=dict(),
    modality="text",
)

What one saved model is, in the terms a screen needs.

Attributes:

Name	Type	Description
`path`	`Path`	The file itself.
`type_id`	`str`	Short kind key: `"ridge"`, `"classifier"`, `"mem"`.
`type_label`	`str`	What to show a person: `"ridge"`, `"MEM topic model"`. Menus put it in brackets after the name, so a library listing reads `age_blogs [ridge]` and the two are told apart at a glance.
`name`	`str`	The researcher's name for the model, defaulting to the file stem.
`outputs`	`tuple of str`	The output labels, after any renaming -- one per predicted outcome, or one per theme.
`columns`	`tuple of str`	The column names those outputs will actually produce, which is not the same list: a classifier writes a predicted class and a probability per class for one outcome.
`inputs`	`tuple of str`	Feature columns the model needs by name. Empty when the model works from text.
`needs`	`str`	`"features"` or `"text"` -- what has to exist upstream before this model can score anything.
`needs_tables`	`tuple of str`	Which feature tables, named by the step that writes them. A model records its predictors by name, which is enough to score a table that already has them and no help at all in getting there: someone who fitted on cohesion features and came back a week later had 165 column names and nothing saying a cohesion step was needed.
`library_kind`	`str`	The :data:`~taters.helpers.library.KINDS` id this model imports as.
`bulk_outputs`	`bool`	True when the outputs are a numbered family (MEM themes) rather than a handful of named quantities, so renaming them means choosing one prefix rather than editing a hundred labels.
`n_outputs`	`int`	How many outputs there are, for a listing that does not want to print a hundred theme names.

display ¶

display()

age_blogs [ridge] -- the one-line form every menu uses.

Source code in src\taters\helpers\model_spec.py

def display(self) -> str:
    """``age_blogs [ridge]`` -- the one-line form every menu uses."""
    return f"{self.name} [{self.type_label}]"

ModelType `dataclass` ¶

ModelType(
    id,
    label,
    kind_tag,
    needs,
    read,
    score=None,
    write_outputs=None,
    bulk_outputs=False,
    classes=None,
    apply_settings=dict(),
    modality="text",
)

One kind of model, and how to read and rewrite its names.

TablePlan `dataclass` ¶

TablePlan(
    stem,
    target,
    instrument,
    assets,
    digest,
    grain,
    replay=None,
)

One feature table a model needs, and exactly how to measure it.

Attributes:

Name	Type	Description
`stem`	`str`	The table's name at fit time. Load-bearing: the stem is what fixed the model's predictor names, so a replayed table has to be written under it or the names stop matching.
`target`	`str`	`"module:function"` of the analyzer, as recorded. Byte-identical to the producing recipe's `target`, which is how a composer joins the two without knowing anything about models.
`instrument`	`dict`	The measuring settings, as literal values. Complete, because they were recorded after defaults were applied -- which is the whole reason a replay is constructible rather than a guess.
`assets`	`dict`	`{parameter: [{name, sha256, text}]}` for the word lists it used, carried inside the model so the replay works on a machine that never had them.
`digest`	`str`	The instrument digest. Belongs in the private output directory: two models needing identical settings then share one extraction for free, and a re-fit under the same model name cannot short-circuit onto the previous private table.
`grain`	`dict`	What one row was at fit time. Informational only -- never compared.
`replay`	`dict or None`	`{"target": "module:function", "assets": {parameter: [...]}}` when the table cannot be measured the same way twice and must instead be produced by applying what the fit-time step fitted -- a topic model's saved themes. The assets carry the fitted file's text, the same way `assets` carries word lists. `None` for a table whose settings alone reproduce it.

UnknownModel ¶

UnknownModel(message, kind=None)

Bases: Exception

The file is not a saved Taters model, or is a kind this build cannot describe. Carries the kind tag it did have, when it had one.

Source code in src\taters\helpers\model_spec.py

def __init__(self, message: str, kind: Optional[str] = None):
    self.kind = kind
    super().__init__(message)

apply_defaults ¶

apply_defaults(doc)

How a model wants to be applied: registry defaults under its own.

Only the settings its kind registers are read, each coerced to its type; a key the file carries from a newer Taters is ignored rather than passed on. An applier takes apply_defaults(doc)[key] wherever its caller left the argument as None, so the model's own settings win over the function's signature but never over an explicit call.

Source code in src\taters\helpers\model_spec.py

def apply_defaults(doc: dict) -> Dict[str, object]:
    """
    How a model wants to be applied: registry defaults under its own.

    Only the settings its kind registers are read, each coerced to its
    type; a key the file carries from a newer Taters is ignored rather than
    passed on. An applier takes ``apply_defaults(doc)[key]`` wherever its
    caller left the argument as None, so the model's own settings win over
    the function's signature but never over an explicit call.
    """
    spec = MODEL_TYPES.get(str(doc.get("kind")))
    if spec is None or not spec.apply_settings:
        return {}
    stored = doc.get("apply") or {}
    out: Dict[str, object] = {}
    for key, setting in spec.apply_settings.items():
        value = setting.default
        if key in stored:
            try:
                value = setting.coerce(stored[key], key)
            except ValueError:
                value = setting.default    # a damaged value shouldn't block scoring
        out[key] = value
    return out

class_label ¶

class_label(doc, outcome, cls)

The label written for one predicted class: its relabel, or itself.

A model fitted on a condition column coded 0/1 predicts "0" and "1" -- correct, and unreadable in a results table a month later. The relabel lives in the model file (class_names), keyed by the class as fitted, and every applier writes predictions and the p_<outcome>_<class> columns through this one function, so a model relabeled in Settings is relabeled however it is invoked.

Source code in src\taters\helpers\model_spec.py

def class_label(doc: dict, outcome: str, cls) -> str:
    """
    The label written for one predicted class: its relabel, or itself.

    A model fitted on a ``condition`` column coded 0/1 predicts ``"0"`` and
    ``"1"`` -- correct, and unreadable in a results table a month later.
    The relabel lives in the model file (``class_names``), keyed by the
    class as fitted, and every applier writes predictions and the
    ``p_<outcome>_<class>`` columns through this one function, so a model
    relabeled in Settings is relabeled however it is invoked.
    """
    names = (doc.get("class_names") or {}).get(outcome) or {}
    label = names.get(str(cls))
    return str(label) if label else str(cls)

describe ¶

describe(model_json)

Describe one saved model, or say why it cannot be described.

Deliberately cheap and structural: it reads the file's own declaration of what it is and does not vet the matrices. Vetting is the scoring loader's job (and the library import gate delegates to it), because a listing that had to fully validate every model would be slow and would hide a damaged model behind a blank row instead of naming it.

Source code in src\taters\helpers\model_spec.py

def describe(model_json: PathLike) -> ModelInfo:
    """
    Describe one saved model, or say why it cannot be described.

    Deliberately cheap and structural: it reads the file's own declaration
    of what it is and does not vet the matrices. Vetting is the scoring
    loader's job (and the library import gate delegates to it), because a
    listing that had to fully validate every model would be slow and would
    hide a damaged model behind a blank row instead of naming it.
    """
    path = Path(model_json)
    doc = _read_doc(path)
    tag = doc.get("kind")
    spec = MODEL_TYPES.get(tag) if isinstance(tag, str) else None
    if spec is None:
        raise UnknownModel(
            f"{path.name} is "
            + (f"a {tag!r} file, which is not a model this build can score"
               if tag else "not a Taters model file at all")
            + ". Models are written by the topic-model, prediction and "
              "classification steps.",
            kind=tag if isinstance(tag, str) else None)
    outputs, columns, inputs = spec.read(doc)
    fitted = dict(spec.classes(doc)) if spec.classes is not None else {}
    relabels = doc.get("class_names") or {}
    return ModelInfo(
        classes={o: tuple(class_label(doc, o, c) for c in cs)
                 for o, cs in fitted.items()},
        class_names={o: {str(k): str(v) for k, v in (m or {}).items()}
                     for o, m in relabels.items() if o in fitted},
        apply=apply_defaults(doc),
        path=path, type_id=spec.id, type_label=spec.label,
        name=str(doc.get("name") or path.stem),
        outputs=tuple(outputs), columns=tuple(columns), inputs=tuple(inputs),
        needs=spec.needs, bulk_outputs=spec.bulk_outputs, modality=spec.modality,
        provenance=((doc.get("needs") or {}).get("feature_provenance") or {}),
        controls=tuple(doc.get("controls") or ()),
        zero_when_absent=tuple(str(c) for c in
                               (doc.get("zero_when_absent") or ())),
        needs_tables=tuple(
            str(s) for s in ((doc.get("needs") or {}).get("feature_tables")
                             or ())))

describe_all ¶

describe_all(paths)

Describe every model that can be described, skipping the rest silently.

Used by listings, where one unreadable file in a library folder must not take the whole menu down -- the import gate already refused anything broken, so a file that fails here arrived some other way.

Source code in src\taters\helpers\model_spec.py

def describe_all(paths: Sequence[PathLike]) -> List[ModelInfo]:
    """
    Describe every model that can be described, skipping the rest silently.

    Used by listings, where one unreadable file in a library folder must not
    take the whole menu down -- the import gate already refused anything
    broken, so a file that fails here arrived some other way.
    """
    out = []
    for path in paths:
        try:
            out.append(describe(path))
        except (UnknownModel, FileNotFoundError, OSError):
            continue
    return out

describe_encoder ¶

describe_encoder(model_json)

One encoder's row: its name and where it came from, then what training did.

Source code in src\taters\helpers\model_spec.py

def describe_encoder(model_json: PathLike) -> Tuple[str, str]:
    """One encoder's row: its name and where it came from, then what training did."""
    path = Path(model_json)
    doc = _read_doc(path)
    name = str(doc.get("name") or path.stem)
    # an encoder trained from scratch has no base model, and "[?]" would read
    # as a broken file rather than as the deliberate absence it is
    if doc.get("trained_from") == "scratch":
        base = "trained from scratch"
    else:
        base = str(doc.get("base_model") or "?")
    bits = []
    layers = doc.get("num_hidden_layers")
    if layers:
        bits.append(f"{layers} layers")
    ev = doc.get("evaluation") or {}
    before, after = ev.get("perplexity_before"), ev.get("perplexity_after")
    if before is not None and after is not None:
        bits.append(f"perplexity {float(before):.1f} → {float(after):.1f}")
    elif ev.get("perplexity_final") is not None:
        bits.append(f"perplexity {float(ev['perplexity_final']):.1f}")
    return f"{name} [{base}]", " · ".join(bits)

edit_model ¶

edit_model(
    model_json,
    *,
    name=None,
    outputs=None,
    prefix=None,
    class_names=None,
    apply=None
)

Change what a model is called, what it writes, and how it is applied.

Parameters:

Name	Type	Description	Default
`model_json`	`PathLike`	The model file, rewritten in place (atomically).	required
`name`	`Optional[str]`	The researcher's name for the model. Shown in every menu.	`None`
`outputs`	`Optional[Sequence[str]]`	One new label per output, in the order :func:`describe` lists them.	`None`
`prefix`	`Optional[str]`	For a model whose outputs are a numbered family (MEM themes), the stem to number from: `"fb_topics"` gives `fb_topics_1` upward. Cannot be combined with `outputs`.	`None`
`class_names`	`Optional[Mapping[str, Mapping[str, str]]]`	`{outcome: {class as fitted: label to write}}` for a model that predicts categories. Merged with what the file already has, one outcome at a time: a class left out keeps its current label, and giving a class its own name (or a blank) removes its relabel. Two classes of one outcome cannot end up with the same label.	`None`
`apply`	`Optional[Mapping[str, object]]`	Values for the settings its kind registers (:attr:`ModelType. apply_settings`), coerced and checked; an unknown key is refused naming the ones there are. Merged with the file's `apply` block.	`None`

Returns:

Type	Description
`ModelInfo`	The model as it now reads.

Notes

Every edit rewrites the file rather than a sidecar, so a model that is copied, zipped or emailed keeps the names and settings it was given -- a sidecar would be left behind by every one of those, and the columns would quietly revert.

Source code in src\taters\helpers\model_spec.py

def edit_model(model_json: PathLike, *, name: Optional[str] = None,
               outputs: Optional[Sequence[str]] = None,
               prefix: Optional[str] = None,
               class_names: Optional[Mapping[str, Mapping[str, str]]] = None,
               apply: Optional[Mapping[str, object]] = None) -> ModelInfo:
    """
    Change what a model is called, what it writes, and how it is applied.

    Parameters
    ----------
    model_json
        The model file, rewritten in place (atomically).
    name
        The researcher's name for the model. Shown in every menu.
    outputs
        One new label per output, in the order :func:`describe` lists them.
    prefix
        For a model whose outputs are a numbered family (MEM themes), the
        stem to number from: ``"fb_topics"`` gives ``fb_topics_1`` upward.
        Cannot be combined with ``outputs``.
    class_names
        ``{outcome: {class as fitted: label to write}}`` for a model that
        predicts categories. Merged with what the file already has, one
        outcome at a time: a class left out keeps its current label, and
        giving a class its own name (or a blank) removes its relabel. Two
        classes of one outcome cannot end up with the same label.
    apply
        Values for the settings its kind registers (:attr:`ModelType.
        apply_settings`), coerced and checked; an unknown key is refused
        naming the ones there are. Merged with the file's ``apply`` block.

    Returns
    -------
    ModelInfo
        The model as it now reads.

    Notes
    -----
    Every edit rewrites the file rather than a sidecar, so a model that is
    copied, zipped or emailed keeps the names and settings it was given --
    a sidecar would be left behind by every one of those, and the columns
    would quietly revert.
    """
    path = Path(model_json)
    doc = _read_doc(path)
    info = describe(path)
    spec = BY_ID[info.type_id]

    if class_names is not None:
        fitted = dict(spec.classes(doc)) if spec.classes is not None else {}
        if not fitted:
            raise ValueError(
                f"a {info.type_label} model predicts no categories, so it "
                f"has no classes to relabel")
        current = {o: dict(m or {}) for o, m in
                   (doc.get("class_names") or {}).items()}
        for outcome, mapping in class_names.items():
            if outcome not in fitted:
                raise ValueError(
                    f"this model has no categorical outcome {outcome!r} "
                    f"(it has {', '.join(fitted) or 'none'})")
            table = current.setdefault(outcome, {})
            for cls, label in (mapping or {}).items():
                cls = str(cls)
                if cls not in [str(c) for c in fitted[outcome]]:
                    raise ValueError(
                        f"outcome {outcome!r} has no class {cls!r} (its "
                        f"classes are {', '.join(map(str, fitted[outcome]))})")
                text = str(label if label is not None else "").strip()
                if not text or text == cls:
                    table.pop(cls, None)
                else:
                    table[cls] = _valid_label(text, what="class label")
            written = [table.get(str(c), str(c)) for c in fitted[outcome]]
            clashes = sorted({w for w in written if written.count(w) > 1})
            if clashes:
                raise ValueError(
                    f"two classes of {outcome!r} cannot share the label "
                    f"{clashes[0]!r} -- the results could not tell them apart")
        doc["class_names"] = {o: m for o, m in current.items() if m}
        if not doc["class_names"]:
            doc.pop("class_names", None)

    if apply is not None:
        if not spec.apply_settings:
            raise ValueError(
                f"a {info.type_label} model has no settings for how it is "
                f"applied")
        block = dict(doc.get("apply") or {})
        for key, value in apply.items():
            setting = spec.apply_settings.get(str(key))
            if setting is None:
                raise ValueError(
                    f"{key!r} is not a setting of a {info.type_label} model "
                    f"(they are: {', '.join(spec.apply_settings)})")
            block[str(key)] = setting.coerce(value, str(key))
        doc["apply"] = block

    if outputs is not None and prefix is not None:
        raise ValueError(
            "give either one label per output or a single prefix, not both")
    if prefix is not None:
        if not info.bulk_outputs:
            raise ValueError(
                f"a {info.type_label} model has {info.n_outputs} named "
                f"output(s), so name them individually rather than by "
                f"prefix")
        outputs = [f"{prefix}_{i + 1}" for i in range(info.n_outputs)]
    if outputs is not None:
        labels = [_valid_label(label) for label in outputs]
        duplicates = sorted({label for label in labels
                             if labels.count(label) > 1})
        if duplicates:
            raise ValueError(
                f"two outputs cannot share a name ({', '.join(duplicates)}) "
                f"-- one column would overwrite the other")
        if spec.write_outputs is None:
            raise ValueError(
                f"a {info.type_label} model's output names cannot be changed")
        spec.write_outputs(doc, labels)
    if name is not None:
        doc["name"] = _valid_label(name, what="model name")

    with atomic_write(path, mode="w", encoding="utf-8") as fh:
        json.dump(doc, fh, indent=1)
    return replace(describe(path), path=path)

encoder_problem ¶

encoder_problem(model_json)

Why this file is not a usable text encoder, or "" when it is.

Structural and torch-free: the manifest's kind and format, and a payload folder beside it holding a config, weights and a tokenizer. Loading the weights is the extractor's job, minutes later, on the device it chose.

Source code in src\taters\helpers\model_spec.py

def encoder_problem(model_json: PathLike) -> str:
    """
    Why this file is not a usable text encoder, or "" when it is.

    Structural and torch-free: the manifest's kind and format, and a payload
    folder beside it holding a config, weights and a tokenizer. Loading the
    weights is the extractor's job, minutes later, on the device it chose.
    """
    path = Path(model_json)
    try:
        doc = _read_doc(path)
    except UnknownModel as e:
        return str(e)
    except FileNotFoundError as e:
        return str(e)
    if doc.get("kind") != ENCODER_KIND:
        return (f"{path.name} is not a text encoder "
                f"(a {doc.get('kind')!r} file).")
    try:
        fmt = int(doc.get("format", 0))
    except (TypeError, ValueError):
        return f"{path.name} has no readable format number."
    if fmt > ENCODER_FORMAT:
        return (f"{path.name} was written by a newer Taters (format {fmt}); "
                f"this build reads format {ENCODER_FORMAT}. Update Taters.")
    from .library import payload_of

    declared = payload_of(path)
    if not declared:
        return (f"{path.name} names no weights folder beside it. The "
                f"encoder's weights folder has to travel with the file.")
    folder = declared[0]
    if not folder.is_dir():
        return (f"{path.name}'s weights folder {folder.name} is not beside "
                f"it. Copy the model file and its weights together.")
    names = {f.name for f in folder.iterdir()}
    for needed in _ENCODER_FILES:
        if needed not in names:
            return f"{folder.name} has no {needed}; the encoder is incomplete."
    if not names & set(_ENCODER_WEIGHTS):
        return f"{folder.name} has no model weights; the encoder is incomplete."
    if not names & set(_ENCODER_TOKENIZER):
        return f"{folder.name} has no tokenizer; the encoder is incomplete."
    return ""

feature_plan ¶

feature_plan(info)

Read a model's own account of the features it was fitted on.

Returns a :class:FeaturePlan. A model that reads text needs no features and gets an empty plan with no problems -- it re-derives its own instrument, so there is nothing for a run to arrange.

Source code in src\taters\helpers\model_spec.py

def feature_plan(info: ModelInfo) -> FeaturePlan:
    """
    Read a model's own account of the features it was fitted on.

    Returns a :class:`FeaturePlan`. A model that reads text needs no features
    and gets an empty plan with no problems -- it re-derives its own
    instrument, so there is nothing for a run to arrange.
    """
    import json

    name_slug = slug(info.name)
    controls = tuple(sorted({str(e.get("column")) for e in info.controls
                             if e.get("column")}))
    if info.needs != "features":
        return FeaturePlan(model=info.display(), slug=name_slug,
                           path=str(info.path))

    recorded = dict(info.provenance or {})
    if not recorded:
        return FeaturePlan(
            model=info.display(), slug=name_slug, path=str(info.path),
            controls=controls,
            problems=(f"{info.display()} does not record how its features "
                      f"were measured, so they cannot be reproduced.",))

    try:
        with Path(info.path).open("r", encoding="utf-8") as fh:
            carried = json.load(fh).get("assets") or {}
    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
        carried = {}

    tables, problems = [], []
    for stem, rec in sorted(recorded.items()):
        if rec.get("state") != "recorded":
            problems.append(
                f"{info.display()} was fitted on the {stem!r} feature table "
                f"and has no record of how it was measured, so that table "
                f"cannot be reproduced.")
            continue
        assets, missing = _carried(rec.get("assets"), carried)
        replay = None
        if rec.get("replay"):
            fitted, gone = _carried((rec["replay"] or {}).get("assets"),
                                    carried)
            missing += gone
            replay = {"target": str((rec["replay"] or {}).get("call") or ""),
                      "assets": fitted}
        if missing:
            problems.append(
                f"{info.display()} used word list(s) it does not carry "
                f"({', '.join(sorted(set(missing)))}), so the {stem!r} table "
                f"cannot be reproduced. Fit the model again with a build of "
                f"Taters that embeds them.")
            continue
        tables.append(TablePlan(
            stem=str(stem), target=str(rec.get("call") or ""),
            instrument=dict(rec.get("instrument") or {}),
            assets=assets,
            digest=str((rec.get("digests") or {}).get("instrument") or ""),
            grain=dict(rec.get("grain") or {}), replay=replay))
    return FeaturePlan(model=info.display(), slug=name_slug,
                       tables=tuple(tables), problems=tuple(problems),
                       path=str(info.path), controls=controls)

library_kind_for ¶

library_kind_for(model_json)

Which library kind a model file belongs in: encoders for an adapted encoder, models for anything that scores.

Source code in src\taters\helpers\model_spec.py

def library_kind_for(model_json: PathLike) -> str:
    """Which library kind a model file belongs in: ``encoders`` for an
    adapted encoder, ``models`` for anything that scores."""
    try:
        doc = _read_doc(Path(model_json))
    except (UnknownModel, FileNotFoundError, OSError):
        return "models"
    return "encoders" if doc.get("kind") == ENCODER_KIND else "models"

models_produced ¶

models_produced(folder)

The model files a run left behind, each with a label for a menu.

Scoring models of every registered kind and adapted encoders alike, found by their manifests (never inside a payload folder), so the finish screen and the training task can offer to add them to the library.

Source code in src\taters\helpers\model_spec.py

def models_produced(folder: PathLike) -> List[Tuple[Path, str]]:
    """
    The model files a run left behind, each with a label for a menu.

    Scoring models of every registered kind and adapted encoders alike,
    found by their manifests (never inside a payload folder), so the finish
    screen and the training task can offer to add them to the library.
    """
    from .library import model_files

    out: List[Tuple[Path, str]] = []
    for path in model_files(Path(folder)):
        try:
            doc = _read_doc(path)
        except (UnknownModel, FileNotFoundError, OSError):
            continue
        if doc.get("kind") == ENCODER_KIND:
            label, _note = describe_encoder(path)
            out.append((path, f"{label} (text encoder)"))
            continue
        try:
            info = describe(path)
        except (UnknownModel, FileNotFoundError, OSError):
            continue
        out.append((path, info.display()))
    return out

one_model_path ¶

one_model_path(model_json)

Exactly one model file, or an actionable refusal.

The library hands steps a folder (or a list) meaning "everything of this kind", which is right for dictionaries and wrong for a model: scoring with an unspecified one of three is not a thing anyone means. A single file passes through, a folder or a list resolves to its .json files, and anything other than exactly one refuses with the fix named.

Owned here because it is about model files, not about any one kind of model: ridge and the topic model each carried a copy, with different refusal wording and one of them letting a folder through unresolved.

Source code in src\taters\helpers\model_spec.py

def one_model_path(model_json) -> Path:
    """
    Exactly one model file, or an actionable refusal.

    The library hands steps a folder (or a list) meaning "everything of this
    kind", which is right for dictionaries and wrong for a model: scoring
    with an unspecified one of three is not a thing anyone means. A single
    file passes through, a folder or a list resolves to its ``.json`` files,
    and anything other than exactly one refuses with the fix named.

    Owned here because it is about model files, not about any one kind of
    model: ridge and the topic model each carried a copy, with different
    refusal wording and one of them letting a folder through unresolved.
    """
    from .library import model_files

    values = list(model_json) if isinstance(model_json, (list, tuple)) \
        else [model_json]
    found: List[Path] = []
    for value in values:
        path = Path(value)
        if path.is_dir():
            # not `rglob`: a checkpoint folder inside a model's payload holds
            # a config.json and a tokenizer.json that aren't model files.
            found += model_files(path)
        else:
            found.append(path)
    if len(found) == 1:
        return found[0]
    if not found:
        raise ValueError(
            f"no model file found at {model_json!r}. Import one into your "
            f"library, or point the step at one.")
    names = ", ".join(p.name for p in found[:6])
    raise ValueError(
        f"{len(found)} model files found ({names}) and a step can only use "
        f"one. Pick it in the step's options.")

output_label ¶

output_label(doc, outcome)

The column-name stem for one outcome: its rename, or the outcome itself.

Read by the scoring functions rather than by the wizard, so a renamed model produces renamed columns however it is invoked -- through the app, through the API, or from the command line.

Source code in src\taters\helpers\model_spec.py

def output_label(doc: dict, outcome: str) -> str:
    """
    The column-name stem for one outcome: its rename, or the outcome itself.

    Read by the scoring functions rather than by the wizard, so a renamed
    model produces renamed columns however it is invoked -- through the app,
    through the API, or from the command line.
    """
    names = doc.get("output_names") or {}
    label = names.get(outcome)
    return str(label) if label else str(outcome)

rename_model ¶

rename_model(
    model_json, *, name=None, outputs=None, prefix=None
)

Name a model and its output columns -- :func:edit_model without the class labels and apply settings. Kept for the callers that only ever name things.

Source code in src\taters\helpers\model_spec.py

def rename_model(model_json: PathLike, *, name: Optional[str] = None,
                 outputs: Optional[Sequence[str]] = None,
                 prefix: Optional[str] = None) -> ModelInfo:
    """Name a model and its output columns -- :func:`edit_model` without
    the class labels and apply settings. Kept for the callers that only
    ever name things."""
    return edit_model(model_json, name=name, outputs=outputs, prefix=prefix)

scorer ¶

scorer(type_id)

The function that scores a new table with this kind of model.

Resolved from the registry rather than chosen at the call site, so that adding a kind of model is adding a registry entry -- and forgetting the entry is this refusal, rather than the model being quietly scored by whichever applier happened to be the else branch.

Source code in src\taters\helpers\model_spec.py

def scorer(type_id: str):
    """
    The function that scores a new table with this kind of model.

    Resolved from the registry rather than chosen at the call site, so that
    adding a kind of model is adding a registry entry -- and forgetting the
    entry is this refusal, rather than the model being quietly scored by
    whichever applier happened to be the `else` branch.
    """
    import importlib

    spec = BY_ID.get(type_id)
    if spec is None or not spec.score:
        raise ValueError(
            f"no scoring function is registered for a {type_id!r} model. Add "
            f"`score=` to its entry in MODEL_TYPES -- see 'Adding a kind of "
            f"model' in this module's docstring.")
    module, _, name = spec.score.partition(":")
    return getattr(importlib.import_module(module), name)

slug ¶

slug(name, fallback='model')

Filesystem- and template-safe: artifact references split on . and :, and a save_as built from this must not contain either.

One spelling for every model-shaped file name. Three private copies had grown -- here, in ridge (fallback "set") and in score_model -- and two of them kept the dot the third refused, so the same set name could produce two different file names depending on which module wrote it.

Source code in src\taters\helpers\model_spec.py

def slug(name: str, fallback: str = "model") -> str:
    """
    Filesystem- and template-safe: artifact references split on ``.`` and
    ``:``, and a ``save_as`` built from this must not contain either.

    One spelling for every model-shaped file name. Three private copies had
    grown -- here, in ridge (fallback "set") and in score_model -- and two of
    them kept the dot the third refused, so the same set name could produce
    two different file names depending on which module wrote it.
    """
    import re

    return re.sub(r"[^0-9A-Za-z_-]+", "-", str(name)).strip("-") or fallback

taters.helpers.settings ¶

Persistent user settings, and the one that matters today: where downloaded models are kept.

Taters has a home folder ($TATERS_HOME or ~/.taters) for the library; this module puts a small settings.json beside it for choices that should outlive a session and are not about any one pipeline. The first such choice is the model cache: every transformer, sentence-transformers and Whisper model is downloaded by the Hugging Face hub library into its cache, which lives under the user's home by default. On a shared server that is the wrong place -- one copy per user of a 500 MB encoder, on a home partition that is small on purpose -- and the fix has been an environment variable that has to be set in every shell. Here it is a setting, chosen once in Settings and applied every time Taters starts, before any of those libraries is imported.

Resolution order for the model cache, most explicit first:

TATERS_MODEL_CACHE in the environment (an administrator's or a test's word, and the one thing that beats a saved setting);
model_cache in settings.json (the user's choice in Settings);
HF_HUB_CACHE or HF_HOME from the environment (the hub library's own conventions, honored as they always were);
the hub library's default, ~/.cache/huggingface/hub.

Whichever wins is exported as HF_HUB_CACHE at import (see :func:apply_model_cache), so the hub library, transformers, sentence-transformers and faster-whisper all download to and read from the same place -- and it is passed explicitly to the transformer steps too, in case those libraries were imported before Taters was.

apply_model_cache ¶

apply_model_cache()

Export the chosen cache as HF_HUB_CACHE so every downloading library agrees with it. Called when Taters is imported. Returns the path when a Taters-level choice (environment or setting) was applied, else None -- the hub library's own environment is left exactly as found.

Source code in src\taters\helpers\settings.py

def apply_model_cache() -> Optional[Path]:
    """
    Export the chosen cache as ``HF_HUB_CACHE`` so every downloading library
    agrees with it. Called when Taters is imported. Returns the path when a
    Taters-level choice (environment or setting) was applied, else None --
    the hub library's own environment is left exactly as found.
    """
    path, source = model_cache_source()
    if source not in ("environment", "setting"):
        return None
    os.environ["HF_HUB_CACHE"] = str(path)
    # older transformers releases read this name; it's harmless on newer ones.
    os.environ.setdefault("TRANSFORMERS_CACHE", str(path))
    return path

clear_setting ¶

clear_setting(key)

Forget one setting; nothing happens when it was not set.

Source code in src\taters\helpers\settings.py

def clear_setting(key: str) -> None:
    """Forget one setting; nothing happens when it was not set."""
    from .atomic import atomic_write

    doc = load_settings()
    if str(key) not in doc:
        return
    del doc[str(key)]
    path = settings_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with atomic_write(path, mode="w", encoding="utf-8") as fh:
        json.dump(doc, fh, indent=1)

describe_model_cache ¶

describe_model_cache()

One line for the setup check: the folder, how it was chosen, and how much is in it.

Source code in src\taters\helpers\settings.py

def describe_model_cache() -> str:
    """One line for the setup check: the folder, how it was chosen, and
    how much is in it."""
    path, source = model_cache_source()
    how = {"environment": f"from {MODEL_CACHE_ENV}",
           "setting": "chosen in Settings",
           "hub environment": "from HF_HUB_CACHE / HF_HOME",
           "default": "the default"}[source]
    if not path.is_dir():
        return f"{path}  ·  {how}  ·  nothing downloaded yet"
    models = [p for p in path.glob("models--*") if p.is_dir()]
    size = sum(f.stat().st_size for f in path.rglob("*") if f.is_file())
    from .library import _human_size

    return f"{path}  ·  {how}  ·  {len(models)} model(s), {_human_size(size)}"

inspect_row_limit ¶

inspect_row_limit()

How many rows to read when looking a spreadsheet over. 0 means all.

All of them by default, because the questions the wizard builds from a spreadsheet are only as true as what it read: on a real file, ten columns looked constant within a group across the first two hundred rows and were not across the other seven hundred, so they were offered as controls that would have come out empty. Reading everything also says at the moment the file is chosen whether its rows match its header, which is much cheaper to learn then than after an hour of extraction.

Lowered by somebody whose files are big enough that a full pass is worth skipping, in Settings. The environment variable wins, for scripts and tests.

Source code in src\taters\helpers\settings.py

def inspect_row_limit() -> int:
    """
    How many rows to read when looking a spreadsheet over. ``0`` means all.

    All of them by default, because the questions the wizard builds from a
    spreadsheet are only as true as what it read: on a real file, ten
    columns looked constant within a group across the first two hundred rows
    and were not across the other seven hundred, so they were offered as
    controls that would have come out empty. Reading everything also says at
    the moment the file is chosen whether its rows match its header, which
    is much cheaper to learn then than after an hour of extraction.

    Lowered by somebody whose files are big enough that a full pass is worth
    skipping, in Settings. The environment variable wins, for scripts and
    tests.
    """
    raw = os.environ.get(INSPECT_ROWS_ENV)
    if raw is None:
        raw = load_settings().get(INSPECT_ROWS_KEY)
    if raw is None or str(raw).strip() == "":
        return 0
    try:
        return max(0, int(float(str(raw).strip())))
    except ValueError:
        # a hand-edited settings file with "lots" in it is not a reason to
        # refuse to start, and reading everything is the safe answer.
        return 0

load_settings ¶

load_settings()

The saved settings, or {}; a damaged file reads as empty rather than stopping Taters from starting.

Source code in src\taters\helpers\settings.py

def load_settings() -> Dict[str, Any]:
    """The saved settings, or ``{}``; a damaged file reads as empty rather
    than stopping Taters from starting."""
    path = settings_path()
    try:
        doc = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError, UnicodeDecodeError):
        return {}
    return doc if isinstance(doc, dict) else {}

model_cache_dir ¶

model_cache_dir()

Where downloaded models live; see the module docstring for the order.

Source code in src\taters\helpers\settings.py

def model_cache_dir() -> Path:
    """Where downloaded models live; see the module docstring for the order."""
    return model_cache_source()[0]

model_cache_source ¶

model_cache_source()

The model cache and which rule chose it: "environment" (TATERS_MODEL_CACHE), "setting" (chosen in Settings), "hub environment" (HF_HUB_CACHE/HF_HOME) or "default".

Source code in src\taters\helpers\settings.py

def model_cache_source() -> Tuple[Path, str]:
    """
    The model cache and which rule chose it: ``"environment"``
    (``TATERS_MODEL_CACHE``), ``"setting"`` (chosen in Settings),
    ``"hub environment"`` (``HF_HUB_CACHE``/``HF_HOME``) or ``"default"``.
    """
    override = os.environ.get(MODEL_CACHE_ENV)
    if override:
        return Path(os.path.expandvars(override)).expanduser(), "environment"
    saved = load_settings().get(MODEL_CACHE_KEY)
    if saved:
        return Path(os.path.expandvars(str(saved))).expanduser(), "setting"
    if os.environ.get("HF_HUB_CACHE") or os.environ.get("HF_HOME"):
        return _hub_default(), "hub environment"
    return _hub_default(), "default"

save_setting ¶

save_setting(key, value)

Write one setting, keeping the others, atomically.

Source code in src\taters\helpers\settings.py

def save_setting(key: str, value: Any) -> Path:
    """Write one setting, keeping the others, atomically."""
    from .atomic import atomic_write

    doc = load_settings()
    doc[str(key)] = value
    path = settings_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with atomic_write(path, mode="w", encoding="utf-8") as fh:
        json.dump(doc, fh, indent=1)
    return path

settings_path ¶

settings_path()

$TATERS_HOME/settings.json (or ~/.taters/settings.json).

Source code in src\taters\helpers\settings.py

def settings_path() -> Path:
    """``$TATERS_HOME/settings.json`` (or ``~/.taters/settings.json``)."""
    return _home() / "settings.json"

taters.helpers.update_check ¶

"A newer version is out" -- said once, quietly, and never in the way.

The rules this is built around, in order of importance:

The menu never waits for the network. The note is read from a cached answer, which is instant; the refresh that produces that answer runs in a background thread and its result is for next launch. A slow network, a proxy that blackholes the request, no network at all -- none of it can delay the first screen or stop Taters starting.
It can be turned off, and turning it off is honored everywhere. Some people run this on data that cannot leave the building, and an unexplained outbound connection is a conversation with IT nobody wants. Setting TATERS_NO_UPDATE_CHECK=1, or the saved setting, stops it dead -- no thread, no request.
It cannot break anything. Every path here is wrapped. A failure leaves no note and no complaint; there is nothing here worth interrupting a run for.
It says one thing. "Newer version available: v0.7.3" under the banner, in dim text. Not an exhortation, not something to dismiss.

note ¶

note()

The line to print under the banner, or "" -- read from cache, instantly.

Returns "" when: the check is off, Taters is running from a source tree with no version metadata, nothing has been cached yet (the first launch), the cached answer is not newer, or anything at all goes wrong.

Source code in src\taters\helpers\update_check.py

def note() -> str:
    """The line to print under the banner, or "" -- read from cache, instantly.

    Returns "" when: the check is off, Taters is running from a source tree
    with no version metadata, nothing has been cached yet (the first launch),
    the cached answer is not newer, or anything at all goes wrong.
    """
    try:
        if _disabled():
            return ""
        here = _release(_installed())
        if here is None:
            # no metadata, or a version this cannot compare. running from a
            # checkout is the normal case for that, and a developer does not
            # need telling that PyPI is behind them.
            return ""
        seen = load_settings().get(SETTING_KEY, {}).get("latest", "")
        there = _release(str(seen))
        if there is None or there <= here:
            return ""
        return f"Newer version available: v{seen}"
    except Exception:
        return ""

refresh_in_background ¶

refresh_in_background()

Start the refresh, or don't. Returns the thread for tests to join.

A daemon thread: if somebody quits Taters two seconds after opening it, the interpreter exits without waiting on a socket nobody is reading.

Source code in src\taters\helpers\update_check.py

def refresh_in_background() -> Optional[threading.Thread]:
    """Start the refresh, or don't. Returns the thread for tests to join.

    A daemon thread: if somebody quits Taters two seconds after opening it,
    the interpreter exits without waiting on a socket nobody is reading.
    """
    try:
        if _disabled() or _release(_installed()) is None:
            return None
        thread = threading.Thread(target=_fetch_and_store, name="taters-update-check",
                                  daemon=True)
        thread.start()
        return thread
    except Exception:
        return None

taters.helpers.feature_columns ¶

What every feature table calls its columns, declared once and checked at build time.

The problem this solves only shows up when two measures meet. Feature tables get joined side by side into one analysis table, and if two of them use the same column name, something has to give. stats.assemble already handles that: it refuses two tables with the same stem, and renames any column used by two tables to <stem>__<column> in every table that has it, so the outcome does not depend on file order.

That is a good safety net and it stays. But it is reactive, and reactive renaming has a property that is poison for research: the name depends on what else ran. A column is Topic_1 in a pipeline with one topic model and lda_topics__Topic_1 in a pipeline with two. The same instrument, the same corpus, two different column names -- so a script written against one study silently fails against the next, and two results tables cannot be compared without knowing what else was in each run.

So the names we choose have to be disjoint up front. Each feature module declares what it writes; :func:overlaps finds any two declarations that could produce the same name; and tests/test_feature_columns.py fails the build if any do, naming both offenders. Collisions among shipped measures therefore never happen, the reactive rename never fires for them, and Topic_1 is Topic_1 in every pipeline forever.

Three kinds of column name exist, and only two of them can be policed here:

Fixed -- names we choose and always write (flesch_reading_ease).
Patterned -- names we choose whose tail is a number or a setting (Theme_{n}, msttr_{n}). Declared as a pattern, matched as one.
Dynamic -- names that come out of the user's own data: the categories in their dictionary, their archetype names, however many dimensions their encoder has. We cannot know these and do not pretend to. They are declared with a sentence saying where they come from, and they are exactly what the reactive rename in assemble is for.

Adding a measure means adding a FEATURE_COLUMNS to its module. The test finds it through the recipe catalog rather than through a list kept here, so forgetting is a failure rather than a silence.

ColumnSpec `dataclass` ¶

ColumnSpec(
    label,
    names=(),
    patterns=(),
    reduces_to="Component",
    dynamic="",
    _module="",
)

One module's claim about the columns it writes.

Parameters:

Name	Type	Description	Default
`label`	`str`	What to call this measure in a collision message. A person reads it.	required
`names`	`tuple of str`	Column names always written, spelled exactly.	`()`
`patterns`	`tuple of str`	Names whose tail varies with a setting, with a placeholder where the varying part goes. `{n}` is a number (`"Theme_{n}"`, `"msttr_{n}"`) and `{}` is anything (`"pos_{}"`, where the tail is a tag). Exactly one placeholder per pattern; the rest is matched literally. The distinction earns its keep: sentence embeddings write `e0, e1, …` and transformer embeddings write `e_1, e_2, …`, which differ by one character and do not collide -- but a placeholder meaning "anything" would report them as if they did, and the noise would train somebody to ignore this check.	`()`
`reduces_to`	`str`	What a PCA over these columns produces, as a singular noun. Reducing a topic model's topics does not give you "components", it gives you something worth naming -- so the topic models say `"Supertopic"` and everything else keeps `"Component"`. It is declared here rather than worked out by the statistics stage because this is where the codebase already knows what a column is.	`'Component'`
`dynamic`	`str`	Non-empty when this module also writes columns whose names come from the user's data rather than from us. The text says where they come from. It is documentation, not an exemption: whatever is declared in `names` and `patterns` is still checked. The undeclarable rest is what `assemble`'s reactive rename exists for.	`''`

Notes

Bookkeeping columns are left out on purpose. Several modules write token_count beside their measures, and declaring it would report a collision on a column no analysis ever treats as a feature -- assemble keeps those aside from the feature sets and renames them harmlessly if two tables carry one. Declare what somebody would analyze.

pattern_regex ¶

pattern_regex(pattern)

A pattern as a regex: everything literal but the one placeholder.

{n} becomes \d+ and {*} becomes .+. Use {n} whenever the varying part really is a number, because it is what tells e{n} and e_{n} apart -- two real patterns in this codebase that would otherwise read as the same one.

Source code in src\taters\helpers\feature_columns.py

def pattern_regex(pattern: str) -> "re.Pattern[str]":
    """A pattern as a regex: everything literal but the one placeholder.

    ``{n}`` becomes ``\\d+`` and ``{*}`` becomes ``.+``. Use ``{n}`` whenever
    the varying part really is a number, because it is what tells ``e{n}`` and
    ``e_{n}`` apart -- two real patterns in this codebase that would otherwise
    read as the same one.
    """
    placeholder, expansion = ("{n}", r"\d+") if "{n}" in pattern else ("{*}", ".+")
    head, _, tail = pattern.partition(placeholder)
    return re.compile(f"{re.escape(head)}{expansion}{re.escape(tail)}$")

overlaps ¶

overlaps(specs)

Every pair of declarations that could write the same column name.

Returns:

Type	Description
`list of (label, label, names)`	One entry per colliding pair, with the names they collide on. Empty when the declarations are disjoint, which is the only acceptable state for the measures Taters ships.

Source code in src\taters\helpers\feature_columns.py

def overlaps(specs: Sequence[ColumnSpec]) -> List[Tuple[str, str, List[str]]]:
    """
    Every pair of declarations that could write the same column name.

    Returns
    -------
    list of (label, label, names)
        One entry per colliding pair, with the names they collide on. Empty
        when the declarations are disjoint, which is the only acceptable state
        for the measures Taters ships.
    """
    found: List[Tuple[str, str, List[str]]] = []
    for i, first in enumerate(specs):
        for second in specs[i + 1:]:
            shared = _collide(first, second)
            if shared:
                found.append((first.label, second.label, shared))
    return found

by_module ¶

by_module(specs)

Declarations keyed by the module that made them, for error messages.

Source code in src\taters\helpers\feature_columns.py

def by_module(specs: Sequence[ColumnSpec]) -> Dict[str, ColumnSpec]:
    """Declarations keyed by the module that made them, for error messages."""
    return {spec._module or spec.label: spec for spec in specs}

registry ¶

registry()

Every declaration, keyed by module. Built once, and never raises.

A module that cannot be imported (an optional dependency is missing) is skipped rather than fatal: the registry is used to name things nicely, and a missing encoder should not stop a run that never wanted one.

Source code in src\taters\helpers\feature_columns.py

def registry() -> Dict[str, ColumnSpec]:
    """Every declaration, keyed by module. Built once, and never raises.

    A module that cannot be imported (an optional dependency is missing) is
    skipped rather than fatal: the registry is used to *name* things nicely,
    and a missing encoder should not stop a run that never wanted one.
    """
    if not _REGISTRY:
        import importlib

        for path in DECLARING_MODULES:
            try:
                spec = getattr(importlib.import_module(path), "FEATURE_COLUMNS", None)
            except Exception:                   # pragma: no cover - optional extras
                continue
            if spec is not None:
                _REGISTRY[path] = spec
    return dict(_REGISTRY)

reduced_name ¶

reduced_name(columns, *, set_name='')

What to call a component built from these columns: "Supertopic", or "Component" when nothing more specific is known.

Matched on the column names themselves rather than threaded down from the pipeline, because by the time the statistics stage reduces a feature set, the set is just a list of column names in a table -- whatever produced it is long out of scope.

Source code in src\taters\helpers\feature_columns.py

def reduced_name(columns: Sequence[str], *, set_name: str = "") -> str:
    """
    What to call a component built from these columns: "Supertopic", or
    "Component" when nothing more specific is known.

    Matched on the column names themselves rather than threaded down from the
    pipeline, because by the time the statistics stage reduces a feature set,
    the set is just a list of column names in a table -- whatever produced it
    is long out of scope.
    """
    trimmed = []
    for column in columns:
        prefix = f"{set_name}{SEPARATOR}"
        trimmed.append(column[len(prefix):]
                       if set_name and column.startswith(prefix) else column)

    if not trimmed:
        return "Component"

    for spec in registry().values():
        if spec.reduces_to == "Component":
            continue
        literal = set(spec.names)
        checks = [pattern_regex(p) for p in spec.patterns]
        # *every* column has to belong, not just one. a feature set that mixes
        # a topic model with readability indices reduces to something that is
        # genuinely not a supertopic, and the combined "all" set -- which the
        # statistics stage builds by default -- is exactly that mixture. an
        # `any` here named its components supertopics on the strength of one
        # matching column.
        if all(c in literal or any(rx.fullmatch(c) for rx in checks)
               for c in trimmed):
            return spec.reduces_to
    return "Component"

Utilities & Helpers¶

taters.helpers.feature_gather ¶

AggregationPlan dataclass ¶

aggregate_features ¶

feature_gather ¶

gather_csvs_to_one ¶

make_plan ¶

taters.helpers.find_files ¶

find_files ¶

taters.helpers.text_gather ¶

csv_to_analysis_ready_csv ¶

resolve_analysis_ready ¶

txt_folder_to_analysis_ready_csv ¶

taters.helpers.library ¶

LibraryCollision ¶

LibraryKind dataclass ¶

asset_problem ¶

delete ¶

display_name ¶

entries ¶

expand ¶

export_to ¶

find_payload ¶

import_into ¶

kind_by_id ¶

kind_dir ¶

library_home ¶

model_files ¶

payload_dirs ¶

payload_of ¶

payload_size ¶

rename ¶

taters.helpers.model_spec ¶

ApplySetting dataclass ¶

coerce ¶

FeaturePlan dataclass ¶

ModelInfo dataclass ¶

display ¶

ModelType dataclass ¶

TablePlan dataclass ¶

UnknownModel ¶

apply_defaults ¶

class_label ¶

describe ¶

describe_all ¶

describe_encoder ¶

edit_model ¶

encoder_problem ¶

feature_plan ¶

library_kind_for ¶

models_produced ¶

one_model_path ¶

output_label ¶

rename_model ¶

scorer ¶

slug ¶

taters.helpers.settings ¶

apply_model_cache ¶

clear_setting ¶

describe_model_cache ¶

inspect_row_limit ¶

load_settings ¶

model_cache_dir ¶

model_cache_source ¶

save_setting ¶

settings_path ¶

taters.helpers.update_check ¶

note ¶

refresh_in_background ¶

taters.helpers.feature_columns ¶

ColumnSpec dataclass ¶

pattern_regex ¶

overlaps ¶

by_module ¶

registry ¶

reduced_name ¶

AggregationPlan `dataclass` ¶

LibraryKind `dataclass` ¶

ApplySetting `dataclass` ¶

FeaturePlan `dataclass` ¶

ModelInfo `dataclass` ¶

ModelType `dataclass` ¶

TablePlan `dataclass` ¶

ColumnSpec `dataclass` ¶