movmaker/movmaker.py
2026-05-25 00:56:49 +00:00

658 lines
24 KiB
Python
Executable file

#!/usr/bin/env python3
"""Simple flat-folder photo/video movie maker using ffmpeg.
Usage:
python movmaker.py input --output movie.mp4
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 re
import shlex
import subprocess
import sys
import tempfile
import unicodedata
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"
def tool_path(local_path: Path, fallback: str) -> str:
return str(local_path) if local_path.exists() else fallback
@dataclass
class MovieData:
title: str = ""
date: str = ""
place: str = ""
captions: dict[str, str] | None = None
@dataclass
class MediaItem:
path: Path
kind: str # image or video
duration: float
def run(cmd: list[str]) -> None:
print("+", shlex.join(cmd))
subprocess.run(cmd, check=True)
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_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 exif_image_date(path: Path) -> datetime | None:
"""Read common JPEG/TIFF EXIF date tags using only the stdlib."""
data = path.read_bytes()[:1024 * 1024]
if data.startswith(b"\xff\xd8"):
pos = 2
tiff: bytes | None = None
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"):
tiff = segment[6:]
break
if tiff is None:
return None
else:
tiff = data
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 media_date(path: Path) -> datetime:
ext = path.suffix.lower()
dt = None
if ext in IMAGE_EXTS:
try:
dt = exif_image_date(path)
except Exception:
dt = None
if dt is None:
dt = ffprobe_creation_date(path)
if dt is None:
dt = datetime.fromtimestamp(path.stat().st_mtime)
return dt
def ascii_safe(value: str) -> str:
replacements = {
"æ": "ae", "Æ": "Ae",
"ø": "o", "Ø": "O",
"å": "a", "Å": "A",
"đ": "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:
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')} {ordinal_day(dt.day)} {dt.year}"
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]:
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}")
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={})
lines = [line.strip() for line in path.read_text(encoding="utf-8", errors="replace").splitlines()]
lines = [line for line in lines if line and not line.startswith("#")]
data = MovieData(captions={})
plain: list[str] = []
for line in lines:
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
low = key.lower()
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 Path(key).suffix.lower() in IMAGE_EXTS | VIDEO_EXTS:
data.captions[key] = value
else:
plain.append(line)
else:
plain.append(line)
# Simple mode: first three non-key lines are title/location/date.
if plain:
data.title = data.title or plain[0]
if len(plain) > 1:
data.place = data.place or plain[1]
if len(plain) > 2:
data.date = data.date or plain[2]
return data
def drawtext_escape(text: str) -> str:
# Escape for ffmpeg drawtext text=... without surrounding quotes.
return (
text.replace("\\", "\\\\")
.replace(":", "\\:")
.replace(",", "\\,")
.replace("%", "\\%")
)
def normalize_clip(item: MediaItem, out: Path, width: int, height: int, fps: int, preset: str, crf: int) -> None:
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 concat_or_mix_audio(audio_files: list[Path], out: Path) -> Path | None:
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,
output: Path,
data: MovieData,
creation_date: datetime,
captions: list[str],
fade: float,
audio_fade: float,
audio_fade_in: float,
video_fade: float,
width: int,
height: int,
fps: int,
preset: str,
crf: int,
text_dir: Path,
) -> None:
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)]
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
fade_label = current
if video_fade > 0:
fade_out_start = max(0.0, total_duration - video_fade)
fade_label = "video_faded"
filters.append(
f"[{current}]fade=t=in:st=0:d={video_fade:.3f},fade=t=out:st={fade_out_start:.3f}:d={video_fade:.3f}[{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 = str(Path.home() / ".local/share/fonts/google/Inter%5Bopsz,wght%5D.ttf")
stamp_bold_font = str(Path.home() / ".local/share/fonts/google/Inter%5Bopsz,wght%5D.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)
title_alpha = "if(lt(t\\,5)\\,1\\,if(lt(t\\,6)\\,6-t\\,0))"
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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[{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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[{title_label}]"
)
else:
filters.append(f"[{current}]copy[{title_label}]")
caption_label = title_label
caption_size = max(16, int((subtitle_font_size if title_lines else max(40, height // 18)) * 0.5))
for idx, caption in enumerate(captions):
if not caption:
continue
if n == 1:
start = 0.0
end = total_duration
elif idx == 0:
start = 0.0
end = transition_offsets[0]
elif idx == n - 1:
start = transition_offsets[idx - 1] + fade
end = total_duration
else:
start = transition_offsets[idx - 1] + fade
end = 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 = [line.strip() for line in caption.split("|") if line.strip()]
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='{stamp_bold_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2:y={y_expr}:"
f"fontsize={caption_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=2:shadowy=2:"
f"alpha='{caption_alpha}':enable='between(t,{start:.3f},{end:.3f})'[{next_label}]"
)
bold_label = f"caption{idx}_{line_idx}b"
filters.append(
f"[{next_label}]drawtext=fontfile='{stamp_bold_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2+1:y={y_expr}:"
f"fontsize={caption_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=2:shadowy=2:"
f"alpha='{caption_alpha}':enable='between(t,{start:.3f},{end:.3f})'[{bold_label}]"
)
caption_label = bold_label
stamp_size = max(14, height // 65)
stamp_lines = [x for x in [data.title, data.date, data.place] if x]
stamp_text = drawtext_escape(" | ".join(stamp_lines))
stamp_left_label = "stamp_left"
if stamp_text:
filters.append(
f"[{caption_label}]drawtext=fontfile='{stamp_font}':{textfile_arg(' | '.join(stamp_lines))}:x=24:y=h-text_h-18:"
f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black[{stamp_left_label}]"
)
else:
filters.append(f"[{caption_label}]copy[{stamp_left_label}]")
filters.append(
f"[{stamp_left_label}]drawtext=fontfile='{stamp_font}':{textfile_arg('@bubulescu')}:x=w-text_w-24:y=h-text_h-18:"
f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black[{final_label}]"
)
cmd += ["-filter_complex", ";".join(filters), "-map", f"[{final_label}]"]
metadata = {
"title": data.title,
"date": data.date,
"location": data.place,
"artist": "@bubulescu",
"comment": "Created with movmaker",
"creation_time": creation_date.isoformat(),
}
for key, value in metadata.items():
if value:
cmd += ["-metadata", f"{key}={value}"]
if audio:
audio_index = len(clips)
audio_fade_out_start = max(0.0, total_duration - audio_fade)
audio_filter = f"afade=t=in:st=0:d={audio_fade_in:.3f},afade=t=out:st={audio_fade_out_start:.3f}:d={audio_fade:.3f}"
cmd += ["-map", f"{audio_index}:a:0", "-af", audio_filter, "-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", str(output)]
run(cmd)
def main(argv: list[str] | None = None) -> int:
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("--output", "-o", type=Path, help="output mp4 path; defaults to YYYYMMDD_title.mp4 in the current 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=6.0, help="seconds each image is visible")
parser.add_argument("--fade", type=float, default=3.0, help="crossfade duration in seconds")
parser.add_argument("--audio-fade", type=float, default=10.0, help="music fade-out duration in seconds")
parser.add_argument("--audio-fade-in", type=float, default=2.0, help="music fade-in duration in seconds")
parser.add_argument("--video-fade", type=float, default=1.5, help="video fade-in/fade-out duration in seconds")
parser.add_argument("--first-transition-at", type=float, default=12.0, help="seconds before first transition starts")
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")
args = parser.parse_args(argv)
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 = 24
preset = "ultrafast"
crf = 28
media, audio_files, data = scan_input(args.input, args.image_duration)
first_media_date = media_date(media[0].path)
if not data.date:
data.date = display_date(first_media_date)
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)
output = args.output or 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.first_transition_at + fade)
elif idx == len(media) - 1:
item.duration = item.duration + fade
else:
item.duration = item.duration + (2 * fade)
print(f"Found {len(media)} media file(s), {len(audio_files)} audio file(s).")
print(f"Output: {output}")
if data.title or data.date or data.place:
print("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:
for item, out in zip(media, clips):
print(f"Preparing {item.kind}: {item.path.name}")
normalize_clip(item, out, width, height, fps, preset, crf)
else:
print(f"Preparing clips with {jobs} parallel jobs")
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as executor:
futures = []
for item, out in zip(media, clips):
print(f"Preparing {item.kind}: {item.path.name}")
futures.append(executor.submit(normalize_clip, item, out, width, height, fps, preset, crf))
for future in concurrent.futures.as_completed(futures):
future.result()
# 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]
audio = concat_or_mix_audio(audio_files, tmp / "music.m4a")
build_video(clips, durations, audio, output, data, first_media_date, captions, fade, audio_fade, audio_fade_in, video_fade, width, height, fps, preset, crf, tmp)
print(f"Done: {output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())