Pipelines

taters.pipelines.run_pipeline

Taters Pipeline Runner (robust templating + flexible call resolution)

• ITEM steps run once per input (fan-out concurrently).
• GLOBAL steps run once (barrier before/after).
• Templating preserves native types when the entire value is a single template (e.g., {{var:text_cols}} → list, not "['text']").
• Calls:
  • "potato.*" → call via a Taters() instance (e.g., potato.text.analyze_with_dictionaries)
  • dotted path → import and call any function (e.g., taters.helpers.feature_gather.aggregate_features)

Usage example: python -m taters.pipelines.run_pipeline --root_dir videos --file_type video --preset conversation_video --workers 4 --var device=cuda --var overwrite_existing=true

available_presets

available_presets(root=None)

Every preset the runner can see, with its metadata.

Both search directories are covered -- the ones that ship with Taters and the ones in ./pipelines/ -- which is what lets a UI offer the built-in pipelines alongside anything the user has built. Sorted by title so the list is stable between runs.

Returns:

Type	Description
list[tuple[Path, dict]]	(path, meta) pairs. meta always has id, title, summary and tags, defaulted from the filename when the file does not declare them.

Source code in src\taters\pipelines\run_pipeline.py

def available_presets(root: Optional[Path] = None) -> List[Tuple[Path, dict]]:
    """
    Every preset the runner can see, with its metadata.

    Both search directories are covered -- the ones that ship with Taters and
    the ones in ``./pipelines/`` -- which is what lets a UI offer the built-in
    pipelines alongside anything the user has built. Sorted by title so the
    list is stable between runs.

    Returns
    -------
    list[tuple[Path, dict]]
        ``(path, meta)`` pairs. ``meta`` always has ``id``, ``title``,
        ``summary`` and ``tags``, defaulted from the filename when the file
        does not declare them.
    """
    out = [(path, _load_preset_meta(path)) for path in _iter_presets(root)]
    return sorted(out, key=lambda pair: str(pair[1].get("title", "")).lower())

discover_inputs

discover_inputs(root_dir, kind)

Recursively discover input files under a root folder.

The preset's ITEM-scoped steps operate over a list of inputs. This function builds that list by scanning root_dir and selecting files by type:

• kind == "video": only common video extensions (e.g., .mp4, .mov, .mkv)
• kind == "audio": only common audio extensions (e.g., .wav, .mp3, .flac)
• kind == "any": all files

Parameters:

Name	Type	Description	Default
root_dir	Path	Directory to scan (will be resolved to an absolute path).	required
kind	('audio', 'video', 'any')	Filter that determines which file extensions are included.	"audio","video","any"

Returns:

Type	Description
List[Path]	Sorted list of absolute file paths.

Raises:

Type	Description
FileNotFoundError	If root_dir does not exist.

Source code in src\taters\pipelines\run_pipeline.py

def discover_inputs(root_dir: Path, kind: str) -> List[Path]:
    """
    Recursively discover input files under a root folder.

    The preset's ITEM-scoped steps operate over a list of inputs. This
    function builds that list by scanning `root_dir` and selecting files by
    type:

    - kind == "video": only common video extensions (e.g., .mp4, .mov, .mkv)
    - kind == "audio": only common audio extensions (e.g., .wav, .mp3, .flac)
    - kind == "any":   all files

    Parameters
    ----------
    root_dir : Path
        Directory to scan (will be resolved to an absolute path).
    kind : {"audio","video","any"}
        Filter that determines which file extensions are included.

    Returns
    -------
    List[Path]
        Sorted list of absolute file paths.

    Raises
    ------
    FileNotFoundError
        If `root_dir` does not exist.
    """
    root_dir = root_dir.resolve()
    if not root_dir.exists():
        raise FileNotFoundError(f"root_dir not found: {root_dir}")
    out: List[Path] = []
    for p in root_dir.rglob("*"):
        if not p.is_file():
            continue
        ext = p.suffix.lower()
        if kind == "video" and ext in _VIDEO_EXTS:
            out.append(p)
        elif kind == "audio" and ext in _AUDIO_EXTS:
            out.append(p)
        elif kind == "any":
            out.append(p)
    return sorted(out)

is_builtin_preset

is_builtin_preset(path)

Whether a preset ships with Taters, and so must not be edited or deleted.

Source code in src\taters\pipelines\run_pipeline.py

def is_builtin_preset(path: Path) -> bool:
    """Whether a preset ships with Taters, and so must not be edited or deleted."""
    try:
        Path(path).resolve().relative_to(_builtin_presets_root())
        return True
    except ValueError:
        return False

load_preset_by_name

load_preset_by_name(name)

Load a named pipeline preset by meta.id or filename stem.

Presets are resolved with :func:resolve_preset_path, which searches the built-in presets/ folder as well as a project-local ./pipelines folder.

Parameters:

Name	Type	Description	Default
name	str	Preset meta.id or filename stem.	required

Returns:

Type	Description
dict	Parsed YAML as a Python dictionary. Returns {} for an empty file.

Raises:

Type	Description
FileNotFoundError	If no preset with that name exists in any search directory.

Source code in src\taters\pipelines\run_pipeline.py

def load_preset_by_name(name: str) -> dict:
    """
    Load a named pipeline preset by `meta.id` or filename stem.

    Presets are resolved with :func:`resolve_preset_path`, which searches the
    built-in `presets/` folder as well as a project-local `./pipelines` folder.

    Parameters
    ----------
    name : str
        Preset `meta.id` or filename stem.

    Returns
    -------
    dict
        Parsed YAML as a Python dictionary. Returns `{}` for an empty file.

    Raises
    ------
    FileNotFoundError
        If no preset with that name exists in any search directory.
    """
    path = resolve_preset_path(name)
    print(f"[pipeline] Using preset: {path}")
    with path.open("r", encoding="utf-8") as f:
        return yaml.safe_load(f) or {}

load_yaml_file

load_yaml_file(path)

Load a YAML file into a Python dictionary.

Parameters:

Name	Type	Description	Default
path	Path	Full path to a YAML file.	required

Returns:

Type	Description
dict	Parsed YAML contents. Empty files yield {}.

Source code in src\taters\pipelines\run_pipeline.py

def load_yaml_file(path: Path) -> dict:
    """
    Load a YAML file into a Python dictionary.

    Parameters
    ----------
    path : Path
        Full path to a YAML file.

    Returns
    -------
    dict
        Parsed YAML contents. Empty files yield `{}`.
    """
    with path.open("r", encoding="utf-8") as f:
        return yaml.safe_load(f) or {}

main

main()

Entry point for the Taters Pipeline Runner.

Responsibilities

• Parse CLI arguments (--preset or --preset-file, optional --vars-file and repeated --var key=value overrides, --workers, --quiet, etc.).
• Load the preset YAML and merge variables from three sources in order: 1) preset vars block 2) --vars-file (YAML) 3) repeated --var CLI flags
• Decide whether input discovery is required:
  • If the preset has any ITEM-scoped steps, --root_dir is required and files are discovered with discover_inputs(...).
  • If there are only GLOBAL steps, discovery is skipped entirely.
• Build a run manifest skeleton (preset name, inputs, vars, globals).
• Create a single Taters() instance (shared across all steps in the run).
• Execute each step in order:
  • ITEM steps: fan out across discovered inputs using a thread or process pool (configurable per step). A given step reuses one pool for all items to amortize worker startup.
  • GLOBAL steps: run once, in order, with a barrier between steps.
• After each step, update and persist the JSON manifest so long-running runs are observable and resumable.
• Print the final manifest path on completion.

Concurrency Notes

• The default executor for ITEM steps is a ThreadPoolExecutor (good for I/O-bound steps and for GPU inference that releases the GIL).
• For heavy Python/CPU work, presets may set engine: process on a step to use a ProcessPoolExecutor. In that case, be mindful that a new Python process is spawned for each worker (model weights may be reloaded once per worker).

Error Handling

• Individual ITEM step failures do not crash the pipeline; they mark that item as "error" in the manifest and continue.
• GLOBAL step failures are terminal for the run (the loop breaks), and the manifest is written before bailing out.
• The process exits with status 1 if anything failed — a global error or any individual item — and 0 only when every step succeeded.

Returns:

Type	Description
None	The function exits the process after writing the manifest.

Source code in src\taters\pipelines\run_pipeline.py

def main():
    """
    Entry point for the Taters Pipeline Runner.

    Responsibilities
    ----------------
    - Parse CLI arguments (`--preset` or `--preset-file`, optional `--vars-file`
      and repeated `--var key=value` overrides, `--workers`, `--quiet`, etc.).
    - Load the preset YAML and merge variables from three sources in order:
        1) preset `vars` block
        2) `--vars-file` (YAML)
        3) repeated `--var` CLI flags
    - Decide whether input discovery is required:
        * If the preset has any ITEM-scoped steps, `--root_dir` is required and
          files are discovered with `discover_inputs(...)`.
        * If there are only GLOBAL steps, discovery is skipped entirely.
    - Build a run manifest skeleton (preset name, inputs, vars, globals).
    - Create a single `Taters()` instance (shared across all steps in the run).
    - Execute each step in order:
        * ITEM steps: fan out across discovered inputs using a thread or process
          pool (configurable per step). A given step reuses one pool for all
          items to amortize worker startup.
        * GLOBAL steps: run once, in order, with a barrier between steps.
    - After each step, update and persist the JSON manifest so long-running runs
      are observable and resumable.
    - Print the final manifest path on completion.

    Concurrency Notes
    -----------------
    - The default executor for ITEM steps is a `ThreadPoolExecutor` (good for
      I/O-bound steps and for GPU inference that releases the GIL).
    - For heavy Python/CPU work, presets may set `engine: process` on a step to
      use a `ProcessPoolExecutor`. In that case, be mindful that a new Python
      process is spawned for each worker (model weights may be reloaded once per
      worker).

    Error Handling
    --------------
    - Individual ITEM step failures do not crash the pipeline; they mark that
      item as `"error"` in the manifest and continue.
    - GLOBAL step failures are terminal for the run (the loop breaks), and the
      manifest is written before bailing out.
    - The process exits with status 1 if anything failed — a global error or any
      individual item — and 0 only when every step succeeded.

    Returns
    -------
    None
        The function exits the process after writing the manifest.
    """
    # ---------------------------
    # CLI
    # ---------------------------
    ap = argparse.ArgumentParser(
        description="Taters Pipeline Runner (robust templating + flexible calls)"
    )
    ap.add_argument("--root_dir", default=None,
                    help="Folder to scan for inputs (required only if preset has ITEM steps)")
    ap.add_argument("--file_type", default="any", choices=["audio", "video", "any"],
                    help="Input type filter for discovery")

    # NOTE: not required here — we enforce after handling list/describe.
    group = ap.add_mutually_exclusive_group(required=False)
    group.add_argument("--preset", help="Preset name (taters/pipelines/presets/<name>.yaml)")
    group.add_argument("--preset-file", dest="preset_file", help="Path to preset YAML")

    ap.add_argument("--vars-file", dest="vars_file", help="YAML file with 'vars' overrides")
    ap.add_argument("--var", action="append", default=[], help="Single override key=value (repeatable)")
    ap.add_argument("--workers", type=int, default=None,
                    help="Parallelism budget for the run (files at a time for per-file steps, processes inside text steps). Default: the preset\'s `workers` variable, else one per CPU core.")
    ap.add_argument("--out-manifest", dest="out_manifest", default=None,
                    help="Run manifest (JSON). Default: ./run_manifest.json")
    ap.add_argument("--quiet", dest="verbose", action="store_false", default=True,
                    help="Suppress the step-by-step chatter, including each step's "
                         "own output (transcription prints a line per segment). The "
                         "final summary and the exit code are unaffected")
    ap.add_argument("--no-log", dest="write_log", action="store_false", default=True,
                    help="Do not write a run log. By default every run leaves one in "
                         "logs/ beside the manifest, holding everything the run "
                         "printed and the full traceback for anything that failed. "
                         "TATERS_RUNLOG=0 does the same thing for every run")

    # discovery / docs helpers
    ap.add_argument("--list-presets", action="store_true",
                    help="List all discovered presets and exit")
    ap.add_argument("--describe-preset", metavar="NAME",
                    help="Show metadata for a preset (by id or filename) and exit")

    args = ap.parse_args()

    # early-exit helpers (no preset required)
    if args.list_presets:
        _cmd_list_presets()
        sys.exit(0)

    if args.describe_preset:
        _cmd_describe_preset(args.describe_preset)
        sys.exit(0)

    # now we enforce that one of --preset/--preset-file is present
    if not (args.preset or args.preset_file):
        ap.error("one of --preset or --preset-file is required "
                 "unless using --list-presets or --describe-preset")

    # ---------------------------
    # Load preset and vars first
    # ---------------------------
    preset = load_preset_by_name(args.preset) if args.preset else load_yaml_file(Path(args.preset_file))
    steps: List[dict] = preset.get("steps", []) or []
    if not steps:
        raise ValueError("Preset has no steps")

    vars_ctx: Dict[str, Any] = dict(preset.get("vars", {}) or {})
    if args.vars_file:
        vars_ctx = merge_vars(vars_ctx, load_yaml_file(Path(args.vars_file)))
    vars_ctx = merge_vars(vars_ctx, parse_var_overrides(args.var))

    # checked here, not in run_preset(), so the message names the CLI flag the
    # user actually typed rather than the library parameter behind it.
    if not args.root_dir and any(st.get("scope", "item") == "item" for st in steps):
        raise ValueError("--root_dir is required because this preset contains ITEM-scoped steps.")

    # the command line is reconstructed rather than passed through: it is what
    # the log header and the manifest both record as "how this was run".
    log = runlog.RunLog(enabled=args.write_log)
    manifest = run_preset(
        preset,
        root_dir=args.root_dir,
        file_type=args.file_type,
        vars_ctx=vars_ctx,
        workers=args.workers,
        out_manifest=args.out_manifest,
        preset_name=args.preset or str(args.preset_file),
        verbose=args.verbose,
        command=" ".join([Path(sys.argv[0]).name] + sys.argv[1:]),
        run_log=log,
    )

    # named unconditionally, like the summary below it: the run that needs this
    # file is the run whose output already scrolled past.
    if manifest.get("log"):
        print(f"[pipeline] Full log: {manifest['log']}")

    # not gated on --quiet, on purpose. this is the outcome, not the chatter:
    # a quiet run still has to say whether it worked, and which files didn't.
    if not summarize_manifest(manifest):
        # exit non-zero so scripts, schedulers and CI can tell a partial run
        # from a clean one. the manifest has the detail.
        sys.exit(1)

merge_vars

merge_vars(base, overlay)

Shallow-merge two variable dictionaries.

Later sources of variables (e.g., --vars-file, then repeated --var overrides) should replace keys from earlier sources. This helper applies a simple dict.update(...) and returns a new dictionary.

Parameters:

Name	Type	Description	Default
base	dict	The starting dictionary of variables.	required
overlay	dict	The dictionary whose keys override entries in base.	required

Returns:

Type	Description
dict	A new dictionary with merged keys/values.

Source code in src\taters\pipelines\run_pipeline.py

def merge_vars(base: dict, overlay: dict) -> dict:
    """
    Shallow-merge two variable dictionaries.

    Later sources of variables (e.g., `--vars-file`, then repeated `--var`
    overrides) should replace keys from earlier sources. This helper
    applies a simple `dict.update(...)` and returns a new dictionary.

    Parameters
    ----------
    base : dict
        The starting dictionary of variables.
    overlay : dict
        The dictionary whose keys override entries in `base`.

    Returns
    -------
    dict
        A new dictionary with merged keys/values.
    """
    out = dict(base or {})
    out.update(overlay or {})
    return out

parse_var_overrides

parse_var_overrides(pairs)

Parse --var key=value CLI overrides into typed Python values.

Typing rules: - "true"/"false" (case-insensitive) → bool - "null"/"none" (case-insensitive) → None - integer or float strings → numeric - all else → raw string

Parameters:

Name	Type	Description	Default
pairs	List[str]	CLI arguments of the form ["k1=v1", "k2=v2", ...].	required

Returns:

Type	Description
dict	Mapping from variable name to parsed value.

Raises:

Type	Description
ValueError	If any entry does not contain an '=' separator.

Source code in src\taters\pipelines\run_pipeline.py

def parse_var_overrides(pairs: List[str]) -> dict:
    """
    Parse `--var key=value` CLI overrides into typed Python values.

    Typing rules:
      - "true"/"false" (case-insensitive) → bool
      - "null"/"none" (case-insensitive)  → None
      - integer or float strings → numeric
      - all else → raw string

    Parameters
    ----------
    pairs : List[str]
        CLI arguments of the form `["k1=v1", "k2=v2", ...]`.

    Returns
    -------
    dict
        Mapping from variable name to parsed value.

    Raises
    ------
    ValueError
        If any entry does not contain an '=' separator.
    """
    out: Dict[str, Any] = {}
    for s in pairs:
        if "=" not in s:
            raise ValueError(f"--var expects key=value, got: {s}")
        k, v = s.split("=", 1)
        vs = v.strip()
        if vs.lower() in {"true", "false"}:
            out[k] = (vs.lower() == "true")
        elif vs.lower() in {"null", "none"}:
            out[k] = None
        else:
            try:
                out[k] = float(vs) if "." in vs else int(vs)
            except Exception:
                out[k] = v
    return out

render_value

render_value(
    val, *, item_ctx, globals_ctx, vars_ctx, input_path
)

Render templating expressions within a value (str, list, or dict).

Behavior

• Dicts/lists/tuples: render recursively.
• If a string is exactly one template token (e.g., "{{var:text_cols}}"), return the native value of that expression (list, int, bool, ...).
• Otherwise, perform string substitution for every {{...}} occurrence and return the resulting string.

Resolution rules (summary)

• {{input}} / {{cwd}}
• {{var:key}}
• {{global.path}} (explicit globals)
• {{pick:name.path}} → search item, then globals
• {{name}} → bare name; search item, then globals

Source code in src\taters\pipelines\run_pipeline.py

def render_value(
    val: Any,
    *,
    item_ctx: dict,
    globals_ctx: dict,
    vars_ctx: dict,
    input_path: Path
) -> Any:
    """
    Render templating expressions within a value (str, list, or dict).

    Behavior
    --------
      - Dicts/lists/tuples: render recursively.
      - If a string is exactly one template token (e.g., "{{var:text_cols}}"),
        return the *native* value of that expression (list, int, bool, ...).
      - Otherwise, perform string substitution for every {{...}} occurrence and
        return the resulting string.

    Resolution rules (summary)
    --------------------------
      - {{input}} / {{cwd}}
      - {{var:key}}
      - {{global.path}} (explicit globals)
      - {{pick:name.path}}  → search item, then globals
      - {{name}}            → bare name; search item, then globals
    """
    if isinstance(val, dict):
        return {
            k: render_value(
                v,
                item_ctx=item_ctx,
                globals_ctx=globals_ctx,
                vars_ctx=vars_ctx,
                input_path=input_path,
            )
            for k, v in val.items()
        }
    if isinstance(val, (list, tuple)):
        return [
            render_value(
                v,
                item_ctx=item_ctx,
                globals_ctx=globals_ctx,
                vars_ctx=vars_ctx,
                input_path=input_path,
            )
            for v in val
        ]
    if not isinstance(val, str):
        return val

    # the entire string is one template → hand back the native type
    m = _VAR_RE.fullmatch(val.strip())
    if m:
        return _eval_expr(
            m.group(1),
            item_ctx=item_ctx,
            globals_ctx=globals_ctx,
            vars_ctx=vars_ctx,
            input_path=input_path,
        )

    # otherwise we substitute each token as a string
    def _subst(match: re.Match) -> str:
        expr = match.group(1)
        v = _eval_expr(
            expr,
            item_ctx=item_ctx,
            globals_ctx=globals_ctx,
            vars_ctx=vars_ctx,
            input_path=input_path,
        )
        return str(v)

    return _VAR_RE.sub(_subst, val)

resolve_call

resolve_call(call_name, potato)

Resolve a call target from a preset step into an actual callable.

Supported forms

1) Taters instance methods (recommended): - "potato.audio.convert_to_wav" - "potato.text.analyze_with_dictionaries" The function is resolved via attribute chaining on a single Taters() instance created for the whole run.

2) Dotted import paths: - "package.module:function" - "package.module.func" - "package.module.Class.method" The target is imported and attributes are resolved. The final target must be callable.

Parameters:

Name	Type	Description	Default
call_name	str	Call string from the preset step's call: field.	required
potato	Taters	The shared Taters instance for resolving "potato.*" calls.	required

Returns:

Type	Description
Callable	The function/object that will be invoked for the step.

Raises:

Type	Description
(AttributeError, KeyError, TypeError)	If the target cannot be resolved or is not callable.

Source code in src\taters\pipelines\run_pipeline.py

def resolve_call(call_name: str, potato: Taters):
    """
    Resolve a call target from a preset step into an actual callable.

    Supported forms
    ---------------
    1) Taters instance methods (recommended):
       - `"potato.audio.convert_to_wav"`
       - `"potato.text.analyze_with_dictionaries"`
       The function is resolved via attribute chaining on a single
       `Taters()` instance created for the whole run.

    2) Dotted import paths:
       - `"package.module:function"`
       - `"package.module.func"`
       - `"package.module.Class.method"`
       The target is imported and attributes are resolved. The final target
       must be callable.

    Parameters
    ----------
    call_name : str
        Call string from the preset step's `call:` field.
    potato : Taters
        The shared `Taters` instance for resolving `"potato.*"` calls.

    Returns
    -------
    Callable
        The function/object that will be invoked for the step.

    Raises
    ------
    AttributeError, KeyError, TypeError
        If the target cannot be resolved or is not callable.
    """
    if call_name.startswith("potato."):
        obj: Any = potato
        for part in call_name.split(".")[1:]:
            if not hasattr(obj, part):
                raise AttributeError(f"{call_name}: '{part}' not found on {obj}")
            obj = getattr(obj, part)
        if not callable(obj):
            raise TypeError(f"{call_name} is not callable")
        return obj

    # allow dotted import paths
    # we support both "pkg.mod:func" and "pkg.mod.func"
    mod_path, sep, tail = call_name.partition(":")
    if not sep:
        # split at last dot for function
        parts = call_name.rsplit(".", 1)
        if len(parts) == 2:
            mod_path, tail = parts
        else:
            raise KeyError(f"Cannot resolve call target: {call_name}")
    module = importlib.import_module(mod_path)
    target = module
    for attr in tail.split("."):
        if not hasattr(target, attr):
            raise AttributeError(f"{call_name}: '{attr}' not found in {target}")
        target = getattr(target, attr)
    if not callable(target):
        raise TypeError(f"{call_name} resolved to non-callable: {target}")
    return target

resolve_preset_path

resolve_preset_path(name)

Find a preset file by its meta.id or filename stem.

Searches every directory returned by _get_preset_dirs() — the built-in taters/pipelines/presets/ folder first, then ./pipelines relative to the current working directory — so anything shown by --list-presets can also be loaded with --preset.

Parameters:

Name	Type	Description	Default
name	str	Preset meta.id or filename stem (with or without a .yaml/.yml extension).	required

Returns:

Type	Description
Path	Path to the matching preset file.

Raises:

Type	Description
FileNotFoundError	If no preset matches, listing the presets that are available.

Source code in src\taters\pipelines\run_pipeline.py

def resolve_preset_path(name: str) -> Path:
    """
    Find a preset file by its `meta.id` or filename stem.

    Searches every directory returned by `_get_preset_dirs()` — the built-in
    `taters/pipelines/presets/` folder first, then `./pipelines` relative to the
    current working directory — so anything shown by `--list-presets` can also
    be loaded with `--preset`.

    Parameters
    ----------
    name : str
        Preset `meta.id` or filename stem (with or without a `.yaml`/`.yml`
        extension).

    Returns
    -------
    Path
        Path to the matching preset file.

    Raises
    ------
    FileNotFoundError
        If no preset matches, listing the presets that *are* available.
    """
    # drop a .yaml/.yml suffix if the caller included one (".yml" is four
    # characters, ".yaml" is five — hence Path.stem rather than slicing).
    wanted = Path(name).stem if Path(name).suffix.lower() in {".yaml", ".yml"} else name
    available: List[str] = []
    for p in _iter_presets():
        meta = _load_preset_meta(p)
        available.append(meta["id"])
        if wanted in {meta["id"], p.stem}:
            return p
    searched = ", ".join(str(d) for d in _get_preset_dirs())
    known = ", ".join(sorted(set(available))) or "(none)"
    raise FileNotFoundError(
        f"Preset not found: {name}\nSearched: {searched}\nAvailable presets: {known}"
    )

run_global_step

run_global_step(
    *,
    step,
    potato,
    globals_ctx,
    vars_ctx,
    manifest_path,
    on_progress=None,
    quiet=False,
    workers=None
)

Execute a single GLOBAL-scoped step (runs once per pipeline).

Differences from ITEM steps

• The templating item_ctx is empty.
• The run manifest path is injected into vars as run_manifest, so presets can reference it in GLOBAL stages.
• On success, any values from save_as: are merged into the globals artifact map.

Parameters:

Name	Type	Description	Default
step	dict	The step definition block from the preset.	required
potato	Taters	Shared Taters instance used to call potato.* targets.	required
globals_ctx	Dict[str, Any]	Accumulated global artifacts (readable by later steps).	required
vars_ctx	Dict[str, Any]	Merged variables.	required
manifest_path	Path	Path where the JSON run manifest is (or will be) saved.	required

Returns:

Type	Description
Tuple[str, Dict[str, Any], Dict[str, Any]]	A tuple (status, new_globals, err) mirroring the ITEM step shape.

Source code in src\taters\pipelines\run_pipeline.py

def run_global_step(
    *, step: dict, potato: Taters, globals_ctx: Dict[str, Any], vars_ctx: Dict[str, Any],
    manifest_path: Path, on_progress: Optional[Callable[[int, int], None]] = None,
    quiet: bool = False, workers: Optional[int] = None
) -> Tuple[str, Dict[str, Any], Dict[str, Any]]:
    """
    Execute a single GLOBAL-scoped step (runs once per pipeline).

    Differences from ITEM steps
    ---------------------------
    - The templating `item_ctx` is empty.
    - The run manifest path is injected into `vars` as `run_manifest`,
      so presets can reference it in GLOBAL stages.
    - On success, any values from `save_as:` are merged into the `globals`
      artifact map.

    Parameters
    ----------
    step : dict
        The step definition block from the preset.
    potato : Taters
        Shared Taters instance used to call `potato.*` targets.
    globals_ctx : Dict[str, Any]
        Accumulated global artifacts (readable by later steps).
    vars_ctx : Dict[str, Any]
        Merged variables.
    manifest_path : Path
        Path where the JSON run manifest is (or will be) saved.

    Returns
    -------
    Tuple[str, Dict[str, Any], Dict[str, Any]]
        A tuple `(status, new_globals, err)` mirroring the ITEM step shape.
    """
    call = step["call"]
    params = step.get("with", {})

    # expose the manifest path via vars
    vars_aug = dict(vars_ctx)
    vars_aug["run_manifest"] = str(manifest_path)

    # --- templating can fail too (e.g., referencing a global that wasn't saved yet)
    try:
        rendered = render_value(
            params,
            item_ctx={},  # no item context in GLOBAL
            globals_ctx=globals_ctx,
            vars_ctx=vars_aug,
            input_path=manifest_path
        )
    except KeyError as e:
        msg = f"Templating failed (likely missing global artifact): {e}"
        return ("error", {}, {"error": f"{call} failed: {msg}",
                              "trace": runlog.exception_text(e)})
    except Exception as e:
        return ("error", {}, {"error": f"{call} failed during templating: {e}",
                              "trace": runlog.exception_text(e)})


    func = resolve_call(call, potato)

    if step.get("assets"):
        try:
            rendered = _materialize_assets(func, rendered, step["assets"])
        except Exception as e:
            return ("error", {}, {"error": f"{call} failed: {type(e).__name__}: {e}",
                                  "trace": runlog.exception_text(e)})

    # a GLOBAL step is a single call, so from out here we can't count its
    # progress: it's either not started or finished. functions that *can*
    # count themselves say so by declaring an `on_progress` parameter and get
    # one injected here -- and the same signature trick hands `workers` (the
    # run's resolved parallelism budget) to any step that can spend it.
    rendered = _inject_runner_kwargs(func, rendered, on_progress=on_progress,
                                     quiet=quiet, workers=workers)

    try:
        result = func(**rendered)
    except Exception as e:
        # the exception's *type* is part of the message: a bare KeyError
        # renders as just its key ("'1467-'"), which told a user nothing about
        # what happened, let alone where.
        return ("error", {}, {"error": f"{call} failed: {type(e).__name__}: {e}",
                              "trace": runlog.exception_text(e)})

    out: Dict[str, Any] = {}
    if "save_as" in step:
        out[step["save_as"]] = result
    return ("ok", out, {})

run_item_step_for_one_input

run_item_step_for_one_input(
    *,
    step,
    input_path,
    potato,
    item_artifacts,
    globals_ctx,
    vars_ctx,
    on_progress=None,
    quiet=False
)

Execute a single ITEM-scoped step for one input path.

Lifecycle

1) Template the step's with: parameters using render_value(...). 2) Validate any require: keys after templating (fail fast if missing). 3) Resolve the callable (Taters method or import path). 4) Invoke with keyword arguments. 5) If the step specified save_as: <name>, store the return value under that name in the item's artifacts dict.

Parameters:

Name	Type	Description	Default
step	dict	The step definition block from the preset.	required
input_path	Path	The current input file for ITEM scope.	required
potato	Taters	Shared Taters instance used to call potato.* targets.	required
item_artifacts	Dict[str, Any]	The current item's artifact dictionary (mutated across steps).	required
globals_ctx	Dict[str, Any]	Global artifacts (from GLOBAL steps).	required
vars_ctx	Dict[str, Any]	Merged variables.	required
on_progress	callable	Progress sink for this file, passed on to the step function when it declares one. An ITEM step is counted from outside -- files finished out of files found -- but that says nothing while a single file is running, and transcription is routinely minutes per file. Without this a stalled step and a working one look identical until the first file lands.	None
quiet	bool	Suppress the step function's own printing. See :func:_inject_runner_kwargs.	False

Returns:

Type	Description
Tuple[str, Dict[str, Any], Dict[str, Any]]	A tuple (status, new_artifacts, err) where: - status is "ok" or "error". - new_artifacts is a (possibly empty) dict of artifacts to merge. - err contains an "error" message on failure.

Source code in src\taters\pipelines\run_pipeline.py

def run_item_step_for_one_input(
    *, step: dict, input_path: Path, potato: Taters, item_artifacts: Dict[str, Any],
    globals_ctx: Dict[str, Any], vars_ctx: Dict[str, Any],
    on_progress: Optional[Callable[..., None]] = None, quiet: bool = False
) -> Tuple[str, Dict[str, Any], Dict[str, Any]]:
    """
    Execute a single ITEM-scoped step for one input path.

    Lifecycle
    ---------
    1) Template the step's `with:` parameters using `render_value(...)`.
    2) Validate any `require:` keys after templating (fail fast if missing).
    3) Resolve the callable (Taters method or import path).
    4) Invoke with keyword arguments.
    5) If the step specified `save_as: <name>`, store the return value under
       that name in the item's `artifacts` dict.

    Parameters
    ----------
    step : dict
        The step definition block from the preset.
    input_path : Path
        The current input file for ITEM scope.
    potato : Taters
        Shared Taters instance used to call `potato.*` targets.
    item_artifacts : Dict[str, Any]
        The current item's artifact dictionary (mutated across steps).
    globals_ctx : Dict[str, Any]
        Global artifacts (from GLOBAL steps).
    vars_ctx : Dict[str, Any]
        Merged variables.
    on_progress : callable, optional
        Progress sink for *this file*, passed on to the step function when it
        declares one. An ITEM step is counted from outside -- files finished out
        of files found -- but that says nothing while a single file is running,
        and transcription is routinely minutes per file. Without this a stalled
        step and a working one look identical until the first file lands.
    quiet : bool, default False
        Suppress the step function's own printing. See
        :func:`_inject_runner_kwargs`.

    Returns
    -------
    Tuple[str, Dict[str, Any], Dict[str, Any]]
        A tuple `(status, new_artifacts, err)` where:
          - `status` is `"ok"` or `"error"`.
          - `new_artifacts` is a (possibly empty) dict of artifacts to merge.
          - `err` contains an `"error"` message on failure.
    """
    call = step["call"]
    params = step.get("with", {})

    # --- templating can fail (e.g. pick:<artifact>.* when a prior step failed)
    try:
        rendered = render_value(
            params,
            item_ctx=item_artifacts,
            globals_ctx=globals_ctx,
            vars_ctx=vars_ctx,
            input_path=input_path
        )
    except KeyError as e:
        # common case: a previous step failed for this item, so the artifact's missing
        msg = f"Templating failed (likely missing artifact): {e}"
        return ("error", {}, {"error": f"{call} failed: {msg}",
                              "trace": runlog.exception_text(e)})
    except Exception as e:
        return ("error", {}, {"error": f"{call} failed during templating: {e}",
                              "trace": runlog.exception_text(e)})


    # required keys check (post-templating)
    for key in step.get("require", []):
        if key not in rendered or rendered[key] in (None, "", []):
            return ("error", {}, {"error": f"Missing required parameter '{key}' after templating"})

    # --- invoke target
    func = resolve_call(call, potato)
    # workers=1: an ITEM step's parallelism is the fan-out itself; a pool
    # inside each fanned-out call would multiply into cores-squared.
    rendered = _inject_runner_kwargs(func, rendered, on_progress=on_progress,
                                     quiet=quiet, workers=1)
    try:
        result = func(**rendered)
    except Exception as e:
        # the exception's *type* is part of the message: a bare KeyError
        # renders as just its key ("'1467-'"), which told a user nothing about
        # what happened, let alone where.
        return ("error", {}, {"error": f"{call} failed: {type(e).__name__}: {e}",
                              "trace": runlog.exception_text(e)})

    out: Dict[str, Any] = {}
    if "save_as" in step:
        out[step["save_as"]] = result
    return ("ok", out, {})

run_preset

run_preset(
    preset,
    *,
    root_dir=None,
    file_type="any",
    vars_ctx=None,
    workers=None,
    out_manifest=None,
    preset_name=None,
    on_event=None,
    verbose=True,
    work_dir=None,
    command=None,
    run_log=None
)

Run a loaded preset and return its manifest.

This is the engine main() wraps. It is separate so that callers other than the command line -- the setup wizard in :mod:taters.ui.wizard, and anything else that already holds a preset dict -- can run a pipeline without building an argv and without the process exiting underneath them.

Parameters:

Name	Type	Description	Default
preset	dict	A loaded preset: steps, and optionally vars and meta.	required
root_dir	Path | str | None	Folder to scan for inputs. Required when the preset has any ITEM-scoped steps; ignored when it does not.	None
file_type	('audio', 'video', 'any')	Extension filter for discovery.	"audio","video","any"
vars_ctx	dict	The variable context, already merged. When omitted, the preset's own vars block is used as-is.	None
workers	int	Default concurrency for ITEM steps. A step's own workers: wins.	4
out_manifest	Path | str | None	Where to write the run manifest. Defaults to ./run_manifest.json.	None
preset_name	str	Recorded in the manifest for provenance.	None
on_event	callable	Called as on_event(name, **payload) as the run progresses, with name one of run_start, step_start, item_done, step_done, run_done. Exceptions raised by the callback are swallowed: a UI bug must not take down a run that may be hours in.	None
verbose	bool	Print progress to stdout. Set False when on_event is doing the reporting instead -- this silences the runner's own lines and is passed down to every step function that accepts a verbose argument. Steps print by default, which is right at a shell prompt and ruinous underneath a live display: text written into the region a display owns corrupts it. Note the asymmetry -- False is pushed down, True is not, so a step that is quiet by default stays quiet.	True

Returns:

Type	Description
dict	The run manifest. Check manifest["errors"] and each item's "status" to tell a clean run from a partial one -- unlike main(), this function never calls sys.exit.

Source code in src\taters\pipelines\run_pipeline.py

def run_preset(
    preset: dict,
    *,
    root_dir: Path | str | None = None,
    file_type: str = "any",
    vars_ctx: Dict[str, Any] | None = None,
    workers: Optional[int] = None,
    out_manifest: Path | str | None = None,
    preset_name: str | None = None,
    on_event=None,
    verbose: bool = True,
    work_dir: Optional[Path] = None,
    command: Optional[str] = None,
    run_log: Optional[runlog.RunLog] = None,
) -> Dict[str, Any]:
    """
    Run a loaded preset and return its manifest.

    This is the engine `main()` wraps. It is separate so that callers other
    than the command line -- the setup wizard in :mod:`taters.ui.wizard`, and
    anything else that already holds a preset dict -- can run a pipeline
    without building an argv and without the process exiting underneath them.

    Parameters
    ----------
    preset : dict
        A loaded preset: `steps`, and optionally `vars` and `meta`.
    root_dir : Path | str | None, optional
        Folder to scan for inputs. Required when the preset has any ITEM-scoped
        steps; ignored when it does not.
    file_type : {"audio","video","any"}, default "any"
        Extension filter for discovery.
    vars_ctx : dict, optional
        The variable context, already merged. When omitted, the preset's own
        `vars` block is used as-is.
    workers : int, default 4
        Default concurrency for ITEM steps. A step's own `workers:` wins.
    out_manifest : Path | str | None, optional
        Where to write the run manifest. Defaults to `./run_manifest.json`.
    preset_name : str, optional
        Recorded in the manifest for provenance.
    on_event : callable, optional
        Called as ``on_event(name, **payload)`` as the run progresses, with
        `name` one of `run_start`, `step_start`, `item_done`, `step_done`,
        `run_done`. Exceptions raised by the callback are swallowed: a UI bug
        must not take down a run that may be hours in.
    verbose : bool, default True
        Print progress to stdout. Set False when `on_event` is doing the
        reporting instead -- this silences the runner's own lines *and* is
        passed down to every step function that accepts a `verbose` argument.
        Steps print by default, which is right at a shell prompt and ruinous
        underneath a live display: text written into the region a display owns
        corrupts it. Note the asymmetry -- False is pushed down, True is not,
        so a step that is quiet by default stays quiet.

    Returns
    -------
    dict
        The run manifest. Check `manifest["errors"]` and each item's
        `"status"` to tell a clean run from a partial one -- unlike `main()`,
        this function never calls `sys.exit`.
    """
    # relative output paths in a preset ("features/readability.csv") resolve
    # against the process's working directory. `work_dir` moves that for the
    # duration of the run, and that's what lets a pipeline own a folder without
    # every recipe having to spell out an absolute path.
    #
    # we restore it in a finally: leaving the interpreter somewhere else
    # afterwards would silently relocate everything the caller does next.
    previous_cwd = None
    if work_dir is not None:
        work_dir = Path(work_dir)
        work_dir.mkdir(parents=True, exist_ok=True)
        # resolved BEFORE the chdir. these paths were given relative to where
        # the caller stood; resolving them after moving re-anchored them under
        # work_dir, so `taters --dir data` wrote its manifest to the phantom
        # data/mypipe/data/mypipe/run_manifest.json while the finish screen
        # pointed at the real path -- which didn't exist.
        if out_manifest is not None:
            out_manifest = Path(out_manifest).resolve()
        if root_dir is not None:
            root_dir = Path(root_dir).resolve()
        previous_cwd = Path.cwd()

    # a log nobody asked for is a no-op object rather than a branch at every
    # call site below.
    log = run_log if run_log is not None else runlog.RunLog(enabled=False)

    # opened before the chdir, for exactly the reason the paths above are
    # resolved before it: a relative path means relative to where the caller
    # stood, and moving first would re-anchor it under work_dir.
    if log.active and log.path is None:
        if work_dir is not None:
            base = Path(work_dir)
        elif out_manifest is not None:
            base = Path(out_manifest).parent
        else:
            base = Path.cwd()
        log.open_run(base, preset=preset_name, workers=workers, command=command)

    if previous_cwd is not None:
        os.chdir(work_dir)
    try:
        # the capture wraps the whole body rather than just the step loop:
        # input discovery prints too, and a step that cannot find its files is
        # a failure worth having the reason for.
        with log.capture_streams():
            return _run_preset(
                preset, root_dir=root_dir, file_type=file_type, vars_ctx=vars_ctx,
                workers=workers, out_manifest=out_manifest, preset_name=preset_name,
                on_event=on_event, verbose=verbose, command=command, run_log=log,
            )
    finally:
        log.close_run()
        if previous_cwd is not None:
            os.chdir(previous_cwd)

summarize_manifest

summarize_manifest(manifest, *, verbose=True)

Report on a finished run and say whether it was clean.

Parameters:

Name	Type	Description	Default
manifest	dict	As returned by :func:run_preset.	required
verbose	bool	Print the summary. When False, only the return value is produced.	True

Returns:

Type	Description
bool	True when every item succeeded and no global step failed.

Source code in src\taters\pipelines\run_pipeline.py

def summarize_manifest(manifest: Dict[str, Any], *, verbose: bool = True) -> bool:
    """
    Report on a finished run and say whether it was clean.

    Parameters
    ----------
    manifest : dict
        As returned by :func:`run_preset`.
    verbose : bool, default True
        Print the summary. When False, only the return value is produced.

    Returns
    -------
    bool
        True when every item succeeded and no global step failed.
    """
    failed_items = [itm for itm in manifest["items"] if itm.get("status") == "error"]
    ok_items = [itm for itm in manifest["items"] if itm.get("status") == "ok"]
    global_errors = manifest["errors"]

    if verbose:
        if manifest["items"]:
            print(f"[pipeline] Items: {len(ok_items)} ok, {len(failed_items)} failed")
        if failed_items:
            for itm in failed_items[:10]:
                first = itm["errors"][0] if itm["errors"] else "unknown error"
                print(f"[pipeline]   FAILED {itm['input']}: {first}")
            if len(failed_items) > 10:
                print(f"[pipeline]   ...and {len(failed_items) - 10} more (see the manifest)")
        for err in global_errors:
            print(f"[pipeline] GLOBAL ERROR: {err}")

    return not (global_errors or failed_items)