xil-status — make-style staleness checker for episode pipeline artifacts.
An episode flows through an ordered chain of file "waypoints", each produced
from the previous one::
Google Drive .gdoc → scripts/*.md → parsed/{slug}/parsed_{tag}.json
→ stems/{slug}/{tag}/*.mp3 → daw/{slug}/{tag}/*.wav → masters/{slug}/{tag}_*.mp3
This utility walks that chain and reports, per stage, whether the outputs are
up to date with their inputs — like make deciding whether a target needs
rebuilding. A stage is:
- MISSING — no output files exist yet,
- STALE — the newest input is newer than the newest output
(
max(inputs) > max(outputs)), i.e. the stage has not run since its
input last changed,
- OK — the stage ran at or after its newest input.
The newest-output rule (rather than strict make's oldest-output) fits this
pipeline's incremental, content-hash-dedup builds: a produce/daw run only
rewrites the outputs that actually changed, so older reused outputs legitimately
coexist with fresh ones. What matters is whether the stage ran after its input
changed, which the newest output captures.
Nothing is ever rebuilt: the tool only reports, then prints the exact xil
commands needed to refresh any stale/missing stages.
The Google Drive source (.gdoc) is included as the chain root when found. Its
directory defaults to $XIL_GDOC_DIR or /mnt/i/My Drive and can be
overridden with --gdoc-dir; a missing mount warns rather than fails.
Exit codes: 0 when every stage is OK, 1 when any stage is STALE or MISSING,
2 on a usage/resolution error.
Usage::
xil status --episode S01E01
xil status S01E01 --show "Night Owls"
xil status --all
xil status --episode S01E01 --json
logger
module-attribute
logger = get_logger(__name__)
StageStatus
dataclass
Freshness result for one pipeline stage.
Source code in src/xil_pipeline/XILU019_episode_status.py
| @dataclass
class StageStatus:
"""Freshness result for one pipeline stage."""
name: str
status: str
newest_input: float | None # mtime epoch, or None
newest_output: float | None
oldest_output: float | None # the value the STALE decision compares against
output_count: int
note: str = ""
refresh: str = "" # suggested xil command (empty if none / OK)
inputs_present: bool = True
output_files: list = None # type: ignore[assignment] # populated by _evaluate_stage
def __post_init__(self) -> None:
if self.output_files is None:
self.output_files = []
|
status
instance-attribute
newest_input: float | None
newest_output
instance-attribute
newest_output: float | None
oldest_output
instance-attribute
oldest_output: float | None
output_count
instance-attribute
note
class-attribute
instance-attribute
refresh
class-attribute
instance-attribute
inputs_present: bool = True
output_files
class-attribute
instance-attribute
output_files: list = None
__init__
__init__(name: str, status: str, newest_input: float | None, newest_output: float | None, oldest_output: float | None, output_count: int, note: str = '', refresh: str = '', inputs_present: bool = True, output_files: list = None) -> None
__post_init__
Source code in src/xil_pipeline/XILU019_episode_status.py
| def __post_init__(self) -> None:
if self.output_files is None:
self.output_files = []
|
evaluate_episode
evaluate_episode(slug: str, tag: str, gdoc_dir: Path, *, include_source: bool = True) -> list[StageStatus]
Build and evaluate the full waypoint chain for one episode.
When include_source is False the Google Drive mount is not probed and
the source / script stages are omitted. Pass this from callers that
only need the pipeline stages (parsed → stems → daw → master), e.g. the GUI
episode table, to avoid slow network-mount probes on every refresh.
Source code in src/xil_pipeline/XILU019_episode_status.py
| def evaluate_episode(
slug: str, tag: str, gdoc_dir: Path, *, include_source: bool = True
) -> list[StageStatus]:
"""Build and evaluate the full waypoint chain for one episode.
When *include_source* is ``False`` the Google Drive mount is not probed and
the ``source`` / ``script`` stages are omitted. Pass this from callers that
only need the pipeline stages (parsed → stems → daw → master), e.g. the GUI
episode table, to avoid slow network-mount probes on every refresh.
"""
root = get_workspace_root()
paths = derive_paths(slug, tag)
gdocs = _gdoc_files(gdoc_dir, tag) if include_source else []
scripts = _script_files(root, tag)
parsed = [Path(paths["parsed"])]
stems = _glob_in(paths["stems"], "*.mp3")
stems_manifest = _glob_in(paths["stems"], "*_stem_manifest.json")
daw = _glob_in(paths["daw"], "*.wav")
masters = _master_files(root, slug, tag)
# script refresh has no CLI equivalent — it is imported via xil-gui.
script_refresh = "(re-import the production doc in xil-gui)"
# Suggest the most recently modified script: with multiple drafts on disk
# (e.g. *_v1, *_v2, *_v3), the newest is the one that triggered staleness
# and the one the user most likely wants to (re)parse.
newest_script = max(scripts, key=lambda p: p.stat().st_mtime) if scripts else None
parse_refresh = (
f"xil parse {newest_script.relative_to(root)} --episode {tag}"
if newest_script
else f"xil parse <script> --episode {tag}"
)
# The manifest is included in the stems stage's OUTPUTS (not daw's inputs):
# the producer rewrites it on every run, including no-op runs where all stems
# are dedup-reused, so it is the only signal that advances when re-producing
# an unchanged episode — without it, a stems stage made stale by a re-parse
# could never be cleared. It must NOT count as daw's INPUT, though: daw
# consumes the stem audio, so a no-op manifest bump (produce re-run that
# changed nothing) must not invalidate an up-to-date daw. Hence daw is judged
# against the stem MP3s only.
stems_outputs = stems + stems_manifest
stages: list[StageStatus] = []
if include_source:
stages += [
_evaluate_stage("source", [], gdocs, ""),
_evaluate_stage("script", gdocs, scripts, script_refresh),
]
# The source stage is informational: a missing/empty gdoc dir is not a failure.
src = stages[0]
if not gdocs:
src.status = _NONE
try:
_gdoc_is_dir = gdoc_dir.is_dir()
except OSError:
_gdoc_is_dir = False
src.note = "no gdoc dir" if not _gdoc_is_dir else "no source doc"
src.refresh = ""
sfx_cfg = [Path(paths["sfx"])] if Path(paths["sfx"]).exists() else []
stages += [
_evaluate_stage("parsed", scripts, parsed, parse_refresh),
_evaluate_stage(
"stems", parsed, stems_outputs, f"xil produce --episode {tag}",
count=len(stems),
),
_evaluate_stage("daw", stems + sfx_cfg, daw, f"xil daw --episode {tag}"),
_evaluate_stage("master", daw, masters, f"xil master --episode {tag}"),
]
return stages
|
get_parser
get_parser() -> argparse.ArgumentParser
Return the argument parser for xil-status.
Source code in src/xil_pipeline/XILU019_episode_status.py
| def get_parser() -> argparse.ArgumentParser:
"""Return the argument parser for xil-status."""
parser = argparse.ArgumentParser(
prog="xil-status",
description=(
"Make-style staleness checker for episode pipeline artifacts. "
"Reports, per stage, whether outputs are up to date with their inputs, "
"and prints the xil commands needed to refresh anything stale. "
"Nothing is rebuilt."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" xil status --episode S01E01\n"
' xil status S01E01 --show "Night Owls"\n'
" xil status --all\n"
" xil status --episode S01E01 --json\n"
),
)
parser.add_argument(
"episode",
nargs="?",
metavar="TAG",
help="Episode tag to check (e.g. S01E01). Omit with --all.",
)
parser.add_argument(
"--episode", "-e",
dest="episode_flag",
default=None,
metavar="TAG",
help="Episode tag (alternative to the positional argument)",
)
parser.add_argument(
"--show", "-s",
default=None,
metavar="SHOW",
help="Show name or slug (default: resolved from project.json / XIL_PROJECTROOT)",
)
parser.add_argument(
"--all", "-a",
action="store_true",
help="Check every episode of the show (summary row per episode)",
)
parser.add_argument(
"--gdoc-dir",
default=_DEFAULT_GDOC_DIR,
metavar="DIR",
help=f"Google Drive dir holding the source .gdoc (default: {_DEFAULT_GDOC_DIR})",
)
parser.add_argument(
"--json",
action="store_true",
help="Emit results as JSON (single-episode mode only)",
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="List each output file with its mtime below the stage row",
)
return parser
|
main
CLI entry point for episode pipeline staleness checking.
Source code in src/xil_pipeline/XILU019_episode_status.py
| def main() -> None:
"""CLI entry point for episode pipeline staleness checking."""
configure_logging()
args = get_parser().parse_args()
slug = resolve_slug(args.show)
gdoc_dir = Path(args.gdoc_dir)
try:
gdoc_available = gdoc_dir.is_dir()
except OSError:
gdoc_available = False
if not gdoc_available:
logger.warning(f"Google Drive dir not available: {gdoc_dir} — skipping source check.")
if args.all:
if args.json:
logger.error("--json is not supported with --all.")
sys.exit(2)
tags = _discover_tags(slug)
sys.exit(_print_all(slug, tags, gdoc_dir))
tag = args.episode_flag or args.episode
if not tag:
logger.error("Provide an episode tag (e.g. S01E01) or use --all.")
sys.exit(2)
stages = evaluate_episode(slug, tag, gdoc_dir)
if args.json:
_emit_json(slug, tag, stages)
else:
_print_episode(slug, tag, stages, verbose=args.verbose)
sys.exit(0 if _worst(stages) == _OK else 1)
|