#!/usr/bin/env python3 """Simple flat-folder photo/video movie maker using ffmpeg. Usage: python movmaker.py input --out-dir . Put images, videos, audio files, and optionally data.txt in one directory. """ from __future__ import annotations import argparse import concurrent.futures import json import math import os import random import re import shlex import subprocess import sys import tempfile import unicodedata import urllib.parse import urllib.request from dataclasses import dataclass from datetime import datetime from pathlib import Path IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"} VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"} AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"} DATA_NAMES = {"data.txt", "info.txt", "title.txt"} BASE_DIR = Path(__file__).resolve().parent FFMPEG = BASE_DIR / "ffmpeg" / "ffmpeg" FFPROBE = BASE_DIR / "ffmpeg" / "ffprobe" INTRO_LOGO_NAME = "intro.png" DEFAULT_INTRO_LOGO = BASE_DIR / "img" / "moto_travel.png" INTRO_LOGO_TARGET_HEIGHT_RATIO = 0.90 INTRO_LOGO_EXTRA_HOLD = 6.0 INTRO_LOGO_FADE_OFFSET = 3.0 INTRO_LOGO_DEFAULT_ALPHA = 1.0 INTRO_TITLE_FADE_DELAY = 2.0 INTRO_TITLE_FADE_LENGTH = 1.0 INTRO_TITLE_VISIBLE = 6.0 DEFAULT_IMAGE_DURATION = 6.0 DEFAULT_CROSSFADE_DURATION = 3.0 FIRST_IMAGE_DURATION_MULTIPLIER = 2.0 DEFAULT_AUDIO_FADE_IN = 2.0 DEFAULT_AUDIO_FADE_OUT = 10.0 DEFAULT_VIDEO_FADE = 1.0 INTRO_VIDEO_FADE_ENABLED = False JAMENDO_ENV = Path.home() / ".config" / "movmaker" / "jamendo.env" LOG_PATH: Path | None = None def tool_path(local_path: Path, fallback: str) -> str: """Return the preferred executable path for ffmpeg/ffprobe style tools. Args: local_path: Path to a bundled copy shipped with movmaker. fallback: Name of the system executable to call when the bundled copy is missing. Returns: The fully-qualified path to execute. """ return str(local_path) if local_path.exists() else fallback @dataclass class MovieData: title: str = "" date: str = "" place: str = "" description: str = "" captions: dict[str, str] | None = None video_audio: set[str] | None = None @dataclass class MediaItem: path: Path kind: str # image or video duration: float def log_message(message: str) -> None: """Append a diagnostic line to the current render log if it is enabled.""" if LOG_PATH: with LOG_PATH.open("a", encoding="utf-8") as log: log.write(message.rstrip() + "\n") def ffmpeg_progress_cmd(cmd: list[str], progress: bool) -> list[str]: """Inject ffmpeg-friendly progress flags when progress monitoring is requested.""" if not cmd or Path(cmd[0]).name != "ffmpeg": return cmd out = [cmd[0], "-hide_banner", "-nostats"] rest = cmd[1:] if rest and rest[0] == "-y": out.append("-y") rest = rest[1:] if progress: out += ["-progress", "pipe:1"] return out + rest def format_bytes(value: int) -> str: size = float(max(0, value)) for unit in ["B", "KB", "MB", "GB"]: if size < 1024 or unit == "GB": return f"{size:.1f}{unit}" size /= 1024 return f"{size:.1f}GB" def run(cmd: list[str], progress_duration: float | None = None) -> None: """Execute a subprocess while streaming progress and teeing output to the log. Args: cmd: argv array ready for subprocess.run. progress_duration: Expected output duration in seconds used to estimate progress; omit for instant commands. """ progress = progress_duration is not None and progress_duration > 0 and Path(cmd[0]).name == "ffmpeg" run_cmd = ffmpeg_progress_cmd(cmd, progress) log_message("+ " + shlex.join(run_cmd)) if not LOG_PATH: subprocess.run(run_cmd, check=True) return with LOG_PATH.open("a", encoding="utf-8") as log: if not progress: subprocess.run(run_cmd, stdout=log, stderr=log, check=True) return process = subprocess.Popen(run_cmd, stdout=subprocess.PIPE, stderr=log, text=True, bufsize=1) fields: dict[str, str] = {} assert process.stdout is not None for raw_line in process.stdout: line = raw_line.strip() log.write(line + "\n") if "=" not in line: continue key, value = line.split("=", 1) fields[key] = value if key in {"out_time_ms", "total_size", "speed", "fps", "progress"}: try: seconds = int(fields.get("out_time_ms", "0")) / 1_000_000 except ValueError: seconds = 0.0 try: written = int(fields.get("total_size", "0")) except ValueError: written = 0 percent = min(100.0, max(0.0, seconds / progress_duration * 100.0)) fps = fields.get("fps", "?") speed = fields.get("speed", "?") sys.stdout.write(f"\r[{timestamp()}] Rendering {percent:5.1f}% fps={fps} speed={speed} written={format_bytes(written)}") sys.stdout.flush() rc = process.wait() sys.stdout.write("\n") if rc != 0: raise subprocess.CalledProcessError(rc, run_cmd) def capture(cmd: list[str]) -> str: return subprocess.check_output(cmd, text=True).strip() def require_tool(name: str) -> None: try: executable = tool_path(FFMPEG, "ffmpeg") if name == "ffmpeg" else tool_path(FFPROBE, "ffprobe") if name == "ffprobe" else name subprocess.run([executable, "-version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) except Exception: raise SystemExit(f"Error: required tool not found: {name}") def parse_resolution(value: str) -> tuple[int, int]: try: w, h = value.lower().split("x", 1) return int(w), int(h) except Exception: raise argparse.ArgumentTypeError("resolution must look like 1920x1080") def ffprobe_duration(path: Path) -> float: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-show_entries", "format=duration", "-of", "json", str(path), ]) data = json.loads(out) return max(0.1, float(data["format"]["duration"])) except Exception: return 3.0 def ffprobe_resolution(path: Path) -> tuple[int, int] | None: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height", "-of", "json", str(path), ]) data = json.loads(out) for stream in data.get("streams", []): width = int(stream.get("width") or 0) height = int(stream.get("height") or 0) if width > 0 and height > 0: return width, height except Exception: return None return None def ffprobe_creation_date(path: Path) -> datetime | None: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-show_entries", "format_tags=creation_time:stream_tags=creation_time", "-of", "json", str(path), ]) data = json.loads(out) candidates: list[str] = [] if isinstance(data.get("format"), dict): candidates.append(data["format"].get("tags", {}).get("creation_time", "")) for stream in data.get("streams", []): candidates.append(stream.get("tags", {}).get("creation_time", "")) for value in candidates: if not value: continue try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: pass except Exception: pass return None def image_tiff_data(path: Path) -> bytes | None: data = path.read_bytes()[:1024 * 1024] if data.startswith(b"\xff\xd8"): pos = 2 while pos + 4 <= len(data): if data[pos] != 0xFF: break marker = data[pos + 1] pos += 2 if marker in {0xD8, 0xD9}: continue if pos + 2 > len(data): break size = int.from_bytes(data[pos:pos + 2], "big") segment = data[pos + 2:pos + size] pos += size if marker == 0xE1 and segment.startswith(b"Exif\x00\x00"): return segment[6:] return None return data def exif_image_date(path: Path) -> datetime | None: """Read common JPEG/TIFF EXIF date tags using only the stdlib.""" tiff = image_tiff_data(path) if tiff is None: return None if tiff[:2] == b"II": endian = "little" elif tiff[:2] == b"MM": endian = "big" else: return None def u16(off: int) -> int: return int.from_bytes(tiff[off:off + 2], endian) def u32(off: int) -> int: return int.from_bytes(tiff[off:off + 4], endian) def read_ascii(offset: int, count: int) -> str: return tiff[offset:offset + count].split(b"\x00", 1)[0].decode("ascii", "ignore").strip() def parse_ifd(ifd_off: int) -> tuple[str | None, int | None]: if ifd_off <= 0 or ifd_off + 2 > len(tiff): return None, None entries = u16(ifd_off) found_date = None exif_ifd = None for i in range(entries): off = ifd_off + 2 + i * 12 if off + 12 > len(tiff): break tag = u16(off) typ = u16(off + 2) count = u32(off + 4) value = u32(off + 8) if tag in {0x0132, 0x9003, 0x9004} and typ == 2: found_date = read_ascii(value, count) if count > 4 else tiff[off + 8:off + 8 + count].split(b"\x00", 1)[0].decode("ascii", "ignore") elif tag == 0x8769: exif_ifd = value return found_date, exif_ifd try: if u16(2) != 42: return None date_text, exif_ifd = parse_ifd(u32(4)) if exif_ifd: exif_date, _ = parse_ifd(exif_ifd) date_text = exif_date or date_text if date_text: return datetime.strptime(date_text[:19], "%Y:%m:%d %H:%M:%S") except Exception: return None return None def exif_image_gps(path: Path) -> tuple[float, float] | None: tiff = image_tiff_data(path) if tiff is None: return None if tiff[:2] == b"II": endian = "little" elif tiff[:2] == b"MM": endian = "big" else: return None def u16(off: int) -> int: return int.from_bytes(tiff[off:off + 2], endian) def u32(off: int) -> int: return int.from_bytes(tiff[off:off + 4], endian) def rational(off: int) -> float: num = u32(off) den = u32(off + 4) return num / den if den else 0.0 def ascii_value(value: int, count: int, entry_off: int) -> str: raw = tiff[entry_off + 8:entry_off + 8 + count] if count <= 4 else tiff[value:value + count] return raw.split(b"\x00", 1)[0].decode("ascii", "ignore").strip() def rationals(value: int, count: int) -> list[float]: return [rational(value + i * 8) for i in range(count)] try: if u16(2) != 42: return None ifd0 = u32(4) entries = u16(ifd0) gps_ifd = None for i in range(entries): off = ifd0 + 2 + i * 12 if off + 12 > len(tiff): break if u16(off) == 0x8825: gps_ifd = u32(off + 8) break if gps_ifd is None or gps_ifd + 2 > len(tiff): return None gps_entries = u16(gps_ifd) lat_ref = lon_ref = "" lat = lon = None for i in range(gps_entries): off = gps_ifd + 2 + i * 12 if off + 12 > len(tiff): break tag = u16(off) typ = u16(off + 2) count = u32(off + 4) value = u32(off + 8) if tag == 1 and typ == 2: lat_ref = ascii_value(value, count, off) elif tag == 2 and typ == 5: lat = rationals(value, count) elif tag == 3 and typ == 2: lon_ref = ascii_value(value, count, off) elif tag == 4 and typ == 5: lon = rationals(value, count) if not lat or not lon or len(lat) < 3 or len(lon) < 3: return None lat_dec = lat[0] + lat[1] / 60 + lat[2] / 3600 lon_dec = lon[0] + lon[1] / 60 + lon[2] / 3600 if lat_ref.upper() == "S": lat_dec = -lat_dec if lon_ref.upper() == "W": lon_dec = -lon_dec return lat_dec, lon_dec except Exception: return None def ffprobe_gps(path: Path) -> tuple[float, float] | None: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path), ]) data = json.loads(out) except Exception: return None values: list[str] = [] def collect(obj: object) -> None: if isinstance(obj, dict): tags = obj.get("tags") if isinstance(tags, dict): for key, value in tags.items(): if "location" in key.lower() or "gps" in key.lower(): values.append(str(value)) for value in obj.values(): collect(value) elif isinstance(obj, list): for value in obj: collect(value) collect(data) for value in values: match = re.search(r"([+-]\d+(?:\.\d+)?)([+-]\d+(?:\.\d+)?)", value) if match: return float(match.group(1)), float(match.group(2)) return None def media_gps(path: Path) -> tuple[float, float] | None: if path.suffix.lower() in IMAGE_EXTS: gps = exif_image_gps(path) if gps: return gps return ffprobe_gps(path) def reverse_geocode_city_country(lat: float, lon: float) -> str | None: last_country = "" for zoom in range(18, 2, -1): query = urllib.parse.urlencode({ "format": "jsonv2", "lat": f"{lat:.8f}", "lon": f"{lon:.8f}", "zoom": str(zoom), "addressdetails": "1", }) request = urllib.request.Request( f"https://nominatim.openstreetmap.org/reverse?{query}", headers={"User-Agent": "movmaker/1.0 (reverse geocoding for local movie metadata)"}, ) try: with urllib.request.urlopen(request, timeout=10) as response: data = json.loads(response.read().decode("utf-8")) address = data.get("address", {}) country = address.get("country", "") last_country = country or last_country city = next((address.get(key) for key in ("city", "town", "village") if address.get(key)), "") if city: return ", ".join(x for x in [city, country] if x) except Exception: continue return last_country or None def infer_place_from_gps(media: list[MediaItem]) -> str | None: for item in media: gps = media_gps(item.path) if not gps: continue place = reverse_geocode_city_country(*gps) if place: console(f"Location from GPS: {place} ({gps[0]:.6f}, {gps[1]:.6f})") return place return None def media_embedded_date(path: Path) -> datetime | None: ext = path.suffix.lower() if ext in IMAGE_EXTS: try: dt = exif_image_date(path) if dt is not None: return dt except Exception: pass return ffprobe_creation_date(path) def media_date(path: Path) -> datetime: dt = media_embedded_date(path) if dt is None: dt = datetime.fromtimestamp(path.stat().st_mtime) if dt.tzinfo is not None: dt = dt.astimezone().replace(tzinfo=None) return dt def ascii_safe(value: str) -> str: """Return a human-readable ASCII approximation of Scandinavian/Slavic letters.""" replacements = { # Danish letters "æ": "ae", "Æ": "Ae", "ø": "o", "Ø": "O", "å": "aa", "Å": "Aa", # German letters "ä": "ae", "Ä": "Ae", "ö": "oe", "Ö": "Oe", "ü": "ue", "Ü": "Ue", "ß": "ss", "ẞ": "SS", # Croatian letters "č": "c", "Č": "C", "ć": "c", "Ć": "C", "ž": "z", "Ž": "Z", "š": "s", "Š": "S", "đ": "d", "Đ": "D", } value = "".join(replacements.get(ch, ch) for ch in value) return unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii") def slugify(value: str) -> str: """Convert an arbitrary string into a filesystem-safe slug.""" value = ascii_safe(value) value = value.strip().lower() value = re.sub(r"[^a-z0-9]+", "_", value) value = value.strip("_") return value or "movie" def ordinal_day(day: int) -> str: if 10 <= day % 100 <= 20: suffix = "th" else: suffix = {1: "st", 2: "nd", 3: "rd"}.get(day % 10, "th") return f"{day}{suffix}" def display_date(dt: datetime) -> str: return f"{dt.strftime('%B')} {dt.year}" def prepare_clip_stamps(media: list[MediaItem], data: MovieData) -> list[str]: """Build per-clip stamp strings containing title/date/place metadata.""" geocode_cache: dict[tuple[float, float], str | None] = {} stamps: list[str] = [] for item in media: embedded_date = media_embedded_date(item.path) date_text = display_date(embedded_date) if embedded_date else data.date place_text = data.place gps = media_gps(item.path) if gps: key = (round(gps[0], 4), round(gps[1], 4)) if key not in geocode_cache: geocode_cache[key] = reverse_geocode_city_country(*gps) place_text = geocode_cache[key] or place_text stamp = " | ".join(x for x in [data.title, date_text, place_text] if x) stamps.append(stamp) if stamp: console(f"Clip stamp: {item.path.name}: {ascii_safe(stamp)}") return stamps def timestamp() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def console(message: str) -> None: print(f"[{timestamp()}] {ascii_safe(message)}") def default_output_path(first_media: Path, data: MovieData, out_dir: Path) -> Path: date = media_date(first_media).strftime("%Y%m%d") title = slugify(data.title or first_media.parent.name or "movie") return out_dir / f"{date}_{title}.mp4" def natural_key(path: Path) -> list[object]: # Simple natural-ish sort: IMG_2 before IMG_10. import re parts = re.split(r"(\d+)", path.name.lower()) return [int(p) if p.isdigit() else p for p in parts] def scan_input(input_dir: Path, image_duration: float) -> tuple[list[MediaItem], list[Path], MovieData]: """Detect media/audio/data files inside the input directory and build metadata.""" if not input_dir.is_dir(): raise SystemExit(f"Input directory does not exist: {input_dir}") media: list[MediaItem] = [] audio: list[Path] = [] data_file: Path | None = None for path in sorted(input_dir.iterdir(), key=natural_key): if not path.is_file(): continue ext = path.suffix.lower() if path.name.lower() in DATA_NAMES: data_file = path elif ext in IMAGE_EXTS: media.append(MediaItem(path=path, kind="image", duration=image_duration)) elif ext in VIDEO_EXTS: media.append(MediaItem(path=path, kind="video", duration=ffprobe_duration(path))) elif ext in AUDIO_EXTS: audio.append(path) if not media: raise SystemExit(f"No images or videos found in {input_dir}") media.sort(key=lambda item: (media_date(item.path), natural_key(item.path))) return media, audio, parse_data_file(data_file) def parse_data_file(path: Path | None) -> MovieData: if not path or not path.exists(): return MovieData(captions={}) data = MovieData(captions={}) lines = path.read_text(encoding="utf-8", errors="replace").splitlines() i = 0 while i < len(lines): line_no = i + 1 line = lines[i].strip() if not line or line.startswith("#"): i += 1 continue if ":" not in line: raise SystemExit(f"Invalid data file syntax in {path} line {line_no}: every entry must be key: value") key, value = line.split(":", 1) key = key.strip() value = value.strip() low = key.lower() if value == "|" and low in {"description", "desc", "synopsis"}: block: list[str] = [] i += 1 while i < len(lines): raw = lines[i] if raw.strip() and not raw.startswith((" ", "\t")): break if raw.startswith(" "): raw = raw[2:] elif raw.startswith((" ", "\t")): raw = raw[1:] block.append(raw.rstrip()) i += 1 data.description = "\n".join(block).strip() continue if low in {"title", "name"}: data.title = value elif low in {"date", "dates", "when"}: data.date = value elif low in {"place", "location", "where"}: data.place = value elif low in {"description", "desc", "synopsis"}: data.description = value elif low.endswith(" audio"): file_key = key[:-6].strip() if file_key and value.lower() in {"1", "yes", "true", "on"}: if data.video_audio is None: data.video_audio = set() data.video_audio.add(file_key) elif key: data.captions[key] = value i += 1 return data def strip_unsupported_caption_chars(text: str) -> str: """Remove characters that bundled caption fonts cannot render reliably. FFmpeg drawtext does not provide dependable emoji fallback for our bundled caption fonts, so emoji/symbol glyphs become unknown-character boxes in the final video. Keep normal letters, numbers, punctuation, spacing, and marks; drop symbols such as emoji. """ cleaned = [] for ch in text: cat = unicodedata.category(ch) if cat.startswith("C"): continue if cat.startswith("S"): continue cleaned.append(ch) return re.sub(r"\s+", " ", "".join(cleaned)).strip() def drawtext_escape(text: str) -> str: # Escape for ffmpeg drawtext text=... without surrounding quotes. return ( text.replace("\\", "\\\\") .replace(":", "\\:") .replace(",", "\\,") .replace("%", "\\%") ) def render_intro_logo_frame(logo: Path, out: Path, width: int, height: int, alpha: float) -> None: """Build a single black frame with the intro logo centered and scaled. Args: logo: Source PNG to overlay. out: Destination PNG path for the composed frame. width: Target video width. height: Target video height. alpha: Desired opacity between 0 and 1. """ if width <= 0 or height <= 0: raise ValueError("intro logo frame must have positive width/height") alpha = min(1.0, max(0.0, alpha)) target_logo_height = max(1, int(round(height * INTRO_LOGO_TARGET_HEIGHT_RATIO))) filters = ( f"[1:v]scale=w={width}:h={target_logo_height}:force_original_aspect_ratio=decrease," f"format=rgba,colorchannelmixer=aa={alpha:.3f}[logo];" f"[0:v][logo]overlay=x=(W-w)/2:y=(H-h)/2:format=auto,format=rgba[vout]" ) cmd = [ tool_path(FFMPEG, "ffmpeg"), "-y", "-f", "lavfi", "-i", f"color=c=black:s={width}x{height}", "-i", str(logo), "-filter_complex", filters, "-map", "[vout]", "-frames:v", "1", "-pix_fmt", "rgba", "-f", "image2", "-update", "1", str(out), ] run(cmd) def normalize_clip(item: MediaItem, out: Path, width: int, height: int, fps: int, preset: str, crf: int) -> None: """Transcode one media item into a padded, FPS-normalized H.264 clip.""" vf = ( f"scale={width}:{height}:force_original_aspect_ratio=decrease," f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black," f"fps={fps},format=yuv420p,setsar=1" ) if item.kind == "image": cmd = [ tool_path(FFMPEG, "ffmpeg"), "-y", "-loop", "1", "-t", f"{item.duration:.3f}", "-i", str(item.path), "-an", "-vf", vf, "-c:v", "libx264", "-preset", preset, "-crf", str(crf), str(out), ] else: cmd = [ tool_path(FFMPEG, "ffmpeg"), "-y", "-i", str(item.path), "-an", "-vf", vf, "-c:v", "libx264", "-preset", preset, "-crf", str(crf), str(out), ] run(cmd) def media_has_audio(path: Path) -> bool: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=index", "-of", "csv=p=0", str(path), ]) return bool(out.strip()) except Exception: return False def audio_metadata(path: Path) -> tuple[str, str]: try: out = capture([ tool_path(FFPROBE, "ffprobe"), "-v", "error", "-show_entries", "format_tags=artist,title", "-of", "json", str(path), ]) tags = json.loads(out).get("format", {}).get("tags", {}) artist = tags.get("artist") or tags.get("ARTIST") or "" title = tags.get("title") or tags.get("TITLE") or "" return str(artist), str(title) except Exception: return "", "" def load_jamendo_client_id() -> str | None: if os.environ.get("JAMENDO_CLIENT_ID"): return os.environ["JAMENDO_CLIENT_ID"] if JAMENDO_ENV.exists(): for line in JAMENDO_ENV.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) if key.strip() == "JAMENDO_CLIENT_ID": return value.strip().strip("'\"") return None def download_jamendo_music(genre: str, out_dir: Path) -> Path | None: """Download a random Jamendo track matching the requested genre.""" client_id = load_jamendo_client_id() if not client_id: console("No audio files found and no Jamendo client ID configured; continuing without music.") return None query = urllib.parse.urlencode({ "client_id": client_id, "format": "json", "limit": "50", "search": genre, "include": "musicinfo licenses", "audioformat": "mp32", "ccnc": "true", "ccnd": "false", }) request = urllib.request.Request( f"https://api.jamendo.com/v3.0/tracks/?{query}", headers={"User-Agent": "movmaker/1.0"}, ) try: with urllib.request.urlopen(request, timeout=20) as response: data = json.loads(response.read().decode("utf-8")) except Exception as exc: console(f"Could not search Jamendo music: {exc}") return None results = [] for track in data.get("results", []): license_url = str(track.get("license_ccurl", "")).lower() if not track.get("audio"): continue if track.get("audiodownload_allowed") is False: continue if "creativecommons.org" not in license_url or "/nd" in license_url: continue results.append(track) if not results: console(f"No Jamendo music found for genre/query: {genre}") return None track = random.choice(results) audio_url = track["audio"] title = track.get("name") or "jamendo_music" artist = track.get("artist_name", "") filename = f"jamendo_{track.get('id', 'music')}_{slugify(title)}.mp3" out_dir.mkdir(parents=True, exist_ok=True) out = out_dir / filename console(f"Downloading Jamendo music: {title}" + (f" by {artist}" if artist else "")) try: urllib.request.urlretrieve(audio_url, out) except Exception as exc: console(f"Could not download Jamendo music: {exc}") return None if title or artist: console("Audio metadata: " + " - ".join(x for x in [artist, title] if x)) log_dir = out_dir / ".logs" log_dir.mkdir(parents=True, exist_ok=True) attribution = log_dir / f"{out.stem}.txt" musicinfo = track.get("musicinfo", {}) if isinstance(track.get("musicinfo"), dict) else {} attribution.write_text( "Jamendo music downloaded by movmaker\n" f"Title: {title}\n" f"Artist: {artist}\n" f"URL: {track.get('shareurl', '')}\n" f"License: {track.get('license_ccurl', '')}\n" "Source: Jamendo\n", encoding="utf-8", ) return out def concat_or_mix_audio(audio_files: list[Path], out: Path) -> Path | None: """Concatenate or mix multiple audio tracks into one temp file.""" if not audio_files: return None if len(audio_files) == 1: return audio_files[0] list_file = out.with_suffix(".txt") def esc(p: Path) -> str: return str(p.resolve()).replace("'", "'\\''") list_file.write_text("".join(f"file '{esc(p)}'\n" for p in audio_files), encoding="utf-8") run([ tool_path(FFMPEG, "ffmpeg"), "-y", "-f", "concat", "-safe", "0", "-i", str(list_file), "-vn", "-c:a", "aac", "-b:a", "192k", str(out), ]) return out def build_video( clips: list[Path], durations: list[float], audio: Path | None, clip_audio_paths: list[Path | None], output: Path, data: MovieData, creation_date: datetime, captions: list[str], clip_stamps: list[str], fade: float, audio_fade: float, audio_fade_in: float, video_fade: float, intro_logo: Path | None, intro_logo_alpha: float, width: int, height: int, fps: int, preset: str, crf: int, text_dir: Path, video_intro_duration: float | None = None, intro_visible_lead: float = 0.0, skip_video_fade_out: bool = False, ) -> None: """Assemble normalized clips, overlays, and audio into the final MP4.""" output.parent.mkdir(parents=True, exist_ok=True) text_file_count = 0 def textfile_arg(text: str) -> str: nonlocal text_file_count path = text_dir / f"drawtext_{text_file_count:04d}.txt" text_file_count += 1 path.write_text(text, encoding="utf-8") return f"textfile='{path}':expansion=none" cmd = [tool_path(FFMPEG, "ffmpeg"), "-y"] for clip in clips: cmd += ["-i", str(clip)] if audio: cmd += ["-stream_loop", "-1", "-i", str(audio)] clip_audio_input_start = len(clips) + (1 if audio else 0) for clip_audio in clip_audio_paths: if clip_audio is not None: cmd += ["-i", str(clip_audio)] filters: list[str] = [] n = len(clips) total_duration = durations[0] transition_offsets: list[float] = [] if n == 1: current = "0:v" else: current = "v01" offset = max(0.0, durations[0] - fade) transition_offsets.append(offset) filters.append(f"[0:v][1:v]xfade=transition=fade:duration={fade:.3f}:offset={offset:.3f}[{current}]") total_duration = durations[0] + durations[1] - fade for i in range(2, n): next_label = f"v{i:02d}" offset = max(0.0, total_duration - fade) transition_offsets.append(offset) filters.append(f"[{current}][{i}:v]xfade=transition=fade:duration={fade:.3f}:offset={offset:.3f}[{next_label}]") current = next_label total_duration += durations[i] - fade intro_duration = max(0.0, video_intro_duration) if video_intro_duration is not None else (audio_fade_in if audio_fade_in > 0 else 0.0) if intro_duration > 0: intro_label = "intro_padded" filters.append( f"[{current}]tpad=start_duration={intro_duration:.3f}:start_mode=add:color=black[{intro_label}]" ) current = intro_label total_duration += intro_duration fade_label = current apply_fade_in = video_fade > 0 and INTRO_VIDEO_FADE_ENABLED apply_fade_out = video_fade > 0 and not skip_video_fade_out if apply_fade_in or apply_fade_out: fade_in_start = intro_duration fade_out_start = max(0.0, total_duration - video_fade) fade_label = "video_faded" fade_parts = [] if apply_fade_in: fade_parts.append(f"fade=t=in:st={fade_in_start:.3f}:d={video_fade:.3f}") if apply_fade_out: fade_parts.append(f"fade=t=out:st={fade_out_start:.3f}:d={video_fade:.3f}") filters.append(f"[{current}]" + ",".join(fade_parts) + f"[{fade_label}]") current = fade_label title_lines = [x for x in [data.title, data.date, data.place] if x] final_label = "vout" title_label = "titled" title_font = str(Path.home() / ".local/share/fonts/google/BebasNeue-Regular.ttf") stamp_font = "/usr/share/fonts/truetype/noto/NotoSans-Bold.ttf" caption_font = "/usr/share/fonts/truetype/ibm-plex/IBMPlexSans-SemiBold.ttf" if title_lines: title_font_size = max(60, height // 10) subtitle_font_size = max(40, height // 18) main_title = data.title or title_lines[0] subtitle = " | ".join(x for x in [data.date, data.place] if x) fade_in_start = INTRO_TITLE_FADE_DELAY fade_in_end = fade_in_start + INTRO_TITLE_FADE_LENGTH visible_end = fade_in_end + INTRO_TITLE_VISIBLE fade_out_end = visible_end + INTRO_TITLE_FADE_LENGTH title_alpha = ( f"if(lt(t\\,{fade_in_start:.3f})\\,0\\," f"if(lt(t\\,{fade_in_end:.3f})\\,(t-{fade_in_start:.3f})/{INTRO_TITLE_FADE_LENGTH:.3f}\\," f"if(lt(t\\,{visible_end:.3f})\\,1\\," f"if(lt(t\\,{fade_out_end:.3f})\\,({fade_out_end:.3f}-t)/{INTRO_TITLE_FADE_LENGTH:.3f}\\,0))))" ) title_enable = f"enable='between(t,{fade_in_start:.3f},{fade_out_end:.3f})'" if subtitle: title_y = f"(h-text_h)/2-{title_font_size // 2}" filters.append( f"[{current}]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2:y={title_y}:" f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:" f"alpha='{title_alpha}':{title_enable}[title0]" ) filters.append( f"[title0]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2+1:y={title_y}:" f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:" f"alpha='{title_alpha}':{title_enable}[title0b]" ) filters.append( f"[title0b]drawtext=fontfile='{title_font}':{textfile_arg(subtitle)}:x=(w-text_w)/2:y=(h-text_h)/2+{subtitle_font_size}:" f"fontsize={subtitle_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=2:shadowy=2:" f"alpha='{title_alpha}':{title_enable}[{title_label}]" ) else: filters.append( f"[{current}]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2:y=(h-text_h)/2:" f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:" f"alpha='{title_alpha}':{title_enable}[title0]" ) filters.append( f"[title0]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2+1:y=(h-text_h)/2:" f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:" f"alpha='{title_alpha}':{title_enable}[{title_label}]" ) else: filters.append(f"[{current}]copy[{title_label}]") caption_label = title_label caption_size = max(24, int((subtitle_font_size if title_lines else max(40, height // 18)) * 0.7)) for idx, caption in enumerate(captions): if not caption: continue if n == 1: start = intro_duration end = total_duration elif idx == 0: start = intro_duration end = intro_duration + transition_offsets[0] elif idx == n - 1: start = intro_duration + transition_offsets[idx - 1] + fade end = total_duration else: start = intro_duration + transition_offsets[idx - 1] + fade end = intro_duration + transition_offsets[idx] if end <= start: continue caption_fade = min(1.0, max(0.0, (end - start) / 2)) caption_alpha = ( f"if(lt(t\\,{start + caption_fade:.3f})\\,(t-{start:.3f})/{caption_fade:.3f}\\," f"if(lt(t\\,{end - caption_fade:.3f})\\,1\\,({end:.3f}-t)/{caption_fade:.3f}))" ) if caption_fade > 0 else "1" caption_lines = [] for line in caption.split("|"): clean_line = strip_unsupported_caption_chars(line.strip()) if clean_line: caption_lines.append(clean_line) line_gap = int(caption_size * 1.2) block_height = line_gap * (len(caption_lines) - 1) for line_idx, caption_line in enumerate(caption_lines): y_expr = f"h-text_h-62-{block_height - line_idx * line_gap}" next_label = f"caption{idx}_{line_idx}" filters.append( f"[{caption_label}]drawtext=fontfile='{caption_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2:y={y_expr}:" f"fontsize={caption_size}:fontcolor=0xF2F2EE:borderw=0:shadowcolor=black@0.80:shadowx=3:shadowy=3:" f"alpha='{caption_alpha}':enable='between(t,{start:.3f},{end:.3f})'[{next_label}]" ) caption_label = next_label stamp_size = max(16, height // 58) stamp_left_label = caption_label for idx, stamp_text in enumerate(clip_stamps): if not stamp_text: continue if n == 1: start = intro_duration end = total_duration elif idx == 0: start = intro_duration end = intro_duration + transition_offsets[0] + fade / 2 elif idx == n - 1: start = intro_duration + transition_offsets[idx - 1] + fade / 2 end = total_duration else: start = intro_duration + transition_offsets[idx - 1] + fade / 2 end = intro_duration + transition_offsets[idx] + fade / 2 if end <= start: continue next_label = f"stamp_left_{idx}" filters.append( f"[{stamp_left_label}]drawtext=fontfile='{stamp_font}':{textfile_arg(stamp_text)}:x=24:y=h-text_h-18:" f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black:enable='between(t,{start:.3f},{end:.3f})'[{next_label}]" ) stamp_left_label = next_label filters.append( f"[{stamp_left_label}]drawtext=fontfile='{stamp_font}':{textfile_arg('Bubulescu.Org')}:x=w-text_w-24:y=h-text_h-18:" f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black[{final_label}]" ) audio_mix_labels: list[str] = [] if audio: audio_index = len(clips) audio_fade_out_start = max(0.0, total_duration - audio_fade) filters.append( f"[{audio_index}:a:0]atrim=0:{total_duration:.3f},asetpts=PTS-STARTPTS," f"afade=t=in:st=0:d={audio_fade_in:.3f},afade=t=out:st={audio_fade_out_start:.3f}:d={audio_fade:.3f}[bg_audio]" ) audio_mix_labels.append("[bg_audio]") clip_audio_input = clip_audio_input_start for idx, clip_audio in enumerate(clip_audio_paths): if clip_audio is None: continue if n == 1: start = intro_duration end = total_duration elif idx == 0: start = intro_duration end = intro_duration + transition_offsets[0] elif idx == n - 1: start = intro_duration + transition_offsets[idx - 1] + fade end = total_duration else: start = intro_duration + transition_offsets[idx - 1] + fade end = intro_duration + transition_offsets[idx] segment = max(0.0, end - start) if segment <= 0: clip_audio_input += 1 continue fade_len = min(fade, segment / 2) delay_ms = max(0, int(start * 1000)) label = f"clip_audio_{idx}" filters.append( f"[{clip_audio_input}:a:0]atrim=0:{segment:.3f},asetpts=PTS-STARTPTS," f"afade=t=in:st=0:d={fade_len:.3f},afade=t=out:st={max(0.0, segment - fade_len):.3f}:d={fade_len:.3f}," f"adelay={delay_ms}:all=1[{label}]" ) audio_mix_labels.append(f"[{label}]") clip_audio_input += 1 if audio_mix_labels: filters.append("".join(audio_mix_labels) + f"amix=inputs={len(audio_mix_labels)}:duration=longest:normalize=0[aout]") cmd += ["-filter_complex", ";".join(filters), "-map", f"[{final_label}]"] metadata = { "title": data.title, "date": data.date, "location": data.place, "place": data.place, "com.apple.quicktime.location.name": data.place, "description": data.description, "synopsis": data.description, "artist": "Bubulescu.Org", "comment": "Created with movmaker", "creation_time": creation_date.isoformat(), } for key, value in metadata.items(): if value: cmd += ["-metadata", f"{key}={value}"] if audio_mix_labels: cmd += ["-map", "[aout]", "-c:a", "aac", "-b:a", "192k", "-shortest"] else: cmd += ["-an"] cmd += ["-c:v", "libx264", "-preset", preset, "-crf", str(crf), "-r", str(fps), "-pix_fmt", "yuv420p", "-movflags", "+faststart+use_metadata_tags", str(output)] run(cmd, total_duration) def main(argv: list[str] | None = None) -> int: """CLI entry point for movmaker.""" parser = argparse.ArgumentParser(description="Make a video from a flat folder of images/videos/audio.") parser.add_argument("input", type=Path, help="flat input directory") parser.add_argument("--out-dir", type=Path, default=Path("."), help="directory for auto-named output videos") parser.add_argument("--image-duration", type=float, default=DEFAULT_IMAGE_DURATION, help="seconds each image is visible") parser.add_argument("--fade", type=float, default=DEFAULT_CROSSFADE_DURATION, help="crossfade duration in seconds") parser.add_argument("--audio-fade", type=float, default=DEFAULT_AUDIO_FADE_OUT, help="music fade-out duration in seconds") parser.add_argument("--audio-fade-in", type=float, default=DEFAULT_AUDIO_FADE_IN, help="music fade-in duration in seconds") parser.add_argument("--video-fade", type=float, default=DEFAULT_VIDEO_FADE, help="video fade-in/fade-out duration in seconds") parser.add_argument("--intro-logo", type=Path, default=DEFAULT_INTRO_LOGO, help="PNG logo shown centered over the black intro") parser.add_argument("--intro-logo-alpha", type=float, default=INTRO_LOGO_DEFAULT_ALPHA, help="opacity for the intro logo, from 0 to 1") parser.add_argument("--resolution", type=parse_resolution, default=(1920, 1080), help="output resolution, e.g. 1920x1080") parser.add_argument("--fps", type=int, default=30, help="output frames per second") parser.add_argument("--preset", default="veryfast", help="x264 speed preset, e.g. ultrafast, superfast, veryfast") parser.add_argument("--crf", type=int, default=20, help="x264 quality; lower is better/larger, higher is faster/smaller") parser.add_argument("--jobs", type=int, default=1, help="parallel clip preparation jobs; try 2 on Raspberry Pi") parser.add_argument("--preview", action="store_true", help="fast preview: 1280x720, 24 fps, ultrafast, CRF 28") parser.add_argument("--music-genre", default="cinematic punk rock", help="Jamendo music search query used only when input has no audio files") args = parser.parse_args(argv) global LOG_PATH log_dir = args.input / ".logs" log_dir.mkdir(parents=True, exist_ok=True) LOG_PATH = log_dir / f"movmaker_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log" LOG_PATH.write_text("movmaker log\n", encoding="utf-8") require_tool("ffmpeg") require_tool("ffprobe") width, height = args.resolution fps = args.fps preset = args.preset crf = args.crf if args.preview: width, height = (1280, 720) fps = 20 preset = "ultrafast" crf = 28 media, audio_files, data = scan_input(args.input, args.image_duration) first_media_date = media_date(media[0].path) first_media_resolution = ffprobe_resolution(media[0].path) if not data.date: data.date = display_date(first_media_date) if not data.place: data.place = infer_place_from_gps(media) or "" fade = min(args.fade, max(0.0, args.image_duration - 0.1)) audio_fade = max(0.0, args.audio_fade) audio_fade_in = max(0.0, args.audio_fade_in) video_fade = max(0.0, args.video_fade) intro_logo_alpha = min(1.0, max(0.0, args.intro_logo_alpha)) intro_logo_path = args.intro_logo if args.intro_logo.exists() else DEFAULT_INTRO_LOGO if DEFAULT_INTRO_LOGO.exists() else None if not audio_files: downloaded_audio = download_jamendo_music(args.music_genre, args.input) if downloaded_audio: audio_files.append(downloaded_audio) output = default_output_path(media[0].path, data, args.out_dir) # Image duration means fully visible time, not including crossfades. # Middle images need both a fade-in and fade-out added to their clip length. if len(media) > 1: for idx, item in enumerate(media): if item.kind != "image": continue if idx == 0: item.duration = max(item.duration, (args.image_duration * FIRST_IMAGE_DURATION_MULTIPLIER) + fade) elif idx == len(media) - 1: item.duration = item.duration + fade else: item.duration = item.duration + (2 * fade) console(f"Found {len(media)} media file(s), {len(audio_files)} audio file(s).") console(f"Audio: {audio_files[0] if audio_files else 'none'}") if audio_files: artist, title = audio_metadata(audio_files[0]) if artist or title: console("Audio metadata: " + " - ".join(x for x in [artist, title] if x)) console(f"Output: {output}") console(f"Log: {LOG_PATH}") if data.title or data.date or data.place: console("Title data: " + " | ".join(x for x in [data.title, data.date, data.place] if x)) with tempfile.TemporaryDirectory(prefix="movmaker-") as tmp_name: tmp = Path(tmp_name) jobs = max(1, args.jobs) clips = [tmp / f"clip_{idx:04d}.mp4" for idx in range(len(media))] if jobs == 1: total_clips = len(media) for done, (item, out) in enumerate(zip(media, clips), start=1): percent = done / total_clips * 100 console(f"Preparing clips {done}/{total_clips} ({percent:.1f}%)") log_message(f"Preparing {item.kind}: {item.path.name}") normalize_clip(item, out, width, height, fps, preset, crf) else: console(f"Preparing clips with {jobs} parallel jobs") with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as executor: futures = [] for item, out in zip(media, clips): log_message(f"Preparing {item.kind}: {item.path.name}") futures.append(executor.submit(normalize_clip, item, out, width, height, fps, preset, crf)) total_clips = len(futures) for done, future in enumerate(concurrent.futures.as_completed(futures), start=1): future.result() percent = done / total_clips * 100 console(f"Preparing clips {done}/{total_clips} ({percent:.1f}%)") # Use probed normalized duration for accuracy. durations = [ffprobe_duration(out) for out in clips] captions = [data.captions.get(item.path.name, data.captions.get(str(item.path), "")) for item in media] video_audio = data.video_audio or set() clip_audio_paths = [] for item in media: use_clip_audio = item.kind == "video" and (item.path.name in video_audio or str(item.path) in video_audio) if use_clip_audio and media_has_audio(item.path): clip_audio_paths.append(item.path) console(f"Using video file audio: {item.path.name}") elif use_clip_audio: clip_audio_paths.append(None) console(f"Video file has no audio stream: {item.path.name}") else: clip_audio_paths.append(None) clip_stamps = prepare_clip_stamps(media, data) audio = concat_or_mix_audio(audio_files, tmp / "music.m4a") if len(audio_files) > 1: console(f"Audio: {audio if audio else 'none'}") if audio: artist, title = audio_metadata(audio) if artist or title: console("Audio metadata: " + " - ".join(x for x in [artist, title] if x)) video_intro_duration = audio_fade_in intro_visible_lead = 0.0 intro_clip_path: Path | None = None intro_clip_duration = 0.0 if intro_logo_path is not None and audio_fade_in > 0: if first_media_resolution is None: console("Warning: Could not determine first media resolution; skipping intro logo frame") else: intro_width, intro_height = first_media_resolution intro_image = tmp / "clip_intro_logo.png" intro_clip = tmp / "clip_intro_logo.mp4" console("Preparing intro logo frame") render_intro_logo_frame(intro_logo_path, intro_image, intro_width, intro_height, intro_logo_alpha) intro_item = MediaItem(path=intro_image, kind="image", duration=audio_fade_in + INTRO_LOGO_EXTRA_HOLD) normalize_clip(intro_item, intro_clip, width, height, fps, preset, crf) clips = [intro_clip, *clips] durations = [intro_item.duration, *durations] intro_clip_path = intro_clip intro_clip_duration = intro_item.duration captions = ["", *captions] clip_audio_paths = [None, *clip_audio_paths] clip_stamps = ["", *clip_stamps] intro_visible_lead = min(intro_item.duration, INTRO_LOGO_FADE_OFFSET) video_intro_duration = 0.0 outro_added = False if intro_clip_path is not None: clips.append(intro_clip_path) durations.append(intro_clip_duration) captions.append("") clip_audio_paths.append(None) clip_stamps.append("") outro_added = True build_video(clips, durations, audio, clip_audio_paths, output, data, first_media_date, captions, clip_stamps, fade, audio_fade, audio_fade_in, video_fade, args.intro_logo, intro_logo_alpha, width, height, fps, preset, crf, tmp, video_intro_duration, intro_visible_lead=intro_visible_lead, skip_video_fade_out=outro_added) console(f"Done: {output}") return 0 if __name__ == "__main__": raise SystemExit(main())