gitprov/gnommo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adding the state system

Browse Source
This commit is contained in:
gitprov
2026-07-15 08:31:33 +02:00
parent 7f9f8e3395
commit c6efbaf2ed
2 changed files with 582 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gnommo/cli.py
+423 -19
View File
@@ -39,6 +39,8 @@ Examples:
gnommo -p video1 import Generate slides.json from images
gnommo -p video1 pre Preprocess videos (chroma key, etc.)
gnommo -p video1 clear Delete preprocessed outputs so preprocess re-runs them
gnommo -p video1 prune Remove unused entries from videos.json/audio.json/narration.json
gnommo -p video1 prune --dry-run Preview which manifest entries would be removed
gnommo -p video1 stitch --res tiny -f Fast stitch with new begin/end values
gnommo -p video1 trim Auto-detect silence and set skip/take in narration.json
gnommo -p video1 trim --force Redo trim even for segments that already have skip/take
@@ -113,6 +115,7 @@ Examples:
"transcode",
"pexels",
"clear",
"prune",
"new",
],
help="Action to perform (default: render)",
@@ -277,6 +280,8 @@ Examples:
return cmd_new(project_path, args.verbose)
elif action == "clear":
return cmd_clear(project_path, args.verbose)
elif action == "prune":
return cmd_prune(project_path, args.verbose, args.dry_run)
elif action in ("preprocess", "pre"):
return cmd_preprocess(
project_path,
@@ -2164,6 +2169,222 @@ def cmd_clear(project_path: Path, verbose: bool) -> int:
return 0
# =============================================================================
# Prune Command
# =============================================================================
# Marker prefixes that reference a videos.json entry, mapped to the length of the
# prefix (so marker[length:] is the video ID). Kept in sync with validator.py.
_PRUNE_VIDEO_PREFIXES = {
"video:": 6,
"vft:": 4, "vfb:": 4, "vfm:": 4,
"vf2t:": 5, "vf2b:": 5, "vf2m:": 5,
"vst:": 4, "vsb:": 4, "vsm:": 4,
"vftp:": 5, "vfbp:": 5, "vfmp:": 5,
"vf2tp:": 6, "vf2bp:": 6, "vf2mp:": 6,
"vstp:": 5, "vsbp:": 5, "vsmp:": 5,
}
def _detect_json_indent(path: Path, default: int = 2) -> int:
"""Return the indentation width of the first indented line in a JSON file."""
try:
for line in path.read_text(encoding="utf-8").splitlines():
stripped = line.lstrip(" ")
if stripped and stripped != line:
return len(line) - len(stripped)
except OSError:
pass
return default
def _write_json_preserve(path: Path, data: dict) -> None:
"""Rewrite a JSON manifest, preserving its existing indentation style."""
indent = _detect_json_indent(path)
text = json.dumps(data, indent=indent, ensure_ascii=False)
try:
trailing_nl = path.read_text(encoding="utf-8").endswith("\n")
except OSError:
trailing_nl = True
path.write_text(text + ("\n" if trailing_nl else ""), encoding="utf-8")
def _prune_manifest(
path: Path,
keep_ids: set[str],
label: str,
lowercase_keys: bool,
verbose: bool,
dry_run: bool,
) -> int:
"""Remove entries from a videos.json/audio.json manifest whose ID isn't in keep_ids.
Returns the number of entries removed (or that would be removed, in dry-run mode).
"""
if not path.exists():
return 0
try:
data = _read_json(path)
except json.JSONDecodeError as e:
print(f" Skipping {path.name}: invalid JSON ({e})")
return 0
if not isinstance(data, dict):
return 0
to_remove = [
key
for key in data
if (key.lower() if lowercase_keys else key) not in keep_ids
]
if not to_remove:
if verbose:
print(f" {path.name}: all {len(data)} {label} entries in use.")
return 0
print(f" {path.name}: removing {len(to_remove)} unused {label} entr"
f"{'y' if len(to_remove) == 1 else 'ies'}:")
for key in to_remove:
print(f" - {key}")
if not dry_run:
for key in to_remove:
del data[key]
_write_json_preserve(path, data)
return len(to_remove)
def _prune_narration(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Remove narration.json entries whose raw source file is missing from raw_mov/.
Returns the number of entries removed (or that would be removed, in dry-run mode).
"""
narration_dir = project_path / "media" / "narration"
narration_json = narration_dir / "narration.json"
if not narration_json.exists():
return 0
try:
data = _read_json(narration_json)
except json.JSONDecodeError as e:
print(f" Skipping narration.json: invalid JSON ({e})")
return 0
if not isinstance(data, dict):
return 0
raw_dir = narration_dir / "raw_mov"
to_remove = []
for seg_id, entry in data.items():
source_file = entry.get("source_file") if isinstance(entry, dict) else None
if source_file:
exists = (narration_dir / source_file).exists()
else:
# No source_file recorded — look for a matching file in raw_mov/
exists = bool(list(raw_dir.glob(f"{seg_id}.*"))) if raw_dir.exists() else False
if not exists:
to_remove.append(seg_id)
if not to_remove:
if verbose:
print(f" narration.json: all {len(data)} entries have a raw_mov/ source.")
return 0
print(f" narration.json: removing {len(to_remove)} entr"
f"{'y' if len(to_remove) == 1 else 'ies'} with no raw_mov/ source:")
for seg_id in to_remove:
print(f" - {seg_id}")
if not dry_run:
for seg_id in to_remove:
del data[seg_id]
_write_json_preserve(narration_json, data)
return len(to_remove)
def cmd_prune(project_path: Path, verbose: bool, dry_run: bool) -> int:
"""Remove unused entries from videos.json, audio.json, and narration.json.
- videos.json / audio.json: drop entries whose IDs are never referenced by a
marker in manuscript.txt. Videos referenced by project.json (outro sequence,
main_video) are kept even when no marker triggers them.
- narration.json: drop entries whose raw source file is missing from
media/narration/raw_mov/.
Only JSON manifest entries are removed — media files on disk are never touched.
Use --dry-run to preview.
"""
from .parser import parse_manuscript, parse_project_config
suffix = " (dry run)" if dry_run else ""
print(f"Pruning manifests: {project_path.name}{suffix}")
config = parse_project_config(project_path)
# --- Collect IDs referenced by the script (manuscript markers) ---
_, markers, _, _ = parse_manuscript(project_path)
referenced_videos: set[str] = set()
referenced_audio: set[str] = set()
for marker in markers:
prefix = next((p for p in _PRUNE_VIDEO_PREFIXES if marker.startswith(p)), None)
if prefix is not None:
referenced_videos.add(marker[_PRUNE_VIDEO_PREFIXES[prefix]:].lower())
elif marker.startswith("narration:"):
referenced_videos.add(marker[10:].lower())
elif marker.startswith("audio:"):
referenced_audio.add(marker[6:])
elif marker.startswith("A") and len(marker) > 1 and marker[1:].isalnum():
referenced_audio.add(marker[1:])
# Videos referenced by project.json (outro sequence, main video, background)
# are kept even though no manuscript marker triggers them.
def _add_video(value) -> None:
if isinstance(value, str) and value:
referenced_videos.add(value.lower())
elif isinstance(value, list):
for item in value:
_add_video(item)
_add_video(config.outro)
_add_video(config.main_video)
_add_video(config.background)
_add_video(config.background_video)
total_removed = 0
total_removed += _prune_manifest(
project_path / config.videos_path,
referenced_videos,
"video",
lowercase_keys=True,
verbose=verbose,
dry_run=dry_run,
)
total_removed += _prune_manifest(
project_path / config.audio_path,
referenced_audio,
"audio",
lowercase_keys=False,
verbose=verbose,
dry_run=dry_run,
)
total_removed += _prune_narration(project_path, verbose, dry_run)
if total_removed == 0:
print(" Nothing to prune — all manifest entries are in use.")
elif dry_run:
print(f"\n Would remove {total_removed} entr"
f"{'y' if total_removed == 1 else 'ies'}. "
f"Re-run without --dry-run to apply.")
else:
print(f"\n Removed {total_removed} entr"
f"{'y' if total_removed == 1 else 'ies'}.")
return 0
def cmd_preprocess(
project_path: Path,
verbose: bool,
@@ -2271,6 +2492,17 @@ def cmd_preprocess(
existing_narration = _read_json(narration_json_path)
# --- Build segments list ---
# Per-segment staleness: reprocess a segment when its raw source's fingerprint
# differs from the one recorded the last time we produced its output. An
# existing output with no recorded fingerprint is adopted (recorded, not
# reprocessed) so introducing state never triggers a needless re-encode.
from . import state as _state
_stage_key = f"preprocess:{res}"
_prev_fps = _state.get_items(project_path, _stage_key)
_seg_source_fp: dict[str, str] = {} # segment_id -> current source fingerprint
_adopted_fps: dict[str, str] = {} # up-to-date segments to (re)record
segments_to_process: list[tuple[str, _VideoSource]] = []
skipped_count = 0
@@ -2294,10 +2526,18 @@ def cmd_preprocess(
output_base = cache_narration_dir or narration_dir
output_path = output_base / output_file
current_fp = _state.fingerprint_path(source_file, _state.META)
_seg_source_fp[segment_id] = current_fp
if output_path.exists() and not force:
recorded = _prev_fps.get(segment_id)
if recorded is None or recorded == current_fp:
# Up to date (or adopting a pre-existing output into state).
print(f" {segment_id}: output exists, skipping (use --force to reprocess)")
skipped_count += 1
_adopted_fps[segment_id] = current_fp
continue
print(f" {segment_id}: raw source changed since last run — reprocessing")
# Filter: from existing narration.json entry (if explicitly set), else talkinghead
existing_entry = existing_narration.get(segment_id, {})
@@ -2427,6 +2667,16 @@ def cmd_preprocess(
with open(narration_json_path, "w", encoding="utf-8") as f:
json.dump(existing_narration, f, indent=2)
# Record source fingerprints for every segment whose output is now present
# (freshly processed + adopted skips) so future runs can detect raw changes.
_record_fps = dict(_adopted_fps)
for _seg_id, _ in successfully_processed:
_fp = _seg_source_fp.get(_seg_id)
if _fp is not None:
_record_fps[_seg_id] = _fp
if _record_fps:
_state.record_items(project_path, _stage_key, _record_fps)
if successfully_processed:
print(f"\n Updated narration.json ({len(successfully_processed)} segment(s))")
@@ -2634,16 +2884,17 @@ def cmd_trim(
narration_json_path = narration_dir / "narration.json"
raw_data: dict = _read_json(narration_json_path)
# Per-segment staleness: re-trim (and re-transcribe) a segment when its raw
# source's fingerprint differs from the one recorded last time it was trimmed.
from . import state as _state
_prev_trim_fps = _state.get_items(project_path, "trim")
_trim_fps: dict[str, str] = {} # segments to (re)record after the loop
updated = 0
for seg_id in sorted(narration.keys()):
seg = narration[seg_id]
existing = raw_data.get(seg_id, {})
has_explicit = "skip" in existing or "take" in existing
if has_explicit and not force:
print(f" {seg_id}: already trimmed, skipping (use --force to redo)")
continue
# Prefer raw file; fall back to source_file from narration.json
source_path = raw_lookup.get(seg_id)
if source_path is None:
@@ -2652,6 +2903,21 @@ def cmd_trim(
print(f" {seg_id}: source file not found, skipping")
continue
# Has the raw source changed since we last trimmed this segment?
current_fp = _state.fingerprint_path(source_path, _state.META)
recorded_fp = _prev_trim_fps.get(seg_id)
source_changed = recorded_fp is not None and recorded_fp != current_fp
seg_force = force or source_changed
existing = raw_data.get(seg_id, {})
has_explicit = "skip" in existing or "take" in existing
if has_explicit and not seg_force:
print(f" {seg_id}: already trimmed, skipping (use --force to redo)")
_trim_fps[seg_id] = current_fp # adopt/keep current fingerprint
continue
if source_changed:
print(f" {seg_id}: raw source changed since last trim — re-trimming")
slide_range = _parse_segment_slide_range(seg_id) if slide_texts else None
if slide_range is not None:
@@ -2663,7 +2929,7 @@ def cmd_trim(
transcript_path = transcripts_dir / f"{seg_id}.json"
try:
if transcript_path.exists() and not force:
if transcript_path.exists() and not seg_force:
words = load_transcript(transcript_path)
print(f" {seg_id}: loaded cached transcript ({len(words)} words)", end="", flush=True)
else:
@@ -2706,6 +2972,7 @@ def cmd_trim(
raw_data[seg_id]["skip"] = new_skip
raw_data[seg_id]["take"] = new_take
_trim_fps[seg_id] = current_fp
updated += 1
continue
@@ -2737,6 +3004,7 @@ def cmd_trim(
raw_data[seg_id]["skip"] = new_skip
raw_data[seg_id]["take"] = new_take
_trim_fps[seg_id] = current_fp
updated += 1
if updated > 0:
@@ -2746,6 +3014,11 @@ def cmd_trim(
else:
print(f"\n No segments updated")
# Persist per-segment source fingerprints (freshly trimmed + adopted skips)
# so a later standalone or 'all' run can tell whether a raw was re-recorded.
if _trim_fps:
_state.record_items(project_path, "trim", _trim_fps)
return 0
@@ -3147,6 +3420,10 @@ def cmd_stitch(
print(" Run 'gnommo -p <project> import' first to populate narration.json")
return 1
# narration.json (skip/take/order) drives stitch — capture its local path for
# fingerprinting before narration_dir is redirected to a cache/res subdir.
_local_narration_json = narration_dir / "narration.json"
# Get videos_dir for output
if config and config.videos_path:
videos_json_path = project_path / config.videos_path
@@ -3191,10 +3468,29 @@ def cmd_stitch(
stitch_output = videos_dir_out / "narration_combined.mov"
if stitch_output.exists() and not force:
print(f"\n Combined narration exists: {stitch_output.name}")
print(" (use --force to regenerate)")
# Stage-level staleness: skip when narration.json and every processed segment
# are unchanged since the last successful stitch AND the output still exists.
# A changed input (re-trim, reprocessed segment) auto-triggers a regenerate.
from . import state as _state
_stitch_key = f"stitch:{res}"
_stitch_inputs = _state.compute(
[("narration.json", _local_narration_json, _state.HASH)]
+ [
(f"seg:{sid}", narration_dir / narration[sid].source_file, _state.META)
for sid in segment_ids
]
)
_stitch_current = _state.is_current(
project_path, _stitch_key, _stitch_inputs, [stitch_output]
)
if stitch_output.exists() and not force and _stitch_current:
print(f"\n Combined narration up to date: {stitch_output.name}")
print(" (inputs unchanged since last stitch — use --force to regenerate)")
else:
if stitch_output.exists() and not force and not _stitch_current:
print("\n Inputs changed since last stitch — regenerating.")
# Extract loudnorm config from talkinghead filter so stitch uses
# per-project settings instead of hardcoded defaults.
_loudnorm_cfg = None
@@ -3217,6 +3513,20 @@ def cmd_stitch(
if not cache_root:
_import_videos(videos_dir_out, config, verbose)
# Record fingerprints (recompute output META now that it exists) so the
# next run can detect whether inputs changed.
_state.record(
project_path,
_stitch_key,
_state.compute(
[("narration.json", _local_narration_json, _state.HASH)]
+ [
(f"seg:{sid}", narration_dir / narration[sid].source_file, _state.META)
for sid in segment_ids
]
),
)
# Always update the MAIN videos.json (parent of subdir when using low/tiny res)
# Downscaled dirs only affect file paths, not JSON metadata updates
main_videos_dir = (
@@ -4057,10 +4367,50 @@ def cmd_render(
print(generate_ffmpeg_command_string(plan, output_path))
return 0
# Stage-level staleness gate. Only whole-project renders are gated — partial
# (--slides) renders and internal chunk sub-renders always run. The render is
# skipped when the combined narration, manifests, manuscript, transcript and
# slide images are all unchanged since the last successful render of this
# resolution and the output file still exists.
from . import state as _state
_render_gateable = (
not force and slide_range is None and _output_path_override is None
)
def _render_input_specs() -> list:
_slides_json = project_path / config.slides_path.lower()
_slides_dir = _slides_json.parent
specs = [
("videos.json", project_path / config.videos_path, _state.HASH),
("audio.json", project_path / config.audio_path, _state.HASH),
("manuscript.txt", project_path / "manuscript.txt", _state.HASH),
("project.json", project_path / "project.json", _state.HASH),
("slides.json", _slides_json, _state.HASH),
("transcript", transcript_path, _state.HASH),
]
if resolved_combined:
specs.append(("narration_combined", resolved_combined, _state.META))
for _sid, _sdef in slides.items():
specs.append((f"slide:{_sid}", _slides_dir / _sdef.image, _state.META))
return specs
_render_key = f"render:{res}"
if _render_gateable and _state.is_current(
project_path, _render_key, _state.compute(_render_input_specs()), [output_path]
):
print(f"\n[4/4] Output up to date: {output_path}")
print(" (inputs unchanged since last render — use --force to re-render)")
print("\nDone.")
return 0
print("\n[4/4] Rendering...")
render(plan, output_path, verbose=verbose)
print(f" Output: {output_path}")
if _render_gateable:
_state.record(project_path, _render_key, _state.compute(_render_input_specs()))
print("\nDone.")
return 0
@@ -4348,6 +4698,46 @@ def _files_modified_since(root: Path, since: float, pattern: str) -> bool:
return False
def _trim_outputs_current(project_path: Path) -> bool:
"""Return True if the trim stage can be safely skipped for every segment.
A segment is considered resolved when narration.json already records an
explicit skip/take for it, or a cached Whisper transcript exists at
narration/transcripts/{seg_id}.json (from which trim would just recompute
the same skip/take). Used by the 'all' pipeline to avoid re-running the
expensive transcription stage when nothing upstream changed.
Returns False (i.e. "run trim") if narration can't be read or any segment
is still unresolved.
"""
from .parser import parse_project_config, parse_narration
try:
config = parse_project_config(project_path)
narration, narration_dir = parse_narration(project_path, config)
except GnommoError:
return False
if not narration:
return False
transcripts_dir = narration_dir / "transcripts"
try:
raw_data = _read_json(narration_dir / "narration.json")
except (OSError, json.JSONDecodeError):
raw_data = {}
for seg_id in narration:
entry = raw_data.get(seg_id, {})
if "skip" in entry or "take" in entry:
continue # already trimmed
if (transcripts_dir / f"{seg_id}.json").exists():
continue # transcript cached — trim would just reuse it
return False # unresolved segment: trim still has work to do
return True
def cmd_all(
project_path: Path,
verbose: bool,
@@ -4355,7 +4745,7 @@ def cmd_all(
res: str = "full",
force: bool = False,
) -> int:
"""Run full pipeline: import → preprocess → trim → stitch → render → push → handoff → up.
"""Run full pipeline: import → prune → preprocess → trim → stitch → render → push → handoff → up.
Cascade rule: if any stage produces output, all subsequent stages are forced
to re-run (cascade_force=True), regardless of whether --force was passed.
@@ -4370,7 +4760,7 @@ def cmd_all(
# True so all downstream stages re-run unconditionally.
cascade_force = force
print(">>> Step 1/8: Import\n")
print(">>> Step 1/9: Import\n")
t0 = time.time()
result = cmd_import(project_path, cascade_force, verbose)
if result != 0:
@@ -4380,7 +4770,14 @@ def cmd_all(
):
cascade_force = True
print("\n>>> Step 2/8: Preprocess\n")
print("\n>>> Step 2/9: Prune\n")
# Drop manifest entries left over from edits (e.g. a video split into two
# projects). Only removes unused entries, so it never forces downstream re-runs.
result = cmd_prune(project_path, verbose, dry_run)
if result != 0:
return result
print("\n>>> Step 3/9: Preprocess\n")
t0 = time.time()
result = cmd_preprocess(
project_path, verbose, dry_run, cascade_force, workers=1, res=res
@@ -4392,7 +4789,14 @@ def cmd_all(
) or _files_modified_since(project_path, t0, "*_processed.webm"):
cascade_force = True
print("\n>>> Step 3/8: Trim\n")
print("\n>>> Step 4/9: Trim\n")
# Skip the (Whisper-heavy) trim stage when nothing upstream changed
# (cache intact) and every segment is already resolved — i.e. it has a
# cached transcript or explicit skip/take. A cascade_force from preprocess
# (a raw video was reprocessed) always forces trim to re-run.
if not cascade_force and _trim_outputs_current(project_path):
print(" Cache intact and transcripts present for all segments — skipping trim.")
else:
t0 = time.time()
result = cmd_trim(project_path, verbose, force=cascade_force, threshold_db=-40.0)
if result != 0:
@@ -4401,7 +4805,7 @@ def cmd_all(
if _files_modified_since(project_path, t0, "narration.json"):
cascade_force = True
print("\n>>> Step 4/8: Stitch\n")
print("\n>>> Step 5/9: Stitch\n")
t0 = time.time()
result = cmd_stitch(project_path, verbose, cascade_force, res=res)
if result != 0:
@@ -4409,22 +4813,22 @@ def cmd_all(
if _files_modified_since(project_path, t0, "narration_combined.mov"):
cascade_force = True
print("\n>>> Step 5/8: Render\n")
print("\n>>> Step 6/9: Render\n")
result = cmd_render(project_path, verbose, dry_run, res=res, force=cascade_force)
if result != 0:
return result
print("\n>>> Step 6/8: Push\n")
print("\n>>> Step 7/9: Push\n")
result = cmd_push(project_path, verbose, force=False, prod=True)
if result != 0:
return result
print("\n>>> Step 7/8: Handoff\n")
print("\n>>> Step 8/9: Handoff\n")
result = cmd_handoff(project_path, verbose, file_override=None, prod=True, res=res)
if result != 0:
return result
print("\n>>> Step 8/8: Upload\n")
print("\n>>> Step 9/9: Upload\n")
from .transfer import cmd_up
return cmd_up(project_path, verbose, dry_run)
gnommo/state.py
+149
View File
@@ -0,0 +1,149 @@
"""Persistent per-stage completion tracking.
Each pipeline stage (preprocess, trim, stitch, render) records a fingerprint of
its inputs in ``.gnommo_state.json`` when it completes successfully. On the next
run a stage can ask whether its inputs are unchanged (and its output still
present) and skip the work — the same staleness intelligence that ``all``'s
in-memory cascade provides, but persisted so it also applies to stages run on
their own.
Fingerprinting is hybrid:
- small text manifests (narration.json, videos.json, manuscript.txt,
project.json, slides.json, audio.json, transcripts) are hashed (sha256) so a
``touch`` or a git checkout that only rewrites mtimes doesn't force a
needless rerun;
- large media (processed segments, narration_combined.mov, source videos and
images) use mtime+size, which is cheap and good enough to detect real edits.
The state file is purely an optimization: any read/parse/write failure degrades
to "not current" (rerun) and never raises, so a corrupt or missing state file
can't break a build.
"""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Optional, Union
STATE_FILENAME = ".gnommo_state.json"
STATE_VERSION = 1
# Fingerprint modes
HASH = "hash" # sha256 of file contents — for small text manifests
META = "meta" # mtime_ns + size — for large media
# An input descriptor is a (label, path, mode) triple.
InputSpec = tuple[str, Path, str]
def _state_path(project_path: Path) -> Path:
return project_path / STATE_FILENAME
def _empty_state() -> dict:
return {"version": STATE_VERSION, "stages": {}}
def load_state(project_path: Path) -> dict:
"""Load the state file, returning an empty skeleton on any problem."""
path = _state_path(project_path)
if not path.exists():
return _empty_state()
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return _empty_state()
if not isinstance(data, dict):
return _empty_state()
data.setdefault("version", STATE_VERSION)
if not isinstance(data.get("stages"), dict):
data["stages"] = {}
return data
def save_state(project_path: Path, state: dict) -> None:
"""Write the state file. Never raises — state is best-effort."""
try:
_state_path(project_path).write_text(
json.dumps(state, indent=2) + "\n", encoding="utf-8"
)
except OSError:
pass
def fingerprint_path(path: Union[str, Path], mode: str) -> Optional[str]:
"""Return a fingerprint for a single file, or None if it can't be read."""
p = Path(path)
try:
if mode == HASH:
h = hashlib.sha256()
with open(p, "rb") as fh:
for chunk in iter(lambda: fh.read(1 << 20), b""):
h.update(chunk)
return f"sha256:{h.hexdigest()}"
st = p.stat()
return f"meta:{st.st_mtime_ns}:{st.st_size}"
except OSError:
return None
def compute(inputs: Iterable[InputSpec]) -> dict:
"""Build a {label: fingerprint} map from (label, path, mode) triples.
A missing file yields a null fingerprint, so a file appearing or disappearing
counts as a change.
"""
return {label: fingerprint_path(path, mode) for label, path, mode in inputs}
def get_stage(project_path: Path, stage_key: str) -> dict:
"""Return the recorded record for a stage (``{}`` if none)."""
return load_state(project_path).get("stages", {}).get(stage_key, {})
def get_items(project_path: Path, stage_key: str) -> dict:
"""Return the per-item fingerprint map recorded for a stage (``{}`` if none)."""
items = get_stage(project_path, stage_key).get("items")
return items if isinstance(items, dict) else {}
def is_current(
project_path: Path,
stage_key: str,
inputs: dict,
outputs: Iterable[Union[str, Path]] = (),
) -> bool:
"""True iff the recorded input fingerprint matches ``inputs`` exactly and
every output in ``outputs`` exists on disk."""
for out in outputs:
if not Path(out).exists():
return False
recorded = get_stage(project_path, stage_key).get("inputs")
return recorded == inputs
def record(project_path: Path, stage_key: str, inputs: dict) -> None:
"""Persist a stage-level input fingerprint, marking the stage complete."""
state = load_state(project_path)
state.setdefault("stages", {})[stage_key] = {
"completed_at": datetime.now(timezone.utc).isoformat(),
"inputs": inputs,
}
save_state(project_path, state)
def record_items(project_path: Path, stage_key: str, items: dict) -> None:
"""Merge per-item fingerprints into a stage's record (for multi-segment
stages like preprocess/trim). Existing items for other keys are preserved."""
state = load_state(project_path)
stage = state.setdefault("stages", {}).setdefault(stage_key, {})
stage["completed_at"] = datetime.now(timezone.utc).isoformat()
merged = stage.get("items")
if not isinstance(merged, dict):
merged = {}
merged.update(items)
stage["items"] = merged
save_state(project_path, state)