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

Fixing black formatting

Browse Source
This commit is contained in:
Glitch University
2026-05-13 21:53:22 +02:00
parent 20aba06be1
commit 980bb84dac
6 changed files with 177 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gnommo/cache.py
+3 -1
View File
@@ -38,7 +38,9 @@ def get_ffmpeg_thread_count() -> int:
cfg.read(config_path)
if cfg.has_option("performance", "cpu_limit"):
try:
_perf_config["cpu_limit"] = float(cfg.get("performance", "cpu_limit"))
_perf_config["cpu_limit"] = float(
cfg.get("performance", "cpu_limit")
)
except ValueError:
pass
gnommo/cli.py
+110 -38
View File
@@ -241,7 +241,9 @@ Examples:
args.res,
)
elif action == "trim":
return cmd_trim(project_path, args.verbose, args.force, args.threshold, args.res)
return cmd_trim(
project_path, args.verbose, args.force, args.threshold, args.res
)
elif action == "transcode":
return cmd_transcode(
project_path,
@@ -449,14 +451,19 @@ def _import_shared_audio(
if added > 0:
with open(audio_json_path, "w", encoding="utf-8") as fh:
json.dump(existing, fh, indent=2)
print(f" Updated {audio_json_path.relative_to(project_path)} (+{added} shared audio files)")
print(
f" Updated {audio_json_path.relative_to(project_path)} (+{added} shared audio files)"
)
else:
if verbose:
print(f" No new shared audio files to add")
def _probe_audio_durations(
project_path: Path, config, force: bool, verbose: bool,
project_path: Path,
config,
force: bool,
verbose: bool,
shared_assets_dir: Optional[Path] = None,
) -> None:
"""Probe and cache audio file durations into audio.json.
@@ -822,7 +829,9 @@ def _generate_slides_json(directory: Path, verbose: bool) -> None:
# Write slides.json only if content changed
output_path = directory / "slides.json"
new_content = json.dumps(sorted_slides, indent=2)
existing_content = output_path.read_text(encoding="utf-8") if output_path.exists() else None
existing_content = (
output_path.read_text(encoding="utf-8") if output_path.exists() else None
)
if new_content != existing_content:
with open(output_path, "w", encoding="utf-8") as f:
f.write(new_content)
@@ -1491,12 +1500,16 @@ def cmd_preprocess(
seg_id, seg_source = future.result()
completed += 1
print(f" Completed: {seg_id} ({completed}/{len(segments_to_process)})")
output_path = (cache_narration_dir or narration_dir) / seg_source.output_file
output_path = (
cache_narration_dir or narration_dir
) / seg_source.output_file
if output_path.exists():
successfully_processed.append((seg_id, seg_source))
else:
for segment_id, segment_source in segments_to_process:
_out_full = (cache_narration_dir or narration_dir) / segment_source.output_file
_out_full = (
cache_narration_dir or narration_dir
) / segment_source.output_file
print(f"\n Processing: {segment_id}")
print(f" Source: {segment_source.source_file}")
print(f" Output: {_out_full}")
@@ -1510,7 +1523,9 @@ def cmd_preprocess(
gnommo_scratch,
res=res,
)
output_path = (cache_narration_dir or narration_dir) / segment_source.output_file
output_path = (
cache_narration_dir or narration_dir
) / segment_source.output_file
if output_path.exists():
successfully_processed.append((segment_id, segment_source))
@@ -1575,7 +1590,13 @@ def cmd_preprocess(
continue
print(f" Processing: {video_id}")
preprocess_video(
videos_dir, video_id, video_source, verbose, force, gnommo_scratch, res=res
videos_dir,
video_id,
video_source,
verbose,
force,
gnommo_scratch,
res=res,
)
print("\nPreprocessing complete.")
@@ -1631,7 +1652,11 @@ def cmd_trim(
for search_dir in (raw_dir, compressed_dir):
if search_dir.exists():
for f in search_dir.iterdir():
if f.is_file() and f.suffix.lower() in _video_exts and not f.name.startswith("."):
if (
f.is_file()
and f.suffix.lower() in _video_exts
and not f.name.startswith(".")
):
stem = f.stem
if stem.endswith("_compressed"):
stem = stem[: -len("_compressed")]
@@ -1658,7 +1683,11 @@ def cmd_trim(
print(f" {seg_id}: source file not found, skipping")
continue
print(f" {seg_id}: analysing {source_path.parent.name}/{source_path.name}...", end="", flush=True)
print(
f" {seg_id}: analysing {source_path.parent.name}/{source_path.name}...",
end="",
flush=True,
)
first_sound, last_sound = detect_silence_bounds(
source_path, noise_threshold_db=threshold_db, verbose=verbose
)
@@ -2137,7 +2166,7 @@ def cmd_stitch(
# per-project settings instead of hardcoded defaults.
_loudnorm_cfg = None
if config and config.default_filters:
for _f in (config.default_filters.get("talkinghead") or []):
for _f in config.default_filters.get("talkinghead") or []:
if isinstance(_f, dict) and _f.get("type") == "audio_normalize":
_loudnorm_cfg = _f
break
@@ -2157,7 +2186,9 @@ def cmd_stitch(
# 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 = videos_dir_out.parent if (res != "full" and not cache_root) else videos_dir
main_videos_dir = (
videos_dir_out.parent if (res != "full" and not cache_root) else videos_dir
)
videos_json_path = main_videos_dir / "videos.json"
if True: # Always update JSON regardless of proxy mode
existing_videos: dict = {}
@@ -2278,10 +2309,18 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
marker_id.startswith(p)
for p in (
"video:",
"vft:", "vfb:", "vf2t:", "vf2b:",
"vst:", "vsb:",
"vftp:", "vfbp:", "vf2tp:", "vf2bp:",
"vstp:", "vsbp:",
"vft:",
"vfb:",
"vf2t:",
"vf2b:",
"vst:",
"vsb:",
"vftp:",
"vfbp:",
"vf2tp:",
"vf2bp:",
"vstp:",
"vsbp:",
)
):
aligned_count += 1
@@ -2289,10 +2328,18 @@ def _print_render_plan_details(plan, marker_timings, slides: dict) -> None:
len(p)
for p in (
"video:",
"vft:", "vfb:", "vf2t:", "vf2b:",
"vst:", "vsb:",
"vftp:", "vfbp:", "vf2tp:", "vf2bp:",
"vstp:", "vsbp:",
"vft:",
"vfb:",
"vf2t:",
"vf2b:",
"vst:",
"vsb:",
"vftp:",
"vfbp:",
"vf2tp:",
"vf2bp:",
"vstp:",
"vsbp:",
)
if marker_id.startswith(p)
)
@@ -2418,7 +2465,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) :]
cutout, layer = implied[0], implied[1]
projection[video_id] = {"cutout": cutout, "layer": layer}
break
@@ -2515,7 +2562,9 @@ def _chunked_render(
import math
# Split slide IDs into groups of chunk_size
groups = [slide_ids[i : i + chunk_size] for i in range(0, len(slide_ids), chunk_size)]
groups = [
slide_ids[i : i + chunk_size] for i in range(0, len(slide_ids), chunk_size)
]
print(
f"\n Auto-chunking: {len(slide_ids)} slides → {len(groups)} chunks of ≤{chunk_size}"
)
@@ -2549,7 +2598,9 @@ def _chunked_render(
chunk_paths.append(chunk_path)
if dry_run:
print(f"\n [dry-run] Would concatenate {len(chunk_paths)} chunks → {final_output}")
print(
f"\n [dry-run] Would concatenate {len(chunk_paths)} chunks → {final_output}"
)
return 0
# Concatenate chunks
@@ -2560,10 +2611,16 @@ def _chunked_render(
f.write(f"file '{p.resolve()}'\n")
concat_cmd = [
"ffmpeg", "-y",
"-f", "concat", "-safe", "0",
"-i", str(concat_list),
"-c", "copy",
"ffmpeg",
"-y",
"-f",
"concat",
"-safe",
"0",
"-i",
str(concat_list),
"-c",
"copy",
str(final_output),
]
result = subprocess.run(concat_cmd, capture_output=True, text=True)
@@ -2639,9 +2696,7 @@ def cmd_render(
# ETL: project shorthand marker semantics (cutout/layer) into videos.json
# before parse_videos reads it, so the render pass is purely data-driven.
_project_markers_to_videos(
markers, project_path / config.videos_path, config
)
_project_markers_to_videos(markers, project_path / config.videos_path, config)
# Override resolution for preview modes
if res != "full":
@@ -2676,7 +2731,11 @@ def cmd_render(
# the renderer doesn't re-resolve it via the local (missing) videos_dir.
if "narration_combined" in videos:
videos["narration_combined"].source_file = str(resolved_combined)
if "narration_combined" in videos and resolved_combined and resolved_combined.exists():
if (
"narration_combined" in videos
and resolved_combined
and resolved_combined.exists()
):
# New workflow: narration_combined was created by 'gnommo concat' and is in videos.json
# This entry has the correct volume setting from videos.json
transcript_path = resolved_combined.with_suffix(".transcript.json")
@@ -2789,8 +2848,6 @@ def cmd_render(
if plan.time_offset > 0:
print(f" Time offset: {plan.time_offset:.1f}s (partial render)")
# Print detailed render plan with alignment info
_print_render_plan_details(plan, marker_timings, slides)
if plan.audio_events:
@@ -2873,12 +2930,20 @@ def cmd_render(
# Check if chunked rendering is needed (avoids filter graph OOM on long videos)
from .cache import get_render_chunk_size
_chunk_size = chunk_slides or get_render_chunk_size() or 0
_slide_ids = [e.slide_id for e in plan.slide_events]
if _chunk_size > 0 and not slide_range and len(_slide_ids) > _chunk_size:
return _chunked_render(
project_path, verbose, dry_run, res, force,
_chunk_size, _slide_ids, out_dir, output_path,
project_path,
verbose,
dry_run,
res,
force,
_chunk_size,
_slide_ids,
out_dir,
output_path,
)
plan.output_path = output_path
@@ -2979,7 +3044,10 @@ def cmd_transcribe(
video_id, video_source = result
video_path = videos_dir / video_source.source_file
if not video_path.exists() and video_source.source_file == "narration_combined.mov":
if (
not video_path.exists()
and video_source.source_file == "narration_combined.mov"
):
found = _resolve_narration_combined(project_path, videos_dir, config)
if found:
video_path = found
@@ -3742,7 +3810,9 @@ def cmd_extract_audio(
# Handle --combined mode: extract from narration_combined.mov
if combined:
videos, videos_dir = parse_videos(project_path, config)
combined_path = _resolve_narration_combined(project_path, videos_dir, config) or (videos_dir / "narration_combined.mov")
combined_path = _resolve_narration_combined(
project_path, videos_dir, config
) or (videos_dir / "narration_combined.mov")
if not combined_path.exists():
print(
@@ -3907,7 +3977,9 @@ def cmd_master(
videos, videos_dir = parse_videos(project_path, config)
# Find narration_combined.mov
combined_path = _resolve_narration_combined(project_path, videos_dir, config) or (videos_dir / "narration_combined.mov")
combined_path = _resolve_narration_combined(project_path, videos_dir, config) or (
videos_dir / "narration_combined.mov"
)
if not combined_path.exists():
print(
f"Error: narration_combined.mov not found at {combined_path}",
gnommo/parser.py
+5 -1
View File
@@ -501,7 +501,11 @@ def parse_videos(
# Handle skip/take - can use begin/end as user-friendly alternatives
skip = float(video_data.get("skip") or 0.0)
take = float(video_data["take"]) if video_data.get("take") not in (None, "") else None
take = (
float(video_data["take"])
if video_data.get("take") not in (None, "")
else None
)
# Convert begin/end to skip/take if provided
if "begin" in video_data and video_data["begin"]:
gnommo/preprocessor.py
+6 -1
View File
@@ -18,9 +18,11 @@ from .models import (
)
from typing import Union, Optional
def _tc() -> str:
"""Return FFmpeg thread count string from ~/.gnommo.conf [performance] cpu_limit."""
from .cache import get_ffmpeg_thread_count
return str(get_ffmpeg_thread_count())
@@ -784,7 +786,10 @@ def apply_combined_video_filters(
cmd.extend(
[
"-probesize", "50000000", "-analyzeduration", "50000000",
"-probesize",
"50000000",
"-analyzeduration",
"50000000",
"-i",
str(input_path),
"-vf",
gnommo/renderer.py
+7 -2
View File
@@ -307,6 +307,7 @@ def build_ffmpeg_command(plan: RenderPlan, output_path: Path) -> list[str]:
# in the filter graph (one per video layer) spawns one swscaler thread per CPU core,
# causing OOM on Apple Silicon where av_cpu_count() returns 10-11.
from .cache import get_ffmpeg_thread_count
_tc = str(get_ffmpeg_thread_count())
cmd.extend(["-threads", _tc, "-filter_threads", _tc])
@@ -463,7 +464,9 @@ def build_ffmpeg_command(plan: RenderPlan, output_path: Path) -> list[str]:
for event in plan.audio_events:
if event.audio_id not in audio_inputs:
if event.audio_def.is_shared and plan.shared_assets_dir:
audio_path = plan.shared_assets_dir / "media" / "audio" / event.audio_def.file
audio_path = (
plan.shared_assets_dir / "media" / "audio" / event.audio_def.file
)
else:
audio_path = audio_dir / event.audio_def.file
audio_path, _ = resolve_with_cache(audio_path, project_path)
@@ -933,7 +936,9 @@ def build_filter_complex(
plan.narration_pauses, plan.total_duration
)
for seg_idx, (src_start, src_end, out_start, out_end) in enumerate(segments):
for seg_idx, (src_start, src_end, out_start, out_end) in enumerate(
segments
):
seg_label = f"av{i}_seg{seg_idx}"
pts_offset = out_start
filters.append(
gnommo/transformer.py
+33 -10
View File
@@ -315,7 +315,9 @@ def _fuzzy_match_ratio(
words_to_check = min(len(phrase_words), window_size)
# Window only needs to cover pre_filler + phrase words + inter_filler slack
transcript_end = min(start_idx + pre_filler + words_to_check + inter_filler, len(transcription))
transcript_end = min(
start_idx + pre_filler + words_to_check + inter_filler, len(transcription)
)
transcript_words = [
_normalize_token(transcription[j].word)
@@ -529,7 +531,10 @@ def align_markers_to_transcription(
else:
prev_idx = seen[timing.marker_id]
prev = deduped[prev_idx]
if prev.context == "(after previous)" and timing.context != "(after previous)":
if (
prev.context == "(after previous)"
and timing.context != "(after previous)"
):
deduped[prev_idx] = MarkerTiming(
marker_id=prev.marker_id,
timestamp=timing.timestamp,
@@ -651,16 +656,28 @@ def build_render_plan(
# Before extracting video events, resolve any referenced videos that are missing
# from the project's videos.json by looking them up in shared_assets/videos.json.
_VIDEO_MARKER_PREFIXES = (
"video:", "narration:", "vft:", "vfb:", "vf2t:", "vf2b:", "vst:", "vsb:",
"vftp:", "vfbp:", "vf2tp:", "vf2bp:", "vstp:", "vsbp:",
"video:",
"narration:",
"vft:",
"vfb:",
"vf2t:",
"vf2b:",
"vst:",
"vsb:",
"vftp:",
"vfbp:",
"vf2tp:",
"vf2bp:",
"vstp:",
"vsbp:",
)
missing_video_ids = [
timing.marker_id[len(prefix):]
timing.marker_id[len(prefix) :]
for timing in marker_timings
if timing.timestamp >= 0
for prefix in _VIDEO_MARKER_PREFIXES
if timing.marker_id.startswith(prefix)
and timing.marker_id[len(prefix):] not in videos
and timing.marker_id[len(prefix) :] not in videos
]
if missing_video_ids:
found = resolve_missing_videos(missing_video_ids, project_path, config)
@@ -676,6 +693,7 @@ def build_render_plan(
)
if video_warnings:
import sys
print("\nWarnings:", file=sys.stderr)
for w in video_warnings:
print(f" ⚠ {w}", file=sys.stderr)
@@ -780,7 +798,10 @@ def build_render_plan(
videos.update(found)
still_missing = [vid_id for vid_id in config.outro if vid_id not in videos]
for vid_id in still_missing:
print(f" WARNING: outro video '{vid_id}' not found in videos.json or shared_assets — skipped", flush=True)
print(
f" WARNING: outro video '{vid_id}' not found in videos.json or shared_assets — skipped",
flush=True,
)
# Build outro events (plays after narration ends)
outro_events = _extract_outro_events(
@@ -997,9 +1018,11 @@ def _extract_video_events(
mid = timing.marker_id
# --- shorthand markers (vft:/vfb:/vst:/vsb: and pause variants) ---
shorthand_match = next((p for p in _SHORTHAND_PREFIXES if mid.startswith(p)), None)
shorthand_match = next(
(p for p in _SHORTHAND_PREFIXES if mid.startswith(p)), None
)
if shorthand_match:
video_id = mid[len(shorthand_match):]
video_id = mid[len(shorthand_match) :]
if video_id not in videos:
warnings.append(
f"[{mid}] references unknown video '{video_id}' — skipped. "
@@ -1052,7 +1075,7 @@ def _extract_video_events(
video_markers.append((timing.timestamp, video_id, "narration", False))
events: list[VideoEvent] = []
for (start_time, video_id, marker_type, pause_narration) in video_markers:
for start_time, video_id, marker_type, pause_narration in video_markers:
video_source = videos[video_id]
# Read cutout and layer directly from videos.json (projected by ETL)