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

Adding fixes to the stitcher

Browse Source
This commit is contained in:
gitprov
2026-07-05 12:16:35 +02:00
parent b9b5a8e77d
commit f9ff847f6b
11 changed files with 1019 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gnommo/cli.py
+640 -73
View File
@@ -76,8 +76,8 @@ Examples:
"-p",
"--project",
type=str,
required=True,
help="Project directory",
default=None,
help="Project directory (required for all actions except 'pexels --search')",
)
parser.add_argument(
"action",
@@ -193,6 +193,12 @@ Examples:
default=-40.0,
help="For trim: silence threshold in dB (default: -40). Raise (e.g. -25) to ignore clothing/room noise.",
)
parser.add_argument(
"--model",
type=str,
default="base",
help="For trim: Whisper model for transcription (tiny/base/small/medium/large, default: base)",
)
parser.add_argument(
"--crf",
type=int,
@@ -216,17 +222,38 @@ Examples:
dest="alpha_quality",
help="For transcode --processed: HEVC alpha quality 0.0-1.0 (default: 0.75; lower=smaller file)",
)
parser.add_argument(
"--search",
type=str,
default=None,
metavar="QUERY",
help="For pexels: search Pexels for QUERY and download all results to the assets disk",
)
parser.add_argument(
"--max",
type=int,
default=200,
dest="search_max",
help="For pexels --search: maximum number of videos to download (default: 200)",
)
args = parser.parse_args()
# Resolve project path
project_path = Path(args.project)
if not project_path.is_absolute():
project_path = Path.cwd() / project_path
# Resolve project path (optional for 'pexels --search')
action = args.action
if args.project is None:
if action == "pexels" and args.search:
project_path = Path.cwd() # placeholder; cmd_pexels won't use it in search mode
else:
parser.error("argument -p/--project is required")
return 1
else:
project_path = Path(args.project)
if not project_path.is_absolute():
project_path = Path.cwd() / project_path
try:
# Handle actions
action = args.action
if action == "import":
return cmd_import(project_path, args.force, args.verbose)
@@ -243,7 +270,7 @@ Examples:
)
elif action == "trim":
return cmd_trim(
project_path, args.verbose, args.force, args.threshold, args.res
project_path, args.verbose, args.force, args.threshold, args.res, args.model
)
elif action == "transcode":
return cmd_transcode(
@@ -312,7 +339,7 @@ Examples:
project_path, args.verbose, args.file, args.prod, args.res
)
elif action == "pexels":
return cmd_pexels(project_path, args.verbose)
return cmd_pexels(project_path, args.verbose, args.search, args.search_max)
except GnommoError as e:
print(f"Error: {e}", file=sys.stderr)
@@ -329,6 +356,213 @@ Examples:
# =============================================================================
def _prompter_break_line(text: str, max_words: int = 10) -> list[str]:
"""Break a long line into short breath-sized chunks for teleprompter display.
Priority order for break points:
1. Sentence end (. ? !) — always break here
2. Em/en-dash ( — / - ) — break if ≥ 4 words accumulated
3. Semicolon (;) — break if ≥ 4 words accumulated
4. Comma (,) — break if ≥ 5 words accumulated
5. Word ceiling (max_words) — force break at word boundary
Each output line is stripped and non-empty.
"""
import re
words = text.split()
if len(words) <= max_words:
return [text.strip()] if text.strip() else []
lines: list[str] = []
current: list[str] = []
# Dash detection: standalone dash token, or word that is entirely dashes/hyphens
_DASH_RE = re.compile(r"^[-–—]+$")
_SENTENCE_END_RE = re.compile(r'[.?!]["\']?$')
_SOFT_END_COMMA = re.compile(r",$")
_SOFT_END_SEMI = re.compile(r";$")
def flush(force: bool = False) -> None:
"""Flush current words to a line.
If the accumulated chunk is very short (≤ 2 words) and this isn't a
sentence-end forced flush, hold it — it will merge into the next line.
This prevents orphaned 1-2-word lines from soft breaks.
"""
if not force and len(current) <= 2:
return # too short to stand alone — let more words join
chunk = " ".join(current).strip()
if chunk:
lines.append(chunk)
current.clear()
i = 0
while i < len(words):
word = words[i]
n = len(current)
# Peek ahead: is the next token a standalone dash?
next_is_dash = (i + 1 < len(words)) and _DASH_RE.match(words[i + 1])
if _DASH_RE.match(word):
# Standalone dash: attach to current line then break.
current.append(word)
if n > 0:
flush()
elif _SENTENCE_END_RE.search(word):
current.append(word)
flush(force=True) # sentence end always flushes regardless of length
elif _SOFT_END_SEMI.search(word) and n >= 4:
current.append(word)
flush()
elif _SOFT_END_COMMA.search(word) and n >= 5:
current.append(word)
flush()
elif next_is_dash and n >= 5:
flush()
current.append(word)
elif n >= max_words:
flush(force=True) # word ceiling: must break even if line is short
current.append(word)
else:
current.append(word)
i += 1
flush()
return lines
def _export_prompter_manuscript(
manuscript_path: Path,
verbose: bool,
prompter_wpm: int = 130,
max_words: int = 10,
) -> None:
"""Generate manuscript_prompter.txt from manuscript.txt.
Strips all bracketed markers except [S*] and [cue:*], auto-breaks long
lines into breath-sized chunks, and converts [pause:Xs] markers into blank
lines calibrated to the teleprompter scroll speed.
Calibration: the scroll speed is assumed to match reading speed (prompter_wpm).
At that speed each line takes (60 * max_words / prompter_wpm) seconds, so a
[pause:Xs] marker becomes round(X * prompter_wpm / (60 * max_words)) blank lines,
preceded by a visible label line ("· · · 8s · · ·").
"""
import re
output_path = manuscript_path.parent / "manuscript_prompter.txt"
text = manuscript_path.read_text(encoding="utf-8")
_MARKER_RE = re.compile(r"\[[^\]]+\]")
_KEEP_RE = re.compile(r"^\[S[^\]]*\]$|^\[cue:[^\]]*\]$")
_SEGMENT_RE = re.compile(r"^\[segment:[^\]]*\]$")
_PAUSE_RE = re.compile(r"^\[pause:(\d+(?:\.\d+)?)s?\]$")
_BLOCK_BREAK = "\x00BLOCK\x00"
def _pause_lines(seconds: float) -> list[str]:
"""Convert a pause duration into a label + blank lines for the teleprompter."""
n_blanks = max(1, round(seconds * prompter_wpm / (60 * max_words)))
label = f"· · · {seconds:g}s · · ·"
return [label] + [""] * n_blanks
# Step 1: strip non-display markers, collect cleaned lines
cleaned_lines: list[str] = []
for line in text.splitlines():
stripped = line.strip()
if _MARKER_RE.fullmatch(stripped):
if _SEGMENT_RE.match(stripped):
cleaned_lines.append(_BLOCK_BREAK)
elif _KEEP_RE.match(stripped):
cleaned_lines.append(stripped)
else:
m = _PAUSE_RE.match(stripped)
if m:
cleaned_lines.extend(_pause_lines(float(m.group(1))))
# else: drop marker-only lines entirely
else:
def _replace(m: re.Match) -> str:
# Handle inline [pause:Xs] — convert to label inline
if _PAUSE_RE.match(m.group(0)):
sec = float(_PAUSE_RE.match(m.group(0)).group(1))
return f"· · · {sec:g}s · · ·"
return m.group(0) if _KEEP_RE.match(m.group(0)) else ""
cleaned = _MARKER_RE.sub(_replace, line).strip()
if cleaned:
cleaned_lines.append(cleaned)
else:
cleaned_lines.append("") # preserve intentional blank lines
# Step 2: auto-break long text lines; pass markers, blanks, and pause labels through
_PAUSE_LABEL_RE = re.compile(r"^· · · .+ · · ·$")
broken: list[str] = []
for line in cleaned_lines:
stripped = line.strip()
if not stripped or stripped == _BLOCK_BREAK or _KEEP_RE.match(stripped) or _PAUSE_LABEL_RE.match(stripped):
broken.append(line)
else:
broken.extend(_prompter_break_line(stripped))
# Step 3: merge each slide marker onto the first word of its slide,
# then lay out the final lines.
# - [S*] markers are prepended inline to the next text line
# - Block-break sentinels expand to a double blank (Elgato block boundary)
# - Ordinary consecutive blank lines are collapsed to one
merged: list[str] = []
pending_marker: str = ""
pending_breaks: int = 0 # block-breaks that arrived while a marker was pending
for line in broken:
stripped = line.strip()
if stripped and _KEEP_RE.match(stripped):
pending_marker = stripped # hold until the next text line
elif pending_marker:
if stripped and stripped != _BLOCK_BREAK:
# Flush any buffered block-breaks first, then attach marker to text
for _ in range(pending_breaks):
merged.append(_BLOCK_BREAK)
pending_breaks = 0
merged.append(f"{pending_marker} {stripped}")
pending_marker = ""
elif stripped == _BLOCK_BREAK:
# Buffer the block-break — emit it after the marker attaches
pending_breaks += 1
else:
# Ordinary blank: carry forward (swallow into the marker's gap)
pass
else:
merged.append(line)
if pending_marker:
for _ in range(pending_breaks):
merged.append(_BLOCK_BREAK)
merged.append(pending_marker)
result: list[str] = []
prev_blank_count = 0
for line in merged:
if line == _BLOCK_BREAK:
needed = 2 - prev_blank_count
result.extend([""] * max(needed, 0))
prev_blank_count = 2
continue
is_blank = not line.strip()
if is_blank:
if prev_blank_count >= 1:
continue # collapse ordinary consecutive blanks
result.append("")
prev_blank_count += 1
else:
result.append(line)
prev_blank_count = 0
output_path.write_text("\n".join(result).strip() + "\n", encoding="utf-8")
print(f" Prompter manuscript → {output_path.name}")
def cmd_import(project_path: Path, force: bool, verbose: bool) -> int:
"""Import assets and generate metadata JSON files."""
from .parser import parse_project_config, _read_json
@@ -421,15 +655,29 @@ def cmd_import(project_path: Path, force: bool, verbose: bool) -> int:
for marker in markers:
for prefix in _SHORTHAND_PREFIXES:
if marker.startswith(prefix):
vid_id = marker[len(prefix):]
vid_id = marker[len(prefix):].lower()
if vid_id not in local_vids and vid_id not in seen_missing:
hint = (
"run 'gnommo pexels' to download"
if vid_id.startswith("pexels/")
else "add it manually"
)
print(
f" ⚠ [{marker}] video '{vid_id}' not found in "
f"videos.json or shared_assets — add it manually"
f"videos.json or shared_assets — {hint}"
)
seen_missing.add(vid_id)
break
# Generate teleprompter version of manuscript
if manuscript_path.exists():
import configparser as _cp
_pcfg = _cp.ConfigParser()
_pcfg.read(Path.home() / ".gnommo.conf")
_wpm = int(_pcfg.get("prompter", "wpm", fallback="130"))
_mpw = int(_pcfg.get("prompter", "max_words_per_line", fallback="10"))
_export_prompter_manuscript(manuscript_path, verbose, prompter_wpm=_wpm, max_words=_mpw)
print("Import complete.")
return 0
@@ -473,7 +721,7 @@ def _import_shared_audio(
added = 0
for f in audio_files:
audio_id = f.stem
audio_id = f.stem.lower()
if audio_id in existing:
if verbose:
print(f" Skipping {audio_id} (already in audio.json)")
@@ -619,21 +867,27 @@ def _probe_video_metadata(
)
# Mirror renderer._resolve_video_path: try output_file first, then source_file
# Use resolve_with_cache so files on external disks (LaCie, GnommoDisk) are found.
from .cache import resolve_with_cache
video_path = None
output_file = video_data.get("output_file")
if output_file:
for candidate_dir in [base_dir, base_dir.parent]:
candidate = candidate_dir / output_file
candidate, _ = resolve_with_cache(candidate, project_path)
if candidate.exists():
video_path = candidate
break
mov_candidate = candidate.with_suffix(".mov")
mov_candidate, _ = resolve_with_cache(mov_candidate, project_path)
if mov_candidate.exists():
video_path = mov_candidate
break
if video_path is None:
source_candidate = base_dir / video_data["source_file"]
source_candidate, _ = resolve_with_cache(source_candidate, project_path)
if source_candidate.exists():
video_path = source_candidate
@@ -717,8 +971,13 @@ def _sync_shared_videos_to_local(
elif verbose:
print(f" Shared '{video_id}': already in local videos.json, skipping")
continue
# New entry — copy from shared and mark it as shared
local_entry = dict(shared_entry)
# New entry — copy base metadata from shared and mark it as shared.
# Exclude presentation fields (cutout, layer) so _project_markers_to_videos
# can set them from the manuscript without stale shared values interfering.
_PRESENTATION_FIELDS = {"cutout", "layer", "pause_narration"}
local_entry = {
k: v for k, v in shared_entry.items() if k not in _PRESENTATION_FIELDS
}
local_entry["is_shared"] = True
local_videos[video_id] = local_entry
added.append(video_id)
@@ -769,15 +1028,15 @@ def _import_shared_assets(shared_assets_dir: Path, verbose: bool) -> None:
video_extensions = {".mov", ".mp4", ".webm", ".avi", ".mkv", ".m4v"}
# Find all video files in shared_assets (root level and subdirectories).
# Also scan the GnommoDisk cache mirror so files placed there are registered.
from .cache import load_cache_config
# Also scan external disk mirrors so files placed there are registered.
from .cache import load_assets_config, load_cache_config
scan_roots: list[Path] = [shared_assets_dir]
cache_base = load_cache_config()
if cache_base:
cache_shared = cache_base / "shared_assets"
if cache_shared.exists() and cache_shared != shared_assets_dir:
scan_roots.append(cache_shared)
for external_base in (load_cache_config(), load_assets_config()):
if external_base:
ext_shared = external_base / "shared_assets"
if ext_shared.exists() and ext_shared != shared_assets_dir:
scan_roots.append(ext_shared)
video_files: list[tuple[Path, Path]] = [] # (relative_path, absolute_path)
seen_rel: set[str] = set() # deduplicate by relative path
@@ -821,12 +1080,28 @@ def _import_shared_assets(shared_assets_dir: Path, verbose: bool) -> None:
if videos_json_path.exists():
existing_videos = _read_json(videos_json_path)
# Remove entries whose source file no longer exists on any scan root
removed_count = 0
stale_keys = [
vid_id
for vid_id, vid_data in existing_videos.items()
if vid_data.get("source_file")
and not any((root / vid_data["source_file"]).exists() for root in scan_roots)
]
for vid_id in stale_keys:
del existing_videos[vid_id]
removed_count += 1
if verbose:
print(f" Removed stale shared entry: {vid_id}")
if removed_count > 0:
print(f" Removed {removed_count} stale shared asset(s) from videos.json")
# Add new videos (don't overwrite existing)
added_count = 0
for rel_path, abs_path in sorted(video_files):
# Use path relative to shared_assets without extension as video_id
# e.g., "Logo" for root files, "pexels/6759604-hd" for subdirectory files
video_id = str(rel_path.with_suffix(""))
# Use path relative to shared_assets without extension as video_id (lowercase)
# e.g., "logo" for root files, "pexels/6759604-hd" for subdirectory files
video_id = str(rel_path.with_suffix("")).lower()
if video_id in existing_videos:
if verbose:
@@ -840,11 +1115,12 @@ def _import_shared_assets(shared_assets_dir: Path, verbose: bool) -> None:
if verbose:
print(f" Added: {video_id}")
if added_count > 0:
if added_count > 0 or removed_count > 0:
# Write updated videos.json
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump(existing_videos, f, indent=2)
print(f" Updated {videos_json_path} (+{added_count} shared assets)")
if added_count > 0:
print(f" Updated {videos_json_path} (+{added_count} shared assets)")
else:
print(f" No new shared assets to add")
@@ -940,14 +1216,32 @@ def _import_videos(videos_dir: Path, config, verbose: bool) -> None:
if videos_json_path.exists():
existing_videos = _read_json(videos_json_path)
# Remove entries whose source file no longer exists on disk.
# Skip is_shared entries — their source files live in shared_assets, not videos_dir.
removed_count = 0
stale_keys = [
vid_id
for vid_id, vid_data in existing_videos.items()
if not vid_data.get("is_shared")
and vid_data.get("source_file")
and not (videos_dir / vid_data["source_file"]).exists()
]
for vid_id in stale_keys:
del existing_videos[vid_id]
removed_count += 1
if verbose:
print(f" Removed stale entry: {vid_id}")
if removed_count > 0:
print(f" Removed {removed_count} stale video(s) from {videos_json_path.name}")
# Get available filter presets from config
default_filters = config.default_filters if config else {}
# Add new videos (don't overwrite existing)
added_count = 0
for video_file in sorted(video_files):
# Use filename without extension as video_id
video_id = video_file.stem
# Use filename without extension as video_id (lowercase for new entries)
video_id = video_file.stem.lower()
if video_id in existing_videos:
if verbose:
@@ -981,11 +1275,12 @@ def _import_videos(videos_dir: Path, config, verbose: bool) -> None:
existing_videos[video_id] = video_entry
added_count += 1
if added_count > 0:
if added_count > 0 or removed_count > 0:
# Write updated videos.json
with open(videos_json_path, "w", encoding="utf-8") as f:
json.dump(existing_videos, f, indent=2)
print(f" Updated {videos_json_path.name} (+{added_count} videos)")
if added_count > 0:
print(f" Updated {videos_json_path.name} (+{added_count} videos)")
else:
print(f" No new videos to add")
@@ -1018,6 +1313,41 @@ def _import_narration_segments(narration_dir: Path, config, verbose: bool) -> No
if narration_json_path.exists():
existing_narration = _read_json(narration_json_path)
# Remove stale entries: no raw_mov file and source_file also gone
_raw_video_exts_set = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
raw_stems: set[str] = set()
if raw_dir.exists():
for _f in raw_dir.iterdir():
if _f.is_file() and _f.suffix.lower() in _raw_video_exts_set:
raw_stems.add(_f.stem.lower())
stale_keys = [
seg_id
for seg_id, seg_data in existing_narration.items()
if seg_id.lower() not in raw_stems
and not (narration_dir / seg_data.get("source_file", "")).exists()
]
for seg_id in stale_keys:
del existing_narration[seg_id]
print(f" Removed stale narration segment: {seg_id}")
# Normalise keys to lowercase, merging case-duplicates by keeping the entry
# with more data (skip/take values are the most important thing to preserve).
normalised: dict = {}
merged_count = 0
for seg_id, seg_data in existing_narration.items():
lower_id = seg_id.lower()
if lower_id in normalised:
existing_has_trim = "skip" in normalised[lower_id] or "take" in normalised[lower_id]
incoming_has_trim = "skip" in seg_data or "take" in seg_data
if incoming_has_trim and not existing_has_trim:
normalised[lower_id] = seg_data
merged_count += 1
print(f" Merged duplicate narration segment: '{seg_id}' → '{lower_id}'")
else:
normalised[lower_id] = seg_data
existing_narration = normalised
default_filters = config.default_filters if config else {}
added_count = 0
@@ -1041,6 +1371,7 @@ def _import_narration_segments(narration_dir: Path, config, verbose: bool) -> No
# Strip _processed suffix for cleaner segment IDs if present
if segment_id.endswith("_processed"):
segment_id = segment_id[:-10]
segment_id = segment_id.lower()
if segment_id in existing_narration:
if verbose:
@@ -1066,7 +1397,7 @@ def _import_narration_segments(narration_dir: Path, config, verbose: bool) -> No
# 2. Scan raw/ — add entries for raw files not yet in narration.json
for video_file in _scan(raw_dir):
segment_id = video_file.stem
segment_id = video_file.stem.lower()
if segment_id in existing_narration:
if verbose:
@@ -1089,12 +1420,20 @@ def _import_narration_segments(narration_dir: Path, config, verbose: bool) -> No
added_count += 1
print(f" Added narration segment: {segment_id} (from raw_mov)")
if added_count > 0 or not narration_json_path.exists():
removed_count = len(stale_keys)
if added_count > 0 or removed_count > 0 or merged_count > 0 or not narration_json_path.exists():
with open(narration_json_path, "w", encoding="utf-8") as f:
json.dump(existing_narration, f, indent=2)
if added_count > 0:
print(f" Updated narration.json (+{added_count} segments)")
if added_count > 0 or removed_count > 0 or merged_count > 0:
parts = []
if added_count:
parts.append(f"+{added_count}")
if removed_count:
parts.append(f"-{removed_count}")
if merged_count:
parts.append(f"merged {merged_count} duplicate(s)")
print(f" Updated narration.json ({', '.join(parts)} segments)")
else:
if not existing_narration:
print(f" narration.json created (empty — add files to processed/ or raw/)")
@@ -1327,7 +1666,12 @@ def _write_tasks_file(
# =============================================================================
def cmd_pexels(project_path: Path, verbose: bool) -> int:
def cmd_pexels(
project_path: Path,
verbose: bool,
search_query: Optional[str] = None,
search_max: int = 200,
) -> int:
"""Download missing Pexels videos and enrich metadata for existing ones."""
from .parser import parse_manuscript, parse_project_config, parse_videos
from .pexels import (
@@ -1336,6 +1680,7 @@ def cmd_pexels(project_path: Path, verbose: bool) -> int:
download_video,
update_videos_json,
enrich_missing_descriptions,
search_and_download,
)
api_key = get_pexels_api_key()
@@ -1350,20 +1695,66 @@ def cmd_pexels(project_path: Path, verbose: bool) -> int:
)
return 1
config = parse_project_config(project_path)
_, markers, _, _ = parse_manuscript(project_path)
videos, _ = parse_videos(project_path, config)
# --- Search mode: download all results for a query to the assets disk ---
if search_query:
from .cache import load_assets_config
# Find shared_assets: prefer assets disk (LaCie), fall back to local
assets_root = load_assets_config()
if assets_root and assets_root.exists():
pexels_dir = assets_root / "shared_assets" / "pexels"
# Use the assets disk's shared_assets/videos.json as the registry
shared_videos_json = assets_root / "shared_assets" / "videos.json"
print(f"Downloading to: {pexels_dir}")
else:
# No assets disk — find local shared_assets from project_path or cwd
local_shared = _find_shared_assets(project_path)
if not local_shared:
print("Error: shared_assets directory not found.", file=sys.stderr)
return 1
pexels_dir = local_shared / "pexels"
shared_videos_json = local_shared / "videos.json"
print(f"Assets disk not found — downloading to local: {pexels_dir}")
dl, sk = search_and_download(
search_query, pexels_dir, shared_videos_json, api_key, max_results=search_max
)
print(f"\nDone — {dl} downloaded, {sk} already present.")
return 0
shared_assets_dir = _find_shared_assets(project_path)
if not shared_assets_dir:
print("Error: shared_assets directory not found.", file=sys.stderr)
return 1
local_videos_json = project_path / config.videos_path
shared_videos_json = shared_assets_dir / "videos.json"
# 1. Download missing files
# --- Normal mode: download missing files referenced in the manuscript ---
config = parse_project_config(project_path)
_, markers, _, _ = parse_manuscript(project_path)
videos, _ = parse_videos(project_path, config)
local_videos_json = project_path / config.videos_path
# 1. Collect missing files: manuscript markers + background
missing = find_missing_pexels_videos(markers, videos, shared_assets_dir)
# Also check background video from project.json
if config and config.background and config.background.startswith("pexels/"):
from .cache import resolve_with_cache
bg_id = config.background
shared_vids_raw = _read_json(shared_videos_json) if shared_videos_json.exists() else {}
bg_entry = shared_vids_raw.get(bg_id)
if bg_entry:
bg_sf = bg_entry.get("source_file", "")
bg_path = shared_assets_dir / bg_sf
resolved, _ = resolve_with_cache(bg_path, shared_assets_dir)
if not resolved.exists():
# Only add if not already in the manuscript-marker list
if not any(vid_id == bg_id for vid_id, _ in missing):
missing.append((bg_id, bg_sf))
failed = 0
if missing:
print(f"Downloading {len(missing)} missing Pexels video(s)...")
@@ -1395,6 +1786,7 @@ def cmd_pexels(project_path: Path, verbose: bool) -> int:
def cmd_validate(project_path: Path, verbose: bool) -> int:
"""Validate project configuration."""
from .parser import (
parse_audio,
parse_manuscript,
parse_project_config,
parse_slides,
@@ -1413,15 +1805,17 @@ def cmd_validate(project_path: Path, verbose: bool) -> int:
config = parse_project_config(project_path)
slides = parse_slides(project_path, config)
videos, videos_dir = parse_videos(project_path, config)
audio, _ = parse_audio(project_path, config)
if verbose:
print(f" - Markers in manuscript: {len(markers)}")
print(f" - Slides defined: {len(slides)}")
print(f" - Videos defined: {len(videos)}")
print(f" - Audio cues defined: {len(audio)}")
# Validate
warnings = validate_project(
project_path, markers, config, slides, videos, videos_dir, malformed
project_path, markers, config, slides, videos, videos_dir, malformed, audio
)
for w in warnings:
print(f" Warning: {w}")
@@ -1440,13 +1834,32 @@ def cmd_validate(project_path: Path, verbose: bool) -> int:
def _resolve_process_cache(project_path: Path, config) -> Optional[Path]:
"""Return per-project cache dir on external disk, or None if not configured."""
if not (config and config.process_cache):
return None
p = Path(config.process_cache)
if not p.is_absolute():
p = (project_path / p).resolve()
return p / project_path.name
"""Return per-project cache dir on an external disk, or None if none is accessible.
Resolution order:
1. config.process_cache (project.json) — e.g. /Volumes/GnommoDisk/gnommocache
Must already exist (disk must be mounted and dir pre-created).
2. [assets] process cache (~/.gnommo.conf) — derived as <assets_path>cache
e.g. /Volumes/LaCie Jens/Projects/gnommocache.
Only requires the parent to exist (volume mounted); the cache dir
itself is created on first use by the caller.
"""
from .cache import load_assets_process_cache
# Primary: explicitly configured in project.json
if config and config.process_cache:
p = Path(config.process_cache)
if not p.is_absolute():
p = (project_path / p).resolve()
if p.exists():
return p / project_path.name
# Fallback: [assets] disk (need only the parent volume/dir to be present)
assets_cache = load_assets_process_cache()
if assets_cache and assets_cache.parent.exists():
return assets_cache / project_path.name
return None
def _narration_combined_hint(project_path: Path, config) -> str:
@@ -1799,9 +2212,104 @@ def cmd_preprocess(
# =============================================================================
# Trim Command — auto-detect silence bounds for narration segments
# Trim Command — transcript-based trimming for slide-range segments
# =============================================================================
# Words so common they're useless for matching slide boundaries.
_TRIM_STOP_WORDS = frozenset({
"a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "of",
"for", "is", "it", "its", "we", "our", "this", "that", "with", "be",
"are", "was", "has", "not", "so", "as", "by", "do", "if", "up",
"you", "i", "he", "she", "they", "them", "just", "now", "can",
"than", "then", "from", "about", "into", "out", "what", "there",
"when", "how", "who", "all", "very", "also", "more", "get", "have",
})
def _parse_segment_slide_range(seg_id: str) -> "tuple[int, int | None] | None":
"""Parse 'S11-24' → (11, 24), 'S1-end' → (1, None), else None."""
m = re.match(r"^s(\d+)-(?:s?(\d+)|(end))$", seg_id.lower())
if not m:
return None
start = int(m.group(1))
end = int(m.group(2)) if m.group(2) else None
return (start, end)
def _extract_slide_texts(manuscript_path: Path) -> "dict[int, str]":
"""Return {slide_num: text} for each [SN] marker in the manuscript."""
text = manuscript_path.read_text(encoding="utf-8")
parts = re.split(r"\[S(\d+)\]", text)
result: dict[int, str] = {}
i = 1
while i + 1 < len(parts):
slide_num = int(parts[i])
content = re.sub(r"\[[^\]]+\]", " ", parts[i + 1])
result[slide_num] = content.strip()
i += 2
return result
def _trim_content_words(text: str) -> "list[str]":
"""Lowercase words stripped of punctuation, excluding stop words."""
words = re.findall(r"[a-zA-Z0-9']+", text.lower())
return [w for w in words if w not in _TRIM_STOP_WORDS and len(w) > 2]
def _find_slide_end_in_transcript(
transcript_words: list,
slide_text: str,
verbose: bool = False,
) -> "float | None":
"""
Locate where slide_text ends in the transcript and return that word's
end timestamp. Searches the tail of the transcript for the last few
content words from slide_text using a sequential fuzzy match.
Returns None if no confident match is found.
"""
target = _trim_content_words(slide_text)[-12:]
if not target:
return None
# Normalise transcript words (strip punctuation, lowercase)
norm = [re.sub(r"[^a-z0-9']", "", w.word.lower()) for w in transcript_words]
# Index into transcript of content words only
content_idxs = [
i for i, w in enumerate(norm)
if w and w not in _TRIM_STOP_WORDS and len(w) > 2
]
if not content_idxs:
return None
n = len(target)
threshold = max(1, int(n * 0.55))
# Scan backwards: find the latest window of n content words that matches
for end_ci in range(len(content_idxs) - 1, n - 2, -1):
window_idxs = content_idxs[max(0, end_ci - n + 1): end_ci + 1]
window = [norm[i] for i in window_idxs]
# Sequential match: iterate target left-to-right, consume window matches
score = 0
wi = 0
for t_word in target:
while wi < len(window):
w = window[wi]
wi += 1
if w == t_word or (len(w) >= 4 and len(t_word) >= 4 and w[:4] == t_word[:4]):
score += 1
break
if score >= threshold:
last_tw = transcript_words[content_idxs[end_ci]]
if verbose:
print(f"\n → matched slide end: '{last_tw.word}' at {last_tw.end:.2f}s")
return last_tw.end
if verbose:
print(f"\n → could not match slide end (target tail: {target[-5:]})")
return None
def cmd_trim(
project_path: Path,
@@ -1809,25 +2317,24 @@ def cmd_trim(
force: bool = False,
threshold_db: float = -40.0,
res: str = "full",
whisper_model: str = "base",
) -> int:
"""
Auto-detect silence bounds for all narration segments and write skip/take
values into narration.json.
Trim narration segments and write skip/take values into narration.json.
For each segment:
skip = max(0, first_sound_time - 0.5)
take = last_sound_time + 3.0 - skip (capped at file duration)
For segments named S{N}-{M}.mov or S{N}-end.mov:
- Transcribes audio with Whisper to get word-level timestamps
- skip = max(0, first_word.start - 0.5)
- take = (end of last word on slide M) + 0.15 - skip
- S{N}-end.mov: only trims the start, no end trim
- Transcripts are cached in narration/transcripts/{seg_id}.json
Segments that already have explicit skip or take values are left unchanged
unless --force is passed.
Use --threshold to adjust sensitivity, e.g. -25 to ignore clothing/room
noise that sits above -40 dB.
For other segments: falls back to silence detection.
"""
from .parser import parse_project_config, parse_narration
from .preprocessor import detect_silence_bounds, get_video_duration
print(f"Auto-trimming narration: {project_path.name}")
print(f"Trimming narration: {project_path.name}")
config = parse_project_config(project_path)
narration, narration_dir = parse_narration(project_path, config)
@@ -1837,12 +2344,18 @@ def cmd_trim(
print(" Run 'gnommo -p <project> import' first.")
return 1
# Build a lookup of raw source files by segment ID. Raw files give cleaner
# silence detection — loudnorm can introduce early peaks in processed audio.
# Load slide texts from manuscript for transcript-based trimming
manuscript_path = project_path / "manuscript.txt"
slide_texts: dict[int, str] = {}
if manuscript_path.exists():
slide_texts = _extract_slide_texts(manuscript_path)
_video_exts = {".mov", ".mp4", ".avi", ".mkv", ".m4v"}
raw_dir = narration_dir / "raw_mov"
compressed_dir = narration_dir / "raw_mp4"
transcripts_dir = narration_dir / "transcripts"
# Build lookup of raw files keyed by lowercased stem
raw_lookup: dict[str, Path] = {}
for search_dir in (raw_dir, compressed_dir):
if search_dir.exists():
@@ -1855,7 +2368,7 @@ def cmd_trim(
stem = f.stem
if stem.endswith("_compressed"):
stem = stem[: -len("_compressed")]
raw_lookup[stem] = f
raw_lookup[stem.lower()] = f
narration_json_path = narration_dir / "narration.json"
raw_data: dict = _read_json(narration_json_path)
@@ -1870,7 +2383,7 @@ def cmd_trim(
print(f" {seg_id}: already trimmed, skipping (use --force to redo)")
continue
# Prefer raw file; fall back to processed if raw not available.
# Prefer raw file; fall back to source_file from narration.json
source_path = raw_lookup.get(seg_id)
if source_path is None:
source_path = narration_dir / seg.source_file
@@ -1878,6 +2391,68 @@ def cmd_trim(
print(f" {seg_id}: source file not found, skipping")
continue
slide_range = _parse_segment_slide_range(seg_id) if slide_texts else None
if slide_range is not None:
# --- Transcript-based trimming ---
from .transcriber import transcribe_video, save_transcript, load_transcript, TranscriptionError
start_slide, end_slide = slide_range
transcripts_dir.mkdir(parents=True, exist_ok=True)
transcript_path = transcripts_dir / f"{seg_id}.json"
try:
if transcript_path.exists() and not force:
words = load_transcript(transcript_path)
print(f" {seg_id}: loaded cached transcript ({len(words)} words)", end="", flush=True)
else:
print(f" {seg_id}: transcribing {source_path.parent.name}/{source_path.name} (model={whisper_model})...", end="", flush=True)
words = transcribe_video(source_path, model=whisper_model)
save_transcript(words, transcript_path)
print(f" {len(words)} words", end="", flush=True)
if not words:
print(f" — no words found, falling back to silence detection")
slide_range = None
else:
total_dur = get_video_duration(source_path)
new_skip = max(0.0, round(words[0].start - 0.5, 3))
if end_slide is None:
# S{N}-end: only trim start
new_take = round(total_dur - new_skip, 3)
print(
f" first={words[0].start:.2f}s"
f" → skip={new_skip:.3f}s take={new_take:.3f}s (no end trim)"
)
else:
last_slide_text = slide_texts.get(end_slide, "")
end_ts = None
if last_slide_text:
end_ts = _find_slide_end_in_transcript(words, last_slide_text, verbose)
if end_ts is None:
print(f"\n ⚠ could not locate S{end_slide} end in transcript — trimming start only")
new_take = round(total_dur - new_skip, 3)
else:
new_take = round(min(end_ts + 0.15 - new_skip, total_dur - new_skip), 3)
new_take = max(0.0, new_take)
print(
f" first={words[0].start:.2f}s S{end_slide}_end={end_ts:.2f}s"
f" → skip={new_skip:.3f}s take={new_take:.3f}s"
)
raw_data[seg_id]["skip"] = new_skip
raw_data[seg_id]["take"] = new_take
updated += 1
continue
except Exception as exc:
from .transcriber import TranscriptionError
label = "Whisper not installed" if "openai-whisper" in str(exc) else str(exc)
print(f"\n ⚠ transcription failed ({label}), falling back to silence detection")
slide_range = None # fall through
# --- Silence-based fallback ---
print(
f" {seg_id}: analysing {source_path.parent.name}/{source_path.name}...",
end="",
@@ -2670,7 +3245,7 @@ def _project_markers_to_videos(
for marker in markers:
for prefix, implied in _SHORTHAND_PREFIXES.items():
if marker.startswith(prefix):
video_id = marker[len(prefix):]
video_id = marker[len(prefix):].lower()
cutout, layer = implied[0], implied[1]
projection[video_id] = {
"cutout": cutout,
@@ -2720,14 +3295,6 @@ def _project_markers_to_videos(
if updated_local:
print(f" Projected marker semantics → videos.json: {', '.join(updated_local)}")
# Also project into shared_assets/videos.json for pexels/library videos
shared_assets_dir = _find_shared_assets(project_path) if project_path else None
if shared_assets_dir:
shared_videos_json = shared_assets_dir / "videos.json"
updated_shared = _apply_projection(shared_videos_json)
if updated_shared:
print(f" Projected marker semantics → shared_assets/videos.json: {', '.join(updated_shared)}")
def _writeback_video_metadata(plan, project_path, config) -> None:
"""Write back cutout/layer derived from shorthand markers to videos.json.