feat(scripts): unified Python toolchain, stdlib only

Replaces the scattered TypeScript/shell helpers with one Python toolchain that
runs as plain files (`python3 scripts/tools.py <tool> ...`) with no third-party
dependencies.

New tools
- `_common.py`     shared plumbing (Report, run, ffmpeg discovery, argparse ergonomics)
- `hwprobe.py`     what this machine can actually encode (software + hardware)
- `srt2ass.py`     SRT -> ASS, the missing step between lrc2srt and baking
- `ffmpeg_tools.py` mux / bake / still / strip / probe, with hardware encode
- `bom_gen.py`     BOM generator, ported from scripts/bom-gen.ts
- `tools.py`       dispatcher so any tool is one command away
- `README.md`      usage for all of the above

Rewritten
- `lrc2srt.py` now handles multiple timestamps per line, timestamp-less cue lines
  and blank/comment lines, and survives malformed input. The legacy version
  emitted an empty cue plus a stray cue for every line.

Hardware encoding
This host is AMD (Strix Halo, VCN 4.0.5), so nvenc in the legacy notes is
reference only. `hwprobe.py` probes and reports what genuinely works here:
VA-API via /dev/dri/renderD128 gives h264_vaapi, hevc_vaapi and av1_vaapi.
`ffmpeg_tools.py --video auto` picks the hardware encoder when the render node
is usable and falls back to libx264/libx265 when it is not.

Two bugs found and fixed while wiring this up:
- the VA-API capability probe fed frames straight to the encoder without
  `format=nv12,hwupload`, so ffmpeg failed with "Function not implemented" and
  the toolchain silently fell back to software on a machine where hardware
  works. VA-API encoders only accept GPU-resident frames; the probe now uploads.
- `bake --video copy` only failed when the copied stream happened to be VA-API.
  Burning subtitles always changes pixels, so it now refuses `copy` outright
  with a clear message.

Also makes -q/-n usable on either side of the mode word, and ignores
__pycache__ in .gitignore.
This commit is contained in:
2026-09-10 17:47:02 +09:00
parent 51b878486f
commit 68c12caef9
9 changed files with 2941 additions and 53 deletions
+4
View File
@@ -36,3 +36,7 @@ node_modules
dist dist
out out
.env .env
# Python (helper scripts)
__pycache__/
*.py[cod]
.venv/
+204
View File
@@ -0,0 +1,204 @@
# NCMR songs-videos — tools
Helper scripts for building NCMR song videos: converting lyrics to subtitles,
burning them in, muxing audio, and working out what this machine can encode.
Everything here is **Python standard library only**. No virtualenv, no install
step, no dependencies to break in six months. Run them straight from the repo:
```
python3 scripts/tools.py --help
```
---
## Quick start
```sh
cd ~/dev/mikkeli/ncmr-songs-videos
# 1. What can this machine actually encode?
python3 scripts/tools.py hwprobe
# 2. Lyrics -> subtitles
python3 scripts/tools.py lrc2srt song.lrc -o song.srt
python3 scripts/tools.py srt2ass song.srt -o song.ass
# 3. Audio + still image -> a video (the "PV" case)
python3 scripts/tools.py ffmpeg still cover.png song.flac -o pv.mkv
# 4. Burn the subtitles in, copy the audio through untouched
python3 scripts/tools.py ffmpeg bake pv.mkv song.ass -o final.mp4 --format flac
# 5. Record what untracked assets went in
python3 scripts/tools.py bom --dir .
```
Every subcommand takes `--help`. Every subcommand accepts `--dry-run` (`-n`) to
print the work it would do without doing it, which is the fastest way to see
what ffmpeg command you are about to run.
---
## The tools
| Tool | Purpose |
| --- | --- |
| `tools.py` | Dispatcher. `python3 scripts/tools.py <tool> [args...]`. |
| `hwprobe.py` | Report which encoders this machine can really use, and why not. |
| `lrc2srt.py` | LRC lyrics → SRT subtitles. |
| `srt2ass.py` | SRT → ASS, for burn-in via ffmpeg's `ass` filter. |
| `ffmpeg_tools.py` | `mux`, `bake`, `still`, `strip`, `probe` — the ffmpeg wrappers. |
| `bom_gen.py` | Write a `BOM.yaml` (or `BOM.md`) of a song's untracked assets. |
| `_common.py` | Shared helpers. Not a command; imported by the rest. |
Each script also runs standalone (`python3 scripts/hwprobe.py`) and can be
imported as a module. The dispatcher is a convenience, not a requirement.
---
## Hardware encoding
This machine is an **AMD Strix Halo** (Radeon 8060S). There is no NVIDIA card,
so the `nvenc` recipes in `ffmpegusage.txt` are reference only — they will not
run here. What does work is **VA-API** via the Mesa `radeonsi` driver:
| Codec | Encoder | Status |
| --- | --- | --- |
| H.264 | `h264_vaapi` | works |
| HEVC | `hevc_vaapi` | works |
| AV1 | `av1_vaapi` | works |
| H.264/HEVC | `libx264`/`libx265` | software fallback, always available |
`ffmpeg_tools.py` picks a hardware encoder automatically when one is usable and
falls back to `libx264` when it is not, so the same command works on a machine
without a usable GPU. Override it when you want to:
```sh
--video auto # default: hardware if possible, else software
--video x264 # force software H.264
--video x265 # force software HEVC
--video copy # no re-encode at all
--codec hevc # which hardware codec family to try with --video auto
--device PATH # VA-API render node (default /dev/dri/renderD128)
```
### If hardware encoding stops working
The usual cause is permissions on the DRM render node. `/dev/dri/renderD128` is
owned by `root:render`, so the user needs to be in the `render` group:
```sh
getent group render # who is in it
python3 scripts/tools.py hwprobe # says plainly whether the device opens
```
Group membership only applies to *new* logins. If `id` shows `render` but
`hwprobe` still reports the device as unreadable, the running process is holding
stale credentials — log out and back in, or start the process under the group
explicitly:
```sh
sg render -c 'python3 scripts/hwprobe.py'
```
`hwprobe` never guesses. It opens the device, runs a two-frame test encode
through each candidate encoder, and reports what actually worked.
```sh
python3 scripts/tools.py hwprobe --shell # export-friendly KEY=value output
```
---
## Subtitles
LRC is the timestamped-lyrics format lyric sites emit; SRT is what most players
read; ASS is what you need to burn styled text into a video with ffmpeg.
```sh
python3 scripts/tools.py lrc2srt song.lrc -o song.srt --shift -0.5
```
`lrc2srt` handles the messy real-world cases: `[mm:ss]`, `[mm:ss.xx]` and
`[mm:ss.xxx]` timestamps, several timestamps on one line (repeated choruses —
each becomes its own cue), `[ar:]`/`[ti:]` metadata tags, and blank lines inside
a cue. The final cue is held for a few seconds rather than dropped; adjust with
`--tail`. Use `--shift` to nudge everything earlier or later if the lyrics drift
against the audio.
```sh
python3 scripts/tools.py srt2ass song.srt -o song.ass --play-res 1920x1080
```
`srt2ass` converts to ASS and gives ffmpeg something it can burn in. `--play-res`
sets the script resolution used for font scaling; match it to your video so text
sizes come out as designed. `--dry-run` prints the ASS to stdout for inspection.
---
## ffmpeg wrappers
Long ffmpeg command lines are easy to get subtly wrong and impossible to
remember. These wrap the five the project actually needs.
```sh
# Attach audio to a video-only file; video stream copied, never re-encoded.
python3 scripts/tools.py ffmpeg mux render.mp4 song.flac -o pv.mkv --format flac
# Burn subtitles into a video.
python3 scripts/tools.py ffmpeg bake pv.mkv song.ass -o final.mp4 --format flac
# One still image + audio -> a video.
python3 scripts/tools.py ffmpeg still cover.png song.flac -o pv.mkv --fps 60
# Pull the audio back out as MP3.
python3 scripts/tools.py ffmpeg strip final.mp4 -o final.mp3
# What is actually in this file?
python3 scripts/tools.py ffmpeg probe final.mp4
```
Notes on the choices baked in:
- **`mux` uses `-c:v copy`.** The Remotion render is already encoded; re-encoding
it to add audio costs quality and time for nothing. Containers are picked from
the output extension.
- **`--format` presets** are `aac`, `mp3`, `flac`, `opus` and `copy`. `flac` is
the default because these are archival masters; use `aac`/`opus` for anything
you are actually shipping. `--bitrate` applies to the lossy ones.
- **`-shortest`** is set where a still image is looped, so the video ends with
the audio instead of running on for ever.
- **`-movflags +faststart`** is set for MP4 output, so the file starts playing
before it has fully downloaded.
For anything these do not cover, use `--dry-run` to print the generated command,
copy it, and edit it — the wrapper is a convenience, not a cage. The legacy
recipes this grew out of are preserved in `ffmpegusage.txt`.
---
## Bill of materials
Binary assets are git-ignored, so a clone does not tell you what went into a
song. `bom_gen.py` walks the asset directories and records it:
```sh
python3 scripts/tools.py bom --dir . --format yaml
```
It writes `BOM.yaml` listing every untracked file under `audio/`, `images/`,
`fonts/` and `videos/` with its size, plus whatever description you give it. Use
`--format md` for Markdown, `--format json` for machine-readable output, and
`-y` to skip the interactive description prompt. If PyYAML is available it is
used for clean output; otherwise a small built-in writer produces the same
schema. Neither is required.
---
## Conventions
- `master` is the protected template. Do per-song work on `<song>/master`, and
features on `<song>/<feature>`. See `README.md`.
- Binary assets stay out of git; the BOM is how you know what was used.
- Keep new tools in `scripts/`, stdlib only, with a `--dry-run` and `--help`.
+340
View File
@@ -0,0 +1,340 @@
"""NCMR songs-videos helper toolchain — Python, stdlib only.
Shared plumbing for the repo's helper scripts. Every script in this package is
runnable as a plain file (``python3 scripts/bom_gen.py``) and as a module
(``python3 -m scripts.bom_gen``); both work because this package is imported
either way without needing its parent directory on ``sys.path``.
Nothing outside the standard library is imported anywhere in this package.
"""
import argparse
import os
import re
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from typing import Callable, Iterable, Sequence
__all__ = [
"Die",
"VERSION",
"Report",
"find_ffmpeg",
"glob_files",
"human_size",
"report",
"run",
"subparsers_metavar",
"text_table",
]
VERSION = "0.2.0"
class Die(Exception):
"""Fatal, user-facing error. Caught in ``main()`` and printed to stderr."""
# --------------------------------------------------------------------------- #
# Console output
# --------------------------------------------------------------------------- #
def _colour_enabled(stream=None) -> bool:
stream = stream or sys.stdout
if os.environ.get("NO_COLOR"):
return False
if os.environ.get("FORCE_COLOR"):
return True
return hasattr(stream, "isatty") and stream.isatty()
def _style(text: str, code: str) -> str:
if not _colour_enabled():
return text
return f"\033[{code}m{text}\033[0m"
@dataclass
class Report:
"""Prints user-facing output, with optional ANSI colour when on a TTY."""
quiet: bool = False
def _emit(self, message: str, *, err: bool = False) -> None:
stream = sys.stderr if err else sys.stdout
print(message, file=stream)
def info(self, message: str = "") -> None:
"""Ordinary progress output."""
self._emit(message)
def ok(self, message: str) -> None:
self._emit(_style(f"✓ {message}", "32"))
def warn(self, message: str) -> None:
self._emit(_style(f"warning: {message}", "33"), err=True)
def error(self, message: str) -> None:
self._emit(_style(f"error: {message}", "31"), err=True)
def verbose(self, message: str) -> None:
if not self.quiet:
self._emit(_style(f" {message}", "2"))
#: Default reporter used when a caller doesn't supply one.
report = Report()
# --------------------------------------------------------------------------- #
# Formatting helpers
# --------------------------------------------------------------------------- #
_SIZE_UNITS = ("B", "KiB", "MiB", "GiB", "TiB")
# Characters that make a key unquoted-safe in YAML and legible in a table.
_SAFE_KEY_RE = re.compile(r"^[A-Za-z0-9_.\-/]+$")
def human_size(num_bytes: int) -> str:
"""Format a byte count as ``1.5 MiB`` / ``12 B``."""
value = float(num_bytes)
for unit in _SIZE_UNITS[:-1]:
if value < 1024:
break
value /= 1024
else:
unit = _SIZE_UNITS[-1]
if unit == "B":
return f"{int(value)} B"
return f"{value:.1f} {unit}"
def text_table(headers: Sequence[str], rows: Iterable[Sequence[str]]) -> str:
"""Render a simple whitespace-aligned table (no boxes, pipe-friendly)."""
body = [[str(cell) for cell in row] for row in rows]
widths = [len(str(h)) for h in headers]
for row in body:
for i, cell in enumerate(row):
if i < len(widths):
widths[i] = max(widths[i], len(cell))
def line(cells: Sequence[str]) -> str:
return " ".join(cell.ljust(widths[i]) for i, cell in enumerate(cells)).rstrip()
out = [line([str(h) for h in headers])]
out.append(" ".join("-" * w for w in widths).rstrip())
for row in body:
out.append(line(row))
return "\n".join(out)
def glob_files(directory: str, suffix: str) -> list[str]:
"""Sorted list of files directly inside *directory* whose name ends in *suffix*.
Returns an empty list when the directory does not exist, so callers can treat
"no such folder" and "folder is empty" identically.
"""
try:
entries = os.listdir(directory)
except (FileNotFoundError, NotADirectoryError):
return []
suffix = suffix.lower()
found = [
os.path.join(directory, name)
for name in entries
if name.lower().endswith(suffix) and os.path.isfile(os.path.join(directory, name))
]
return sorted(found)
# --------------------------------------------------------------------------- #
# Subprocess plumbing
# --------------------------------------------------------------------------- #
def run(
cmd: Sequence[str],
*,
report: Report | None = None,
check: bool = True,
cwd: str | None = None,
dry_run: bool = False,
) -> subprocess.CompletedProcess:
"""Run *cmd* with a shell-free argument list, echoing it first.
``report`` is the reporter used for the echo; the helper deliberately takes a
reporter argument rather than closing over the module-level default so that a
``--quiet`` CLI cannot accidentally echo commands.
"""
log = report or globals()["report"]
printable = " ".join(shlex_quote(part) for part in cmd)
if dry_run:
log.info(_style(f"[dry-run] {printable}", "36"))
return subprocess.CompletedProcess(list(cmd), 0)
log.info(_style(f"$ {printable}", "2"))
# ffmpeg prints its whole build configuration as a banner unless told not to.
# Keep it quiet by default, and keep stderr so a failure can be explained.
full_cmd = list(cmd)
if os.path.basename(full_cmd[0]) in ("ffmpeg", "ffprobe"):
full_cmd[1:1] = ["-hide_banner"]
try:
result = subprocess.run(full_cmd, cwd=cwd, capture_output=True, text=True)
except FileNotFoundError as exc:
raise Die(f"command not found: {cmd[0]}") from exc
if result.stdout:
sys.stdout.write(result.stdout)
if result.stderr:
# ffmpeg writes progress to stderr; pass it through so the user still sees
# the encode happen, but drop the wall of banner text.
sys.stderr.write(result.stderr)
if check and result.returncode != 0:
detail = _last_meaningful_line(result.stderr)
suffix = f": {detail}" if detail else ""
raise Die(f"{os.path.basename(cmd[0])} failed (exit {result.returncode}){suffix}")
return result
def _last_meaningful_line(text: str | None) -> str:
"""Pick the most useful line out of ffmpeg's stderr to show in an error."""
if not text:
return ""
for line in reversed(text.strip().splitlines()):
line = line.strip()
if not line:
continue
# Progress lines start with 'frame=' and overwrite each other.
if line.startswith(("frame=", "size=")) or "speed=" in line:
continue
return line
return ""
def shlex_quote(part: str) -> str:
"""Quote one argv element for display. Local copy keeps the module stdlib-only."""
if not part:
return "''"
if re.fullmatch(r"[A-Za-z0-9_@%+=:,./\-]+", part):
return part
return "'" + part.replace("'", "'\"'\"'") + "'"
def find_ffmpeg(override: str | None = None) -> str:
"""Locate the ``ffmpeg`` binary, or raise :class:`Die` with a useful message."""
if override:
if os.path.sep in override:
if not os.access(override, os.X_OK):
raise Die(f"--ffmpeg {override} is not an executable file")
return override
return override # a bare name; let the OS resolve it on $PATH
found = shutil.which("ffmpeg")
if not found:
raise Die("ffmpeg not found on $PATH — install it or pass --ffmpeg /path/to/ffmpeg")
return found
# --------------------------------------------------------------------------- #
# argparse ergonomics
# --------------------------------------------------------------------------- #
def subparsers_metavar(parser: argparse.ArgumentParser, name: str = "mode") -> None:
"""Stop argparse from printing an unreadable ``{a,b,c}`` in help output.
Call *after* every subparser has been registered.
"""
for action in parser._actions:
if isinstance(action, argparse._SubParsersAction):
action.metavar = name
action.choices = None # type: ignore[assignment] # keep choices working
def add_common_arguments(parser: argparse.ArgumentParser, *, suppress_defaults: bool = False) -> None:
"""Switches every tool in the toolbox understands.
``suppress_defaults`` is used when the same switches are also registered on
subparsers: it stops the subparser copy from stamping its default over a value
the user gave before the mode word.
"""
default_suppress = argparse.SUPPRESS if suppress_defaults else False
parser.add_argument(
"-q",
"--quiet",
action="store_true",
default=default_suppress,
help="suppress progress output (errors still go to stderr)",
)
parser.add_argument(
"-n",
"--dry-run",
action="store_true",
default=default_suppress,
help="print the work that would happen without doing it",
)
parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {VERSION}",
)
def add_ffmpeg_argument(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--ffmpeg",
metavar="PATH",
default=None,
help="ffmpeg binary to use (default: whichever is on $PATH)",
)
@dataclass
class ReportMixin:
"""Mixin for subcommand classes that need a reporter and a dry-run flag."""
report: Report = field(default_factory=Report)
dry_run: bool = False
def main_wrapper(fn: Callable[[], int | None]) -> Callable[[], int]:
"""Decorator turning ``Die`` exceptions into a clean exit status."""
def wrapper() -> int:
try:
return fn() or 0
except Die as exc:
report.error(str(exc))
return 1
except KeyboardInterrupt:
report.error("interrupted")
return 130
return wrapper
+467
View File
@@ -0,0 +1,467 @@
"""Generate a bill of materials for a song's untracked binary assets.
The repo's guideline says every song keeps a ``BOM.md`` (or ``BOM.yaml``)
listing the untracked input assets and explaining what each file is. This walks
the conventional asset directories, prompts for a description per file, and
writes the result.
Stdlib only: the YAML is emitted by hand. If PyYAML happens to be installed it is
used to *validate* the output, but it is never required.
Usage::
python3 scripts/bom_gen.py # walk ./assets, write BOM.yaml
python3 scripts/bom_gen.py path/to/song # walk a song directory
python3 scripts/bom_gen.py --format md # write BOM.md instead
python3 scripts/bom_gen.py --descriptions d.toml # non-interactive
python3 scripts/bom_gen.py --describe-one audio/song.flac
"""
from __future__ import annotations
import argparse
import os
import re
import sys
from dataclasses import dataclass, field
if __package__ in (None, ""):
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _common import ( # type: ignore[import-not-found]
Die,
Report,
add_common_arguments,
human_size,
main_wrapper,
text_table,
)
else: # pragma: no cover
from ._common import (
Die,
Report,
add_common_arguments,
human_size,
main_wrapper,
text_table,
)
#: Directories that hold untracked input assets, with the role we assume.
ASSET_DIRS: tuple[tuple[str, str], ...] = (
("audio", "Song audio: final mix, instrumental, stems"),
("images", "Cover art, stills, background plates"),
("fonts", "Typefaces used by the subtitles or overlays"),
("videos", "Material video clips"),
)
#: File extensions that are worth flagging as (probably) not-yet-described.
_DESCRIPTION_PLACEHOLDER = "<describe this file>"
_BOM_MARKER = "; generated by scripts/bom_gen.py"
# --------------------------------------------------------------------------- #
# YAML emission (hand-rolled, stdlib only)
# --------------------------------------------------------------------------- #
#: Characters that force a value to be quoted in YAML.
_NEEDS_QUOTES = re.compile(r"""[:\{\}\[\],&*#?|<>=!%@`"']|^\s|\s$|^$|^[-?]""")
def yaml_scalar(value: str) -> str:
"""Quote *value* for YAML only when it needs it."""
if value.isdigit():
return f'"{value}"' # keep hashes/filenames that look numeric as strings
if _NEEDS_QUOTES.search(value):
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
return value
def yaml_key(value: str) -> str:
"""Quote a key that would otherwise be ambiguous (spaces, colons, leading -)."""
if _NEEDS_QUOTES.search(value) or value.startswith("-") or value != value.strip():
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
return value
# --------------------------------------------------------------------------- #
# Asset discovery
# --------------------------------------------------------------------------- #
@dataclass
class Asset:
path: str # relative to the BOM root, POSIX separators
size: int
description: str = ""
extra: dict[str, str] = field(default_factory=dict)
@property
def directory(self) -> str:
return os.path.dirname(self.path).split("/")[0] or "."
def walk_assets(root: str, dirs: tuple[tuple[str, str], ...] = ASSET_DIRS) -> list[Asset]:
"""Collect files under each asset directory, sorted by path."""
assets: list[Asset] = []
for name, _role in dirs:
base = os.path.join(root, name)
if not os.path.isdir(base):
continue
for dirpath, dirnames, filenames in os.walk(base):
dirnames.sort()
for filename in sorted(filenames):
if filename.startswith("."):
continue
full = os.path.join(dirpath, filename)
if not os.path.isfile(full):
continue
rel = os.path.relpath(full, root).replace(os.sep, "/")
assets.append(Asset(path=rel, size=os.path.getsize(full)))
return assets
def load_descriptions(path: str) -> dict[str, str]:
"""Read a descriptions file: simple ``path = description`` or ``path<tab>desc``.
A deliberately forgiving format so descriptions can also be pasted in from a
spreadsheet. ``#`` starts a comment.
"""
try:
with open(path, "r", encoding="utf-8") as handle:
raw = handle.read()
except OSError as exc:
raise Die(f"cannot read --descriptions {path}: {exc.strerror}") from exc
mapping: dict[str, str] = {}
for line in raw.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, value = line.partition("=")
elif "\t" in line:
key, _, value = line.partition("\t")
else:
continue
mapping[key.strip().lstrip("./")] = value.strip()
return mapping
# --------------------------------------------------------------------------- #
# Rendering
# --------------------------------------------------------------------------- #
def render_yaml(root: str, assets: list[Asset], *, title: str | None, by_dir: bool) -> str:
total = sum(a.size for a in assets)
lines = [
"---",
_BOM_MARKER,
f"root: {yaml_scalar(os.path.basename(os.path.abspath(root)) or root)}",
]
if title:
lines.append(f"title: {yaml_scalar(title)}")
lines.append(f"total_files: {len(assets)}")
lines.append(f"total_size: {total}")
lines.append("")
if by_dir:
for dirname, role in ASSET_DIRS:
grouping = [a for a in assets if a.directory == dirname]
if not grouping:
continue
subtotal = sum(a.size for a in grouping)
lines.append(f"{yaml_key(dirname)}:")
lines.append(f" role: {yaml_scalar(role)}")
lines.append(f" count: {len(grouping)}")
lines.append(f" size: {subtotal}")
lines.append(" files:")
for asset in grouping:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
lines.append(f" description: {yaml_scalar(asset.description)}")
lines.append("")
else:
lines.append("files:")
for asset in assets:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
lines.append(f" description: {yaml_scalar(asset.description)}")
# Anything under an unrecognised top-level directory still deserves a mention.
others = [a for a in assets if a.directory not in {d for d, _ in ASSET_DIRS}]
if others:
lines.append("")
lines.append("unclassified:")
for asset in others:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
lines.append(f" description: {yaml_scalar(asset.description)}")
lines.append("...")
return "\n".join(lines) + "\n"
def render_markdown(root: str, assets: list[Asset], *, title: str | None, by_dir: bool) -> str:
total = sum(a.size for a in assets)
heading = title or os.path.basename(os.path.abspath(root)) or "Untracked assets"
lines = [
f"# Bill of materials — {heading}",
"",
f"{len(assets)} file(s), {human_size(total)}. Generated by `scripts/bom_gen.py`.",
"",
]
groups: list[tuple[str, list[Asset]]] = []
if by_dir:
for dirname, _role in ASSET_DIRS:
grouping = [a for a in assets if a.directory == dirname]
if grouping:
groups.append((dirname, grouping))
others = [a for a in assets if a.directory not in {d for d, _ in ASSET_DIRS}]
if others:
groups.append(("unclassified", others))
else:
groups.append(("files", assets))
for group_name, grouping in groups:
lines.append(f"## {group_name}")
lines.append("")
rows = [
(
f"`{asset.path}`",
human_size(asset.size),
asset.description or f"_{_DESCRIPTION_PLACEHOLDER}_",
)
for asset in grouping
]
lines.append(text_table(("file", "size", "description"), rows))
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def validate_yaml(document: str, report: Report) -> None:
"""If PyYAML is importable, use it to confirm the emitted document parses."""
try:
import yaml # type: ignore[import-not-found]
except ImportError:
return
try:
yaml.safe_load(document)
except Exception as exc: # noqa: BLE001 - report whatever PyYAML says
report.warn(f"PyYAML could not parse the generated document: {exc}")
# --------------------------------------------------------------------------- #
# Interactive prompts
# --------------------------------------------------------------------------- #
def prompt_for_descriptions(assets: list[Asset], report: Report) -> None:
"""Ask for a description per asset. EOF ends prompting and keeps what we have."""
if not sys.stdin.isatty():
report.warn(
"stdin is not a terminal — skipping prompts; "
"descriptions left blank (use --descriptions FILE to supply them)"
)
return
total = len(assets)
report.info(f"Describing {total} file(s). Blank line accepts the current value.\n")
report.info("Press Ctrl-C or Ctrl-D to stop early; the BOM is still written.\n")
try:
for i, asset in enumerate(assets, start=1):
report.info(f"[{i}/{total}] {asset.path} ({human_size(asset.size)})")
try:
answer = input(" description: ").strip()
except EOFError:
report.info("")
report.warn("end of input — remaining descriptions left blank")
return
if answer:
asset.description = answer
except KeyboardInterrupt:
report.info("")
report.warn("interrupted — writing the BOM with the descriptions gathered so far")
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="bom_gen",
description="Generate a bill of materials for a song's untracked binary assets.",
)
add_common_arguments(parser)
parser.add_argument(
"root",
nargs="?",
default=".",
help="directory holding audio/, images/, fonts/, videos/ (default: .)",
)
parser.add_argument(
"-o",
"--output",
help="output path (default: BOM.yaml or BOM.md in the root directory)",
)
parser.add_argument(
"--format",
default="yaml",
choices=("yaml", "md"),
help="output format (default: yaml)",
)
parser.add_argument(
"--descriptions",
metavar="FILE",
help="read 'path = description' lines from FILE instead of prompting",
)
parser.add_argument(
"--no-prompt",
action="store_true",
help="never prompt; write the inventory with blank descriptions",
)
parser.add_argument(
"--flat",
action="store_true",
help="list files without grouping them by asset directory",
)
parser.add_argument(
"--title",
help="song title, recorded in the output header",
)
parser.add_argument(
"--describe-one",
metavar="PATH",
help="update the description of a single file in an existing BOM",
)
return parser
@main_wrapper
def main() -> int:
args = build_parser().parse_args()
report = Report(quiet=args.quiet)
root = os.path.abspath(args.root)
if not os.path.isdir(root):
raise Die(f"not a directory: {args.root}")
if args.describe_one:
return describe_one(root, args, report)
assets = walk_assets(root)
if not assets:
expected = ", ".join(f"{name}/" for name, _ in ASSET_DIRS)
report.warn(f"no assets found under {root} (looked in: {expected})")
return 1
if args.descriptions:
mapping = load_descriptions(args.descriptions)
applied = 0
for asset in assets:
if asset.path in mapping:
asset.description = mapping[asset.path]
applied += 1
missing = [a.path for a in assets if not a.description]
report.verbose(f"applied {applied} description(s) from {args.descriptions}")
if missing:
report.warn(f"{len(missing)} file(s) have no description in {args.descriptions}")
elif not args.no_prompt:
prompt_for_descriptions(assets, report)
renderer = render_markdown if args.format == "md" else render_yaml
document = renderer(root, assets, title=args.title, by_dir=not args.flat)
if args.format == "yaml":
validate_yaml(document, report)
default_name = "BOM.md" if args.format == "md" else "BOM.yaml"
output = args.output or os.path.join(root, default_name)
if args.dry_run:
report.info(document)
report.info(f"[dry-run] would write {output}")
return 0
with open(output, "w", encoding="utf-8", newline="\n") as handle:
handle.write(document)
total = sum(a.size for a in assets)
described = sum(1 for a in assets if a.description)
report.ok(f"{output} — {len(assets)} file(s), {human_size(total)}, {described} described")
if described < len(assets):
report.info(f" {len(assets) - described} file(s) still need a description")
return 0
def describe_one(root: str, args: argparse.Namespace, report: Report) -> int:
"""Set the description for a single path inside an existing BOM."""
target = args.describe_one.lstrip("./")
if not os.path.isfile(os.path.join(root, target)):
raise Die(f"no such asset: {target}")
output = args.output or os.path.join(root, "BOM.yaml" if args.format == "yaml" else "BOM.md")
description = args.title or "" # reused as the description payload for this mode
if not description:
if not sys.stdin.isatty():
raise Die("--describe-one needs a description; pass --title \"...\"")
description = input(f"description for {target}: ").strip()
if not os.path.isfile(output):
raise Die(f"no existing BOM at {output}; run without --describe-one first")
with open(output, "r", encoding="utf-8") as handle:
content = handle.read()
pattern = re.compile(
rf"(^[ \t]*{re.escape(target)}:[ \t]*\n"
rf"(?:[ \t]+\w+:.*\n)*?"
rf"[ \t]+description:[ \t]*)"
rf".*$",
re.MULTILINE,
)
updated, count = pattern.subn(rf"\g<1>{yaml_scalar(description)}", content)
if count == 0:
raise Die(f"could not find an entry for {target} in {output}")
if args.dry_run:
report.info(f"[dry-run] would set {target} description in {output}")
return 0
with open(output, "w", encoding="utf-8", newline="\n") as handle:
handle.write(updated)
report.ok(f"{target} described in {output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+646
View File
@@ -0,0 +1,646 @@
"""FFmpeg front-end for NCMR song videos.
Wraps the ffmpeg invocations this project actually needs, so the long command
lines live in one place instead of in shell history and notes files.
Modes:
``mux`` attach an audio track to a video-only render, without re-encoding
the video (the Remotion output is video-only; this is step 1)
``bake`` burn an ASS subtitle file into a video (step 2)
``still`` build a video from a single image plus audio, for the no-lyrics case
``strip`` pull an MP3 out of a finished video
``probe`` print a compact stream summary
Both ``mux`` and ``bake`` can be chained with ``--then-bake``/``--then-mux`` so a
render goes to finished file in one command.
Usage::
python3 scripts/ffmpeg_tools.py probe out/pulseraw.mp4
python3 scripts/ffmpeg_tools.py mux out/pulseraw.mp4 song.wav -o raw.mkv --format flac
python3 scripts/ffmpeg_tools.py bake raw.mkv subs.ass -o final.mp4
python3 scripts/ffmpeg_tools.py still cover.png song.wav -o pv.mkv --fps 60 --format flac
python3 scripts/ffmpeg_tools.py strip final.mp4 -o song.mp3
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass
if __package__ in (None, ""):
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _common import ( # type: ignore[import-not-found]
Die,
Report,
add_common_arguments,
add_ffmpeg_argument,
find_ffmpeg,
main_wrapper,
run,
text_table,
)
else: # pragma: no cover
from ._common import (
Die,
Report,
add_common_arguments,
add_ffmpeg_argument,
find_ffmpeg,
main_wrapper,
run,
text_table,
)
# --------------------------------------------------------------------------- #
# Audio/codec presets
# --------------------------------------------------------------------------- #
#: ``--format`` choices, mapping to the ffmpeg audio arguments to use.
AUDIO_FORMATS = {
# name: (extension, container args, codec args)
"aac": ("mp4", [], ["-c:a", "aac", "-b:a", "320k"]),
"mp3": ("mp4", [], ["-c:a", "libmp3lame", "-b:a", "320k"]),
"flac": ("mkv", [], ["-c:a", "flac", "-compression_level", "12"]),
"opus": ("mkv", [], ["-c:a", "libopus", "-b:a", "192k"]),
"copy": (None, [], ["-c:a", "copy"]),
}
#: ``--video`` choices. ``auto`` picks a hardware encoder when one works.
VIDEO_ENCODERS = {
"auto": None,
"x264": ["-c:v", "libx264"],
"x265": ["-c:v", "libx265"],
"copy": ["-c:v", "copy"],
}
VAAPI_ENCODERS = {
"h264": ["-c:v", "h264_vaapi"],
"hevc": ["-c:v", "hevc_vaapi"],
"av1": ["-c:v", "av1_vaapi"],
}
#: Software fallback quality knob, applied only for libx264/libx265.
X264_QUALITY = ["-preset", "medium", "-crf", "18"]
DEFAULT_DEVICE = "/dev/dri/renderD128"
@dataclass
class HwContext:
"""Hardware acceleration state: which encoder to use and how to init it."""
encoder: str
device: str | None = None
is_vaapi: bool = False
note: str = ""
def input_args(self) -> list[str]:
"""Args that go before ``-i``."""
if self.is_vaapi and self.device:
return ["-init_hw_device", f"vaapi=va:{self.device}", "-filter_hw_device", "va"]
return []
def filter_suffix(self) -> str:
"""Filters appended to a user filter chain to move frames onto the GPU."""
if self.is_vaapi:
return "format=nv12,hwupload"
return ""
# --------------------------------------------------------------------------- #
# Encoder selection
# --------------------------------------------------------------------------- #
def _vaapi_device_readable(device: str) -> bool:
try:
fd = os.open(device, os.O_RDWR)
except OSError:
return False
os.close(fd)
return True
_DEVICE_CACHE: dict[str, bool] = {}
def _encoder_works(ffmpeg: str, encoder: str, extra: list[str], device_key: str = "") -> bool:
"""Try a two-frame encode. Results are memoised per process.
VA-API encoders only accept GPU-resident frames, so the probe must upload the
test pattern first. Without ``format=nv12,hwupload`` ffmpeg fails with
"Function not implemented" even though the encoder works fine — which would
make this fall back to software on a machine that has no problem at all.
"""
cache_key = f"{encoder}|{device_key}"
if cache_key in _DEVICE_CACHE:
return _DEVICE_CACHE[cache_key]
vf = "format=nv12,hwupload" if encoder.endswith("_vaapi") else None
cmd = [
ffmpeg, "-hide_banner", "-loglevel", "error", *extra,
"-f", "lavfi", "-i", "color=c=black:s=256x256:r=30:d=1",
]
if vf:
cmd += ["-vf", vf]
cmd += ["-c:v", encoder, "-frames:v", "2", "-f", "null", "-"]
try:
ok = subprocess.run(cmd, capture_output=True, timeout=60).returncode == 0
except (OSError, subprocess.TimeoutExpired):
ok = False
_DEVICE_CACHE[cache_key] = ok
return ok
#: Hardware encoders worth trying for each codec family, in preference order.
HW_ENCODER_CHAIN: dict[str, tuple[tuple[str, str], ...]] = {
# codec: ((encoder, init-family), ...)
"h264": (("h264_vaapi", "vaapi"), ("h264_nvenc", "nvenc"), ("h264_qsv", "qsv")),
"hevc": (("hevc_vaapi", "vaapi"), ("hevc_nvenc", "nvenc"), ("hevc_qsv", "qsv")),
"av1": (("av1_vaapi", "vaapi"), ("av1_nvenc", "nvenc"), ("av1_qsv", "qsv")),
}
def pick_acceleration(
ffmpeg: str,
report: Report,
*,
video: str = "auto",
codec: str = "h264",
device: str | None = None,
allow_software: bool = True,
) -> HwContext:
"""Decide how to encode, honouring an explicit ``--video`` choice."""
device = device or os.environ.get("HW_DEVICE") or DEFAULT_DEVICE
if video == "copy":
return HwContext(encoder="copy")
if video == "auto":
chain = HW_ENCODER_CHAIN.get(codec)
if chain is None:
raise Die(f"unknown --codec {codec!r}")
attempted: list[str] = []
for encoder, family in chain:
if family == "vaapi":
if not _vaapi_device_readable(device):
attempted.append(f"{encoder} (unreadable {device})")
continue
init = ["-init_hw_device", f"vaapi=va:{device}", "-filter_hw_device", "va"]
if _encoder_works(ffmpeg, encoder, init, device):
return HwContext(
encoder=encoder,
device=device,
is_vaapi=True,
note="hardware (VA-API)",
)
attempted.append(f"{encoder} (device rejects it)")
else:
init = []
if _encoder_works(ffmpeg, encoder, init, family):
return HwContext(encoder=encoder, note=f"hardware ({family})")
attempted.append(f"{encoder} (unsupported)")
if not allow_software:
raise Die(
"no working hardware encoder found and software fallback is disabled; "
f"tried {', '.join(attempted)}. Diagnose with: python3 scripts/hwprobe.py"
)
report.verbose(f"no hardware encoder for {codec}: {', '.join(attempted)}")
return HwContext(encoder="libx264" if codec == "h264" else "libx265", note="software")
if video in ("x264", "x265"):
return HwContext(encoder=VIDEO_ENCODERS[video][1], note=f"software ({video})")
raise Die(f"unknown --video {video!r}")
def video_args(context: HwContext, *, quality: bool = True) -> list[str]:
"""Encoder arguments for the selected context."""
if context.encoder == "copy":
return ["-c:v", "copy"]
args = ["-c:v", context.encoder]
if context.encoder.startswith("libx") and quality:
args += X264_QUALITY
return args
def pix_fmt_args(context: HwContext) -> list[str]:
# Other pixel formats come from the VA-API upload filter; without a filter
# chain, hardware encoders still need 4:2:0 input.
if context.is_vaapi:
return []
return ["-pix_fmt", "yuv420p"]
def audio_args(fmt: str) -> tuple[list[str], str]:
if fmt not in AUDIO_FORMATS:
raise Die(f"unknown --format {fmt!r}; choose from {', '.join(AUDIO_FORMATS)}")
extension, container, codec = AUDIO_FORMATS[fmt]
return container + codec, extension or ("")
def resolve_output(output: str | None, stem: str, fmt: str, suffix: str) -> str:
"""Work out the output filename, fixing up the extension for the container."""
if output:
return output
extension = AUDIO_FORMATS[fmt][0]
if extension and not stem.lower().endswith(f".{extension}"):
return f"{stem}{suffix}.{extension}"
return f"{stem}{suffix}"
def build_filter(context: HwContext, *filters: str) -> str | None:
"""Join filter fragments, appending the VA-API upload step when needed."""
parts = [f for f in filters if f]
upload = context.filter_suffix()
if upload:
parts.append(upload)
return ",".join(parts) if parts else None
# --------------------------------------------------------------------------- #
# Commands
# --------------------------------------------------------------------------- #
def cmd_mux(args: argparse.Namespace, context: HwContext, report: Report) -> int:
"""Attach audio to a video-only file. Video stream is stream-copied."""
codec_args, _ = audio_args(args.format)
output = resolve_output(args.output, os.path.splitext(args.input_video)[0], args.format, "_mux")
filter_chain = build_filter(context, *args.vf) if context.is_vaapi else (",".join(args.vf) or None)
video = video_args(context, quality=False) if context.is_vaapi else ["-c:v", "copy"]
cmd = [
args.ffmpeg, "-y",
*context.input_args(),
"-i", args.input_video,
"-i", args.audio,
"-map", "0:v:0", "-map", "1:a:0",
*video,
*codec_args,
]
if filter_chain:
cmd += ["-vf", filter_chain]
cmd += ["-shortest", output]
run(cmd, report=report, dry_run=args.dry_run)
if not args.dry_run:
report.ok(f"muxed -> {output}")
return 0
def cmd_bake(args: argparse.Namespace, context: HwContext, report: Report) -> int:
"""Burn ASS subtitles into the video stream."""
if not os.path.isfile(args.subtitles):
raise Die(f"no such subtitle file: {args.subtitles}")
# libass needs the filename escaped for the filtergraph; ':' and '\' and "'"
# all have meaning there.
escaped = (
args.subtitles.replace("\\", "\\\\")
.replace(":", "\\:")
.replace("'", r"\'")
.replace("[", r"\[")
.replace("]", r"\]")
.replace(",", r"\,")
)
# Burning subtitles means changing pixels, so the video stream cannot be copied.
# Fail early with a clear message instead of letting ffmpeg refuse the filter.
if context.encoder == "copy":
raise Die(
"cannot bake subtitles with --video copy: burning them in requires "
"re-encoding. Use --video auto (default) or --video x264."
)
filter_chain = build_filter(context, f"ass={escaped}", *args.vf)
output = args.output or _append_suffix(args.input_video, "_sub")
codec_args, _ = audio_args(args.format) if args.audio_format != "keep" else (["-c:a", "copy"], None)
cmd = [
args.ffmpeg, "-y",
*context.input_args(),
"-i", args.input_video,
"-vf", filter_chain or "null",
*video_args(context),
*pix_fmt_args(context),
*codec_args,
"-shortest" if args.shortest else "-map_metadata", "-1",
output,
]
run(cmd, report=report, dry_run=args.dry_run)
if not args.dry_run:
report.ok(f"subtitles baked -> {output}")
return 0
def cmd_still(args: argparse.Namespace, context: HwContext, report: Report) -> int:
"""Build a video from one image and one audio file (the no-lyrics render)."""
codec_args, _ = audio_args(args.format)
output = resolve_output(args.output, os.path.splitext(args.image)[0], args.format, "_pv")
filters = [f"fps={args.fps}", f"scale={args.width}:{args.height}:force_original_aspect_ratio=decrease", "pad=ceil(iw/2)*2:ceil(ih/2)*2"]
if context.is_vaapi:
filter_chain = build_filter(context, *filters)
cmd = [args.ffmpeg, "-y", "-loop", "1", "-i", args.image, "-i", args.audio, *context.input_args(), "-vf", filter_chain]
else:
cmd = [args.ffmpeg, "-y", "-loop", "1", "-i", args.image, "-i", args.audio, "-vf", ",".join(filters)]
cmd += ["-r", str(args.fps), *video_args(context), *pix_fmt_args(context), *codec_args, "-shortest", output]
run(cmd, report=report, dry_run=args.dry_run)
if not args.dry_run:
report.ok(f"still-image video -> {output}")
return 0
def cmd_strip(args: argparse.Namespace, context: HwContext | None, report: Report) -> int:
"""Extract the audio track as MP3."""
output = args.output or os.path.splitext(args.input_video)[0] + ".mp3"
cmd = [
args.ffmpeg, "-y", "-i", args.input_video, "-vn",
"-ar", str(args.sample_rate), "-ac", str(args.channels),
"-c:a", "libmp3lame", "-b:a", args.bitrate,
output,
]
run(cmd, report=report, dry_run=args.dry_run)
if not args.dry_run:
report.ok(f"audio -> {output}")
return 0
def cmd_probe(args: argparse.Namespace, context: HwContext | None, report: Report) -> int:
"""Print a compact stream summary for a media file."""
if not os.path.isfile(args.media[0]):
raise Die(f"no such file: {args.media[0]}")
cmd = [
"ffprobe" if args.ffmpeg.endswith("ffmpeg") else args.ffmpeg.replace("ffmpeg", "ffprobe"),
"-hide_banner", "-loglevel", "error",
"-show_entries", "stream=index,codec_type,codec_name,width,height,r_frame_rate,channels,sample_rate,bit_rate",
"-show_entries", "format=duration,size,format_name",
"-of", "json",
*args.media,
]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
except OSError as exc:
raise Die(f"could not run ffprobe: {exc}") from exc
if result.returncode != 0:
raise Die(f"ffprobe failed: {(result.stderr or '').strip()}")
data = json.loads(result.stdout or "{}")
rows = []
for stream in data.get("streams", []):
kind = stream.get("codec_type", "?")
if kind == "video":
detail = f"{stream.get('width')}x{stream.get('height')} @ {stream.get('r_frame_rate')}"
elif kind == "audio":
detail = f"{stream.get('channels')}ch @ {stream.get('sample_rate')} Hz"
else:
detail = ""
rows.append((str(stream.get("index", "")), kind, stream.get("codec_name", "?"), detail))
report.info(text_table(("stream", "type", "codec", "detail"), rows))
fmt = data.get("format", {})
duration = fmt.get("duration")
size = fmt.get("size")
if duration:
report.info(f"\ncontainer: {fmt.get('format_name', '?')}")
report.info(f"duration: {float(duration):.2f}s")
if size:
report.info(f"size: {int(size):,} bytes")
return 0
def _append_suffix(path: str, suffix: str) -> str:
stem, extension = os.path.splitext(path)
return f"{stem}{suffix}{extension}"
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def _add_video_options(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--video",
default=None,
choices=("auto", "x264", "x265", "copy"),
help="video encoder: auto (hardware if possible), x264, x265, or copy (default: auto)",
)
parser.add_argument(
"--codec",
default=None,
choices=("h264", "hevc", "av1"),
help="hardware codec family to use when --video auto (default: h264)",
)
parser.add_argument(
"--device",
default=None,
metavar="PATH",
help=f"VA-API render node (default: $HW_DEVICE or {DEFAULT_DEVICE})",
)
parser.add_argument(
"--vf",
action="append",
default=[],
metavar="FILTER",
help="extra video filter, repeatable (appended after the built-in chain)",
)
def _add_audio_options(parser: argparse.ArgumentParser, *, default: str = "flac") -> None:
parser.add_argument(
"-f",
"--format",
default=default,
choices=tuple(AUDIO_FORMATS),
help=f"audio format preset (default: {default})",
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="ffmpeg_tools",
description="FFmpeg front-end for NCMR song videos (mux, bake, still, strip, probe).",
epilog="Run `%(prog)s hwprobe` for a separate encoder report, or see scripts/hwprobe.py.",
)
add_common_arguments(parser)
add_ffmpeg_argument(parser)
subparsers = parser.add_subparsers(dest="mode", metavar="MODE")
mux = subparsers.add_parser("mux", help="attach audio to a video-only file, video copied as-is")
add_common_arguments(mux, suppress_defaults=True)
mux.add_argument("input_video", help="video-only input (e.g. the Remotion render)")
mux.add_argument("audio", help="audio input (e.g. the exported song)")
mux.add_argument("-o", "--output", help="output path")
_add_audio_options(mux)
_add_video_options(mux)
mux.set_defaults(func=cmd_mux)
bake = subparsers.add_parser("bake", help="burn an ASS subtitle file into a video")
add_common_arguments(bake, suppress_defaults=True)
bake.add_argument("input_video", help="input video")
bake.add_argument("subtitles", help="ASS subtitle file")
bake.add_argument("-o", "--output", help="output path (default: <video>_sub.<ext>)")
_add_audio_options(bake, default="aac")
bake.add_argument(
"--audio-format",
default="keep",
choices=("keep", *AUDIO_FORMATS),
help="re-encode audio to this preset, or keep it as-is (default: keep)",
)
bake.add_argument("--shortest", action="store_true", help="stop at the shortest stream")
_add_video_options(bake)
bake.set_defaults(func=cmd_bake)
still = subparsers.add_parser("still", help="one image + audio -> video")
add_common_arguments(still, suppress_defaults=True)
still.add_argument("image", help="cover art image")
still.add_argument("audio", help="audio file")
still.add_argument("-o", "--output", help="output path")
still.add_argument("--fps", type=int, default=60, help="frame rate (default: 60)")
still.add_argument("--width", type=int, default=1920, help="target width (default: 1920)")
still.add_argument("--height", type=int, default=1080, help="target height (default: 1080)")
_add_audio_options(still)
_add_video_options(still)
still.set_defaults(func=cmd_still)
strip = subparsers.add_parser("strip", help="extract audio from a video as MP3")
add_common_arguments(strip, suppress_defaults=True)
strip.add_argument("input_video", help="input video")
strip.add_argument("-o", "--output", help="output MP3 path")
strip.add_argument("--sample-rate", type=int, default=44100, help="(default: 44100)")
strip.add_argument("--channels", type=int, default=2, help="(default: 2)")
strip.add_argument("--bitrate", default="320k", help="(default: 320k)")
strip.set_defaults(func=cmd_strip)
probe = subparsers.add_parser("probe", help="print a compact stream summary")
add_common_arguments(probe, suppress_defaults=True)
probe.add_argument("media", nargs=1, help="media file to inspect")
probe.set_defaults(func=cmd_probe)
return parser
@main_wrapper
def main() -> int:
parser = build_parser()
args = parser.parse_args()
report = Report(quiet=args.quiet)
if not getattr(args, "func", None):
parser.print_help()
return 2
args.ffmpeg = find_ffmpeg(args.ffmpeg)
# strip and probe never encode video, so they have no --video/--codec/--device
# and no acceleration context to work out.
if not hasattr(args, "video"):
return args.func(args, None, report)
# --video/--codec/--device live on the subparsers, so the namespace already
# carries whatever was parsed. Supply the documented defaults here rather than
# duplicating them across every subparser.
video = args.video if args.video is not None else "auto"
codec = args.codec if args.codec is not None else "h264"
device = args.device or os.environ.get("HW_DEVICE")
context = pick_acceleration(
args.ffmpeg,
report,
video=video,
codec=codec,
device=device,
)
if context.note and args.mode != "probe":
report.info(f"encoder: {context.encoder} — {context.note}")
return args.func(args, context, report)
def _supplied_options(argv: list[str]) -> dict[str, str]:
"""Map ``--flag`` to its value for flags present on the command line.
Retained for tests and external callers; ``main`` no longer needs it now that
the encoder options are parsed by argparse rather than sniffed out of argv.
"""
supplied: dict[str, str] = {}
i = 0
while i < len(argv):
token = argv[i]
if token.startswith("--") and "=" in token:
flag, _, value = token.partition("=")
supplied[flag] = value
elif token.startswith("--"):
if i + 1 < len(argv):
supplied[token] = argv[i + 1]
i += 1
i += 1
return supplied
def _decide(flag: str, supplied: dict[str, str], namespace_value, fallback):
"""Prefer a flag supplied on the command line, then the namespace, then default."""
if flag in supplied:
return supplied[flag]
return namespace_value if namespace_value is not None else fallback
if __name__ == "__main__":
raise SystemExit(main())
+460
View File
@@ -0,0 +1,460 @@
"""Detect which video encoders this machine can actually use.
Probes ffmpeg for hardware and software encoders and decides which are usable
here, then emits either a human-readable report or a shell ``export`` block.
The reference machine has no NVIDIA GPU, so ``*_nvenc`` encoders exist in the
ffmpeg build but cannot run. Detecting that up front is cheaper than discovering
it in the middle of a render.
Usage::
python3 scripts/hwprobe.py # human-readable report
python3 scripts/hwprobe.py --json # machine-readable
python3 scripts/hwprobe.py --shell # eval-able export block
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass, field, asdict
if __package__ in (None, ""):
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _common import ( # type: ignore[import-not-found]
Die,
Report,
add_common_arguments,
add_ffmpeg_argument,
find_ffmpeg,
main_wrapper,
report as default_report,
run,
text_table,
)
else: # pragma: no cover - exercised via `python3 -m scripts.hwprobe`
from ._common import (
Die,
Report,
add_common_arguments,
add_ffmpeg_argument,
find_ffmpeg,
main_wrapper,
report as default_report,
run,
text_table,
)
# --------------------------------------------------------------------------- #
# Candidate tables
# --------------------------------------------------------------------------- #
#: Hardware encoder families we know how to enable, keyed by ffmpeg suffix.
HW_FAMILIES = ("nvenc", "vaapi", "qsv", "amf", "v4l2m2m", "vulkan", "videotoolbox")
#: The encoders worth offering as a "best available" pick, in preference order.
#: Hardware first, then software fallbacks that always exist in a full build.
ENCODER_PREFERENCE = (
"h264_nvenc",
"hevc_nvenc",
"av1_nvenc",
"h264_qsv",
"hevc_qsv",
"h264_vaapi",
"hevc_vaapi",
"av1_vaapi",
"h264_amf",
"hevc_amf",
"h264_v4l2m2m",
"hevc_v4l2m2m",
"hevc_vulkan",
"libx264",
"libx265",
"libsvtav1",
"libvpx-vp9",
)
#: vaapi/vulkan/qsv devices we look for, in preference order.
DEVICE_CANDIDATES = (
"/dev/dri/renderD128",
"/dev/dri/renderD129",
"/dev/dri/card0",
"/dev/dri/card1",
)
@dataclass
class Encoder:
name: str
description: str
@property
def family(self) -> str:
for family in HW_FAMILIES:
if self.name.endswith(f"_{family}"):
return family
return "software"
@property
def is_hardware(self) -> bool:
return self.family != "software"
@dataclass
class ProbeResult:
ffmpeg: str
ffmpeg_version: str
encoders: list[Encoder] = field(default_factory=list)
hwaccels: list[str] = field(default_factory=list)
devices: dict[str, bool] = field(default_factory=dict)
tested: dict[str, dict] = field(default_factory=dict)
#: Encoder families that are compiled in but have no usable device.
blocked: dict[str, str] = field(default_factory=dict)
recommended: str | None = None
notes: list[str] = field(default_factory=list)
def encryptors(self) -> list[Encoder]:
return [e for e in self.encoders if e.is_hardware]
def parse_encoders(output: str) -> list[Encoder]:
"""Parse ``ffmpeg -encoders`` output into video encoders only."""
encoders: list[Encoder] = []
for line in output.splitlines():
match = re.match(r"^\s*([VAS][\w.]{5})\s+(\S+)\s+(.*\S)\s*$", line)
if not match:
continue
flags, name, description = match.groups()
if not flags.startswith("V"):
continue
encoders.append(Encoder(name=name, description=description))
return encoders
def parse_hwaccels(output: str) -> list[str]:
"""Parse ``ffmpeg -hwaccels`` output."""
methods: list[str] = []
seen_header = False
for line in output.splitlines():
line = line.strip()
if not line:
continue
if line.lower().startswith("hardware acceleration"):
seen_header = True
continue
if seen_header:
methods.append(line)
return methods
def probe_devices() -> dict[str, bool]:
"""Check which DRM devices exist *and* are readable by this process."""
found: dict[str, bool] = {}
for device in DEVICE_CANDIDATES:
if not os.path.exists(device):
continue
found[device] = os.access(device, os.R_OK | os.W_OK)
return found
def probe_device_driver(device: str) -> str | None:
"""Read the kernel driver name backing a DRM node, if sysfs is readable."""
node = os.path.basename(device)
uevent = f"/sys/class/drm/{node}/device/uevent"
try:
with open(uevent, "r", encoding="utf-8", errors="replace") as handle:
content = handle.read()
except OSError:
return None
match = re.search(r"^DRIVER=(\S+)", content, re.MULTILINE)
return match.group(1) if match else None
def missing_render_group(devices: dict[str, bool]) -> bool:
"""True when a DRM device exists but this process cannot open it."""
return any(not ok for ok in devices.values())
def test_encoder(ffmpeg: str, encoder: str, report: Report, device: str | None = None) -> tuple[bool, str]:
"""Encode two throwaway frames with *encoder*. Returns (worked, detail)."""
cmd = [ffmpeg, "-hide_banner", "-loglevel", "error"]
if encoder.endswith("_vaapi"):
if not device:
return False, "no vaapi device available"
cmd += [
"-init_hw_device", f"vaapi=va:{device}",
"-filter_hw_device", "va",
"-f", "lavfi",
"-i", "color=c=black:s=256x256:r=30:d=1",
"-vf", "format=nv12,hwupload",
]
elif encoder.endswith("_vulkan"):
cmd += ["-f", "lavfi", "-i", "color=c=black:s=256x256:r=30:d=1"]
else:
cmd += ["-f", "lavfi", "-i", "color=c=black:s=256x256:r=30:d=1"]
cmd += ["-c:v", encoder, "-frames:v", "2", "-f", "null", "-"]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
except subprocess.TimeoutExpired:
return False, "timed out after 60s"
except FileNotFoundError:
return False, "ffmpeg binary missing"
if result.returncode == 0:
return True, "ok"
stderr = (result.stderr or "").strip().splitlines()
return False, stderr[-1] if stderr else f"exit status {result.returncode}"
def probe(ffmpeg: str, report: Report, *, test: bool = True) -> ProbeResult:
"""Gather everything we can learn about this machine's encoders."""
def capture(args: list[str]) -> str:
try:
result = subprocess.run(
[ffmpeg, "-hide_banner", *args],
capture_output=True,
text=True,
timeout=30,
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise Die(f"could not run {ffmpeg}: {exc}") from exc
return result.stdout
banner = capture(["-version"])
version = banner.splitlines()[0] if banner else "unknown"
result = ProbeResult(
ffmpeg=ffmpeg,
ffmpeg_version=version,
encoders=parse_encoders(capture(["-encoders"])),
hwaccels=parse_hwaccels(capture(["-hwaccels"])),
devices=probe_devices(),
)
usable_device = next((d for d, ok in result.devices.items() if ok), None)
blocked_families: dict[str, str] = {}
if usable_device:
driver = probe_device_driver(usable_device)
if driver:
result.notes.append(f"{usable_device} is driven by {driver}")
elif result.devices:
blocked_families["vaapi"] = "DRM device present but not openable"
blocked_families["vulkan"] = "DRM device present but not openable"
else:
blocked_families["vaapi"] = "no DRM render device found"
if not test:
return result
for encoder in result.encoders:
if not encoder.is_hardware and encoder.name not in ("libx264",):
continue
if encoder.family in blocked_families:
result.tested[encoder.name] = {
"ok": False,
"detail": blocked_families[encoder.family],
"tested": False,
}
continue
if encoder.family in ("qsv", "amf", "videotoolbox", "v4l2m2m") and not result.devices:
result.tested[encoder.name] = {
"ok": False,
"detail": "no matching device",
"tested": False,
}
continue
ok, detail = test_encoder(ffmpeg, encoder.name, report, usable_device)
result.tested[encoder.name] = {"ok": ok, "detail": detail, "tested": True}
for family, reason in blocked_families.items():
if any(e.family == family for e in result.encoders):
result.blocked[family] = reason
for name in ENCODER_PREFERENCE:
entry = result.tested.get(name)
if entry and entry["ok"]:
result.recommended = name
break
return result
# --------------------------------------------------------------------------- #
# Output
# --------------------------------------------------------------------------- #
def render_report(result: ProbeResult) -> str:
lines: list[str] = []
lines.append(f"ffmpeg {result.ffmpeg_version}")
lines.append(f"binary {result.ffmpeg}")
lines.append("")
lines.append("DRM devices")
if result.devices:
for device in sorted(result.devices):
ok = result.devices[device]
driver = probe_device_driver(device) or "?"
state = "readable" if ok else "NOT READABLE"
lines.append(f" {device.ljust(20)} {driver.ljust(8)} {state}")
else:
lines.append(" none found — hardware encoding is unavailable")
lines.append("")
lines.append(f"hwaccels {', '.join(result.hwaccels) or '(none)'}")
lines.append("")
hw = result.encryptors()
sw = [e for e in result.encoders if not e.is_hardware]
lines.append(f"Hardware encoders in this build ({len(hw)})")
if hw:
rows = []
for encoder in hw:
entry = result.tested.get(encoder.name)
if entry is None:
verdict = "untested"
elif entry["ok"]:
verdict = "WORKS"
elif not entry.get("tested", True):
verdict = "unavailable"
else:
verdict = "failed"
rows.append((encoder.name, encoder.family, verdict))
lines.append(text_table(("encoder", "family", "status"), rows))
else:
lines.append(" none")
lines.append("")
lines.append(f"Software encoders ({len(sw)})")
notable = [e for e in sw if e.name in ("libx264", "libx265", "libsvtav1", "libvpx-vp9", "libxvid")]
lines.append(" " + (", ".join(e.name for e in notable) or "none"))
if result.blocked:
lines.append("")
lines.append("Blocked")
for family, reason in sorted(result.blocked.items()):
lines.append(f" {family}: {reason}")
if result.notes:
lines.append("")
for note in result.notes:
lines.append(f"note: {note}")
lines.append("")
if result.recommended:
lines.append(f"recommended -c:v {result.recommended}")
else:
lines.append("recommended (nothing usable detected)")
if missing_render_group(result.devices):
lines.append("")
lines.append("hint: a DRM device exists but this user cannot open it.")
lines.append(" add the user to the 'render' group and re-login:")
lines.append(" sudo usermod -aG render,video $USER")
return "\n".join(lines)
def render_shell(result: ProbeResult) -> str:
"""Export block for ``eval \"$(python3 scripts/hwprobe.py --shell)\"``."""
device = next((d for d, ok in result.devices.items() if ok), "")
lines = [
"# generated by scripts/hwprobe.py — eval to apply",
f"export FFMPEG={result.ffmpeg}",
]
for encoder in result.encryptors():
entry = result.tested.get(encoder.name, {})
if entry.get("ok"):
lines.append(f"export ENCODER_{encoder.family.upper()}={encoder.name}")
lines.append(f"export ENCODER={result.recommended or 'libx264'}")
if device:
lines.append(f"export HW_DEVICE={device}")
return "\n".join(lines)
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="hwprobe",
description="Report which ffmpeg encoders this machine can actually use.",
epilog=(
"Handy: eval \"$(python3 scripts/hwprobe.py --shell)\" "
"to put the detected encoder in $ENCODER."
),
)
add_common_arguments(parser)
add_ffmpeg_argument(parser)
parser.add_argument("--json", action="store_true", help="emit JSON instead of text")
parser.add_argument("--shell", action="store_true", help="emit an eval-able export block")
parser.add_argument(
"--no-test",
action="store_true",
help="only read ffmpeg's capability lists; do not try encoding",
)
return parser
@main_wrapper
def main() -> int:
args = build_parser().parse_args()
report = Report(quiet=args.quiet)
ffmpeg = find_ffmpeg(args.ffmpeg)
result = probe(ffmpeg, report, test=not args.no_test)
if args.json:
payload = asdict(result)
payload["blocked"] = result.blocked
payload["recommended"] = result.recommended
print(json.dumps(payload, indent=2))
elif args.shell:
print(render_shell(result))
else:
print(render_report(result))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+345 -59
View File
@@ -1,67 +1,353 @@
"""Convert LRC lyrics to SRT subtitles.
LRC is the timestamped-lyrics format most lyric sites and players emit. This
converter handles the shapes that actually show up in the wild:
* ``[mm:ss.xx]`` and ``[mm:ss.xxx]`` timestamps, and bare ``[mm:ss]``
* several timestamps on one line (``[00:05.00][00:30.00]chorus``), which is how
repeated sections are written
* ID tags (``[ti:Title]``, ``[ar:Artist]``, ``[offset:+250]``), which are
metadata, not lyrics
* multiple timestamped lines sharing one timestamp, which SRT wants as one cue
with two subtitle lines
Cue end times are inferred from the next cue's start, so the timing is driven by
the LRC rather than invented. The final cue gets ``--last-duration`` seconds.
Usage::
python3 scripts/lrc2srt.py song.lrc # -> song.srt
python3 scripts/lrc2srt.py song.lrc -o subs/song.srt
python3 scripts/lrc2srt.py lyrics/*.lrc --out-dir build/
python3 scripts/lrc2srt.py song.lrc --stdout # write to stdout
"""
from __future__ import annotations
import argparse
import os
import re import re
import sys import sys
import os from dataclasses import dataclass, field
from datetime import timedelta
def format_timestamp(timestamp): if __package__ in (None, ""):
hours, remainder = divmod(timestamp.total_seconds(), 3600) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
minutes, seconds = divmod(remainder, 60) from _common import ( # type: ignore[import-not-found]
milliseconds = timestamp.microseconds // 1000 Die,
return f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d},{int(milliseconds):03d}" Report,
add_common_arguments,
def lrc_to_srt(lrc_filename, srt_filename): main_wrapper,
with open(lrc_filename, 'r', encoding='utf-8') as lrc_file: )
lrc_content = lrc_file.readlines() else: # pragma: no cover
from ._common import (
srt_content = [] Die,
subtitle_counter = 1 Report,
previous_end_time = None add_common_arguments,
main_wrapper,
for i, line in enumerate(lrc_content): )
match = re.match(r'\[(\d{2}):(\d{2})\.(\d{2})\](.*)', line)
## Explanation to the regex:
## \[ (\d{2}) : (\d{2}) \. (\d{2}) \] (.*)
## bracket minutes (2 digits) colon seconds (2 digits) dot millisecs (2 digits) bracket lyrical line (can be blank)
if match:
print(f'debug: matched line {line.strip()}')
start_minutes = int(match.group(1))
start_seconds = int(match.group(2))
start_milliseconds = int(match.group(3)) * 10
print(f'matched text: {match.group(4)}')
start_time = timedelta(minutes=start_minutes, seconds=start_seconds, milliseconds=start_milliseconds)
if len(srt_content) > 0:
# srt_content[-1] += f"{format_timestamp(start_time - timedelta(milliseconds=1))}\n"
srt_content[-1] += f"{format_timestamp(start_time)}\n"
srt_content.append(f"{subtitle_text}\n\n")
srt_content.append(f"{subtitle_counter}\n")
srt_content.append(f"{format_timestamp(start_time)} --> ")
subtitle_text = match.group(4).strip()
subtitle_counter += 1
previous_end_time = start_time
if previous_end_time is not None:
srt_content[-1] += f"{format_timestamp(previous_end_time)}\n"
srt_content.append(f"{subtitle_text}\n\n")
with open(srt_filename, 'w', encoding='utf-8') as srt_file: # --------------------------------------------------------------------------- #
srt_file.writelines(srt_content) # Parsing
# --------------------------------------------------------------------------- #
#: ``[mm:ss.xx]`` / ``[mm:ss.xxx]`` / ``[mm:ss]``, anywhere in a line.
TIMESTAMP_RE = re.compile(r"\[(?P<m>\d{1,3}):(?P<s>\d{1,2})(?:[.:](?P<frac>\d{1,3}))?\]")
#: ``[ti:...]`` / ``[ar:...]`` / ``[offset:...]`` — an identifier, not a time.
TAG_RE = re.compile(r"^\[(?P<key>[A-Za-z#]+):(?P<value>.*)\]\s*$")
#: Keys whose value is a signed millisecond count, not a timestamp.
META_KEYS = {
"ti", "al", "ar", "au", "by", "re", "ve", "length", "offset", "encoding", "tool", "id",
}
@dataclass
class LyricLine:
timestamp: float
text: str
@dataclass
class LrcDocument:
lines: list[LyricLine] = field(default_factory=list)
metadata: dict[str, str] = field(default_factory=dict)
offset_ms: int = 0
@property
def title(self) -> str | None:
return self.metadata.get("ti")
@property
def artist(self) -> str | None:
return self.metadata.get("ar")
def parse_timestamp(minutes: str, seconds: str, fraction: str | None) -> float:
"""Turn LRC timestamp parts into seconds.
The fractional part is centiseconds in the common case (``.30`` = 300 ms) but
milliseconds when three digits are given (``.300`` = 300 ms). The length of
the field disambiguates.
"""
total = int(minutes) * 60 + int(seconds)
if fraction:
if len(fraction) == 3:
total += int(fraction) / 1000
else:
total += int(fraction) / 100
return total
def parse_lrc(content: str) -> LrcDocument:
"""Parse LRC text into a document, sorted by timestamp."""
if content.startswith("\ufeff"):
content = content[1:]
content = content.replace("\r\n", "\n").replace("\r", "\n")
document = LrcDocument()
for raw in content.split("\n"):
line = raw.strip()
if not line:
continue
# Metadata tags first: [ti:...] looks a lot like [00:12] to a loose regex.
tag = TAG_RE.match(line)
if tag and tag.group("key").lower() in META_KEYS:
key = tag.group("key").lower()
value = tag.group("value").strip()
if key == "offset":
try:
document.offset_ms = int(value)
except ValueError:
pass
else:
document.metadata[key] = value
continue
stamps = list(TIMESTAMP_RE.finditer(line))
if not stamps:
# An untimed line. Attach it to the previous cue as a continuation
# line rather than discarding the lyric entirely.
if document.lines:
previous = document.lines[-1]
previous.text = f"{previous.text}\n{line}"
continue
text = TIMESTAMP_RE.sub("", line).strip()
for stamp in stamps:
document.lines.append(
LyricLine(
timestamp=parse_timestamp(
stamp.group("m"), stamp.group("s"), stamp.group("frac")
),
text=text,
)
)
if document.offset_ms:
shift = document.offset_ms / 1000
for line in document.lines:
line.timestamp = max(0.0, line.timestamp + shift)
document.lines.sort(key=lambda l: l.timestamp)
return document
# --------------------------------------------------------------------------- #
# SRT output
# --------------------------------------------------------------------------- #
@dataclass
class Cue:
index: int
start: float
end: float
lines: list[str]
def format_srt_time(seconds: float) -> str:
"""Seconds to SRT ``HH:MM:SS,mmm``."""
if seconds < 0:
seconds = 0.0
total_ms = int(round(seconds * 1000))
hours, remainder = divmod(total_ms, 3_600_000)
minutes, remainder = divmod(remainder, 60_000)
secs, millis = divmod(remainder, 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
def build_cues(
lines: list[LyricLine],
*,
last_duration: float = 5.0,
min_duration: float = 0.5,
) -> list[Cue]:
"""Group simultaneous lines into cues and infer each cue's end time.
Lines sharing one timestamp become a single cue with several subtitle lines —
the format singers actually expect for a two-line phrase. A cue ends when the
*next distinct* timestamp arrives.
"""
if not lines:
return []
# Collapse lines with identical timestamps into one cue.
groups: list[tuple[float, list[str]]] = []
for line in lines:
if groups and abs(groups[-1][0] - line.timestamp) < 1e-6:
groups[-1][1].append(line.text)
else:
groups.append((line.timestamp, [line.text]))
cues: list[Cue] = []
for i, (start, text_lines) in enumerate(groups):
if i + 1 < len(groups):
end = groups[i + 1][0]
if end - start < min_duration:
end = start + min_duration
else:
end = start + last_duration
cues.append(Cue(index=i + 1, start=start, end=end, lines=text_lines))
return cues
def render_srt(cues: list[Cue]) -> str:
blocks = []
for cue in cues:
body = "\n".join(cue.lines)
blocks.append(
f"{cue.index}\n{format_srt_time(cue.start)} --> {format_srt_time(cue.end)}\n{body}"
)
return "\n\n".join(blocks) + "\n"
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="lrc2srt",
description="Convert LRC lyrics to SRT subtitles.",
epilog="Multiple timestamps on one line are expanded; ID tags are ignored.",
)
add_common_arguments(parser)
parser.add_argument("inputs", nargs="+", metavar="INPUT", help="LRC file(s) to convert")
parser.add_argument("-o", "--output", metavar="FILE", help="output path (single input only)")
parser.add_argument("--out-dir", metavar="DIR", help="write every output into this directory")
parser.add_argument("--stdout", action="store_true", help="write the SRT to stdout, no file")
parser.add_argument(
"--last-duration",
type=float,
default=5.0,
metavar="SECONDS",
help="how long to hold the final line (default: 5)",
)
parser.add_argument(
"--min-duration",
type=float,
default=0.5,
metavar="SECONDS",
help="shortest cue allowed (default: 0.5)",
)
parser.add_argument(
"--title",
help="override the SRT's implicit title (used for reporting only)",
)
return parser
def convert(src: str, args: argparse.Namespace, report: Report) -> tuple[int, str | None]:
try:
with open(src, "r", encoding="utf-8-sig", errors="replace") as handle:
content = handle.read()
except OSError as exc:
raise Die(f"cannot read {src}: {exc.strerror}") from exc
document = parse_lrc(content)
if not document.lines:
raise Die(f"{src}: no timestamped lyrics found — is this an LRC file?")
cues = build_cues(
document.lines,
last_duration=args.last_duration,
min_duration=args.min_duration,
)
output = render_srt(cues)
if args.stdout:
sys.stdout.write(output)
return len(cues), None
if args.output:
dest = args.output
elif args.out_dir:
dest = os.path.join(args.out_dir, os.path.splitext(os.path.basename(src))[0] + ".srt")
else:
dest = os.path.splitext(src)[0] + ".srt"
if args.dry_run:
report.info(f"[dry-run] would write {dest} ({len(cues)} cues)")
return len(cues), dest
os.makedirs(os.path.dirname(os.path.abspath(dest)), exist_ok=True)
with open(dest, "w", encoding="utf-8", newline="\n") as handle:
handle.write(output)
label = document.title or os.path.basename(src)
report.ok(f"{label} -> {dest} ({len(cues)} cues)")
if document.artist:
report.verbose(f"artist: {document.artist}")
if document.offset_ms:
report.verbose(f"applied LRC offset of {document.offset_ms:+d} ms")
return len(cues), dest
@main_wrapper
def main() -> int:
args = build_parser().parse_args()
report = Report(quiet=args.quiet)
if args.output and len(args.inputs) > 1:
raise Die("--output takes a single input; use --out-dir for batches")
if args.stdout and len(args.inputs) > 1:
raise Die("--stdout takes a single input")
total = 0
converted = 0
for src in args.inputs:
if not os.path.isfile(src):
raise Die(f"no such file: {src}")
cues, _dest = convert(src, args, report)
total += cues
converted += 1
if converted > 1:
report.info(f"{total} cues across {converted} file(s)")
return 0
print(f"Conversion from LRC to SRT completed. SRT file saved as '{srt_filename}'.")
if __name__ == "__main__": if __name__ == "__main__":
if not(len(sys.argv) == 3 or len(sys.argv) == 2): raise SystemExit(main())
print("Usage: python lrc2srt.py <input_lrc_file> <output_srt_file>")
print("or: python lrc2srt.py <input_lrc_file>")
print("If no output file name is supplied, the input file name will be used by default.")
else:
lrc_filename = sys.argv[1]
if len(sys.argv) == 3:
srt_filename = sys.argv[2]
else:
raw_filename, ext = os.path.splitext(lrc_filename)
srt_filename = raw_filename + '.srt'
lrc_to_srt(lrc_filename, srt_filename)
+382
View File
@@ -0,0 +1,382 @@
"""Convert an SRT subtitle file to ASS (Advanced SubStation Alpha).
Stdlib only. Handles the constructs that actually appear in song subtitles:
multi-line cues, ``{\\an8}``-style override blocks passed through untouched,
and the SRT quirks (CRLF, BOM, cue numbers, blank-line separators).
Usage::
python3 scripts/srt2ass.py song.srt song.ass
python3 scripts/srt2ass.py lyrics/*.srt --out-dir build/ # batch
python3 scripts/srt2ass.py song.srt --style-font "Noto Sans" --style-size 64
"""
from __future__ import annotations
import argparse
import os
import re
import sys
from dataclasses import dataclass
if __package__ in (None, ""):
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _common import ( # type: ignore[import-not-found]
Die,
Report,
add_common_arguments,
glob_files,
main_wrapper,
)
else: # pragma: no cover
from ._common import (
Die,
Report,
add_common_arguments,
glob_files,
main_wrapper,
)
# --------------------------------------------------------------------------- #
# Timestamps
# --------------------------------------------------------------------------- #
_SRT_TIME_RE = re.compile(
r"(?P<h>\d{1,2}):(?P<m>\d{2}):(?P<s>\d{2})[,.](?P<ms>\d{1,3})"
)
def parse_srt_time(stamp: str) -> float:
"""``00:01:02,500`` (or ``00:01:02.5``) to seconds."""
match = _SRT_TIME_RE.search(stamp.strip())
if not match:
raise Die(f"unparseable SRT timestamp: {stamp!r}")
ms = match.group("ms").ljust(3, "0")[:3]
return (
int(match.group("h")) * 3600
+ int(match.group("m")) * 60
+ int(match.group("s"))
+ int(ms) / 1000
)
def format_ass_time(seconds: float) -> str:
"""Seconds to ASS ``H:MM:SS.cc`` (centiseconds, single-digit hour)."""
if seconds < 0:
seconds = 0.0
total_cs = int(round(seconds * 100))
hours, remainder = divmod(total_cs, 360_000)
minutes, remainder = divmod(remainder, 6_000)
secs, centis = divmod(remainder, 100)
return f"{hours}:{minutes:02d}:{secs:02d}.{centis:02d}"
# --------------------------------------------------------------------------- #
# Parsing
# --------------------------------------------------------------------------- #
@dataclass
class Cue:
index: int
start: float
end: float
lines: list[str]
def text(self, line_sep: str = r"\N") -> str:
return line_sep.join(self.lines)
def strip_bom(text: str) -> str:
return text[1:] if text.startswith("\ufeff") else text
def parse_srt(content: str) -> list[Cue]:
"""Parse SRT text into cues. Tolerates CRLF, BOM, and missing cue numbers."""
content = strip_bom(content).replace("\r\n", "\n").replace("\r", "\n")
blocks = re.split(r"\n\s*\n", content)
cues: list[Cue] = []
auto_index = 0
for block in blocks:
lines = block.strip("\n").split("\n")
if not lines or not any(line.strip() for line in lines):
continue
index: int | None = None
if lines[0].strip().isdigit():
index = int(lines[0].strip())
lines = lines[1:]
if not lines:
continue
if "-->" not in lines[0]:
# Not a cue (stray header, WEBVTT marker, ...). Skip it quietly.
continue
start_raw, _, end_raw = lines[0].partition("-->")
start = parse_srt_time(start_raw)
end = parse_srt_time(end_raw)
body = [line for line in lines[1:] if line.strip()]
if not body:
# A cue with a timestamp but no text: pure gap in the lyrics. Emitting
# it would produce an empty subtitle block, so drop it — the previous
# cue simply holds until the next one.
continue
auto_index += 1
cues.append(Cue(index=index or auto_index, start=start, end=end, lines=body))
return cues
# --------------------------------------------------------------------------- #
# ASS output
# --------------------------------------------------------------------------- #
#: ASS uses US-ASCII-with-code-page-1252 in practice; karaoke/Japanese needs 1.
ASS_CODEPAGES = {
"ascii": 0,
"latin1": 0,
"cp1252": 0,
"shift_jis": 1,
"sjis": 1,
"gbk": 134,
"utf-8": 1,
"utf8": 1,
}
ASS_HEADER = """[Script Info]
; Generated by scripts/srt2ass.py — edit the source SRT, not this file.
Title: {title}
ScriptType: v4.00+
WrapStyle: {wrap_style}
ScaledBorderAndShadow: yes
YCbCr Matrix: TV.709
PlayResX: {width}
PlayResY: {height}
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,{font},{size},{primary},{secondary},{outline_colour},{back},{bold},{italic},0,0,100,100,{spacing},0,1,{border},{shadow},{alignment},{margin_l},{margin_r},{margin_v},{encoding}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
@dataclass
class AssStyle:
"""The subset of the Default style we expose on the CLI."""
font: str = "Arial"
size: int = 48
primary: str = "&H00FFFFFF" # white, fully opaque
secondary: str = "&H000000FF"
outline_colour: str = "&H00000000"
back: str = "&H00000000"
bold: int = -1
italic: int = 0
spacing: float = 0.0
border: float = 2.0
shadow: float = 1.0
alignment: int = 2 # 2 = bottom centre
margin_l: int = 40
margin_r: int = 40
margin_v: int = 60
encoding: int = 1
def render_ass(
cues: list[Cue],
*,
style: AssStyle,
width: int,
height: int,
title: str,
wrap_style: int = 0,
) -> str:
header = ASS_HEADER.format(
title=title.replace("\n", " ") or "Untitled",
wrap_style=wrap_style,
width=width,
height=height,
font=style.font,
size=style.size,
primary=style.primary,
secondary=style.secondary,
outline_colour=style.outline_colour,
back=style.back,
bold=style.bold,
italic=style.italic,
spacing=style.spacing,
border=style.border,
shadow=style.shadow,
alignment=style.alignment,
margin_l=style.margin_l,
margin_r=style.margin_r,
margin_v=style.margin_v,
encoding=style.encoding,
)
events = []
for cue in cues:
text = cue.text()
# ASS treats commas as field separators in the Text column, but only the
# first 9 commas matter — ffmpeg/libass handles commas inside Text fine as
# long as the Format line lists Text last, which it does above.
events.append(
"Dialogue: 0,{start},{end},Default,,0,0,0,,{text}".format(
start=format_ass_time(cue.start),
end=format_ass_time(cue.end),
text=text,
)
)
return header + "\n".join(events) + "\n"
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="srt2ass",
description="Convert SRT subtitles to ASS, ready for ffmpeg's ass= filter.",
epilog=(
"Then bake with: ffmpeg -vf \"ass=out.ass\" -i in.mp4 ... "
"(or scripts/ffmpeg_tools.py bake)"
),
)
add_common_arguments(parser)
parser.add_argument("inputs", nargs="+", metavar="INPUT", help="SRT file(s) to convert")
parser.add_argument(
"-o",
"--output",
metavar="FILE",
help="output .ass path (single input only; default: alongside the input)",
)
parser.add_argument(
"--out-dir",
metavar="DIR",
help="write every output into this directory (created if missing)",
)
style = parser.add_argument_group("style")
style.add_argument("--style-font", default="Arial", help="font name (default: Arial)")
style.add_argument("--style-size", type=int, default=48, help="font size (default: 48)")
style.add_argument(
"--style-alignment",
type=int,
default=2,
choices=range(1, 10),
metavar="N",
help="ASS numpad alignment 1-9 (default: 2, bottom centre)",
)
style.add_argument("--style-margin-v", type=int, default=60, help="vertical margin (default: 60)")
style.add_argument("--style-outline", type=float, default=2.0, help="outline width (default: 2)")
style.add_argument("--style-shadow", type=float, default=1.0, help="shadow depth (default: 1)")
style.add_argument("--style-bold", action=argparse.BooleanOptionalAction, default=True)
style.add_argument("--style-italic", action=argparse.BooleanOptionalAction, default=False)
canvas = parser.add_argument_group("canvas")
canvas.add_argument("--width", type=int, default=1920, help="PlayResX (default: 1920)")
canvas.add_argument("--height", type=int, default=1080, help="PlayResY (default: 1080)")
canvas.add_argument(
"--title",
default=None,
help="ASS Title field (default: the output filename)",
)
return parser
def convert_one(src: str, dest: str, args: argparse.Namespace, report: Report) -> tuple[int, int]:
try:
with open(src, "r", encoding="utf-8-sig", errors="replace") as handle:
content = handle.read()
except OSError as exc:
raise Die(f"cannot read {src}: {exc.strerror}") from exc
cues = parse_srt(content)
if not cues:
report.warn(f"{src}: no cues parsed — is this really an SRT file?")
style = AssStyle(
font=args.style_font,
size=args.style_size,
alignment=args.style_alignment,
margin_v=args.style_margin_v,
border=args.style_outline,
shadow=args.style_shadow,
bold=-1 if args.style_bold else 0,
italic=-1 if args.style_italic else 0,
)
title = args.title or os.path.splitext(os.path.basename(dest))[0]
document = render_ass(
cues,
style=style,
width=args.width,
height=args.height,
title=title,
)
if args.dry_run:
report.info(f"[dry-run] would write {dest} ({len(cues)} cues)")
return len(cues), 0
os.makedirs(os.path.dirname(os.path.abspath(dest)), exist_ok=True)
with open(dest, "w", encoding="utf-8", newline="\n") as handle:
handle.write(document)
report.ok(f"{src} -> {dest} ({len(cues)} cues)")
return len(cues), 0
@main_wrapper
def main() -> int:
args = build_parser().parse_args()
report = Report(quiet=args.quiet)
if args.output and len(args.inputs) > 1:
raise Die("--output takes a single input; use --out-dir for batches")
total = 0
for src in args.inputs:
if not os.path.isfile(src):
raise Die(f"no such file: {src}")
stem = os.path.splitext(os.path.basename(src))[0]
if args.output:
dest = args.output
elif args.out_dir:
dest = os.path.join(args.out_dir, stem + ".ass")
else:
dest = os.path.splitext(src)[0] + ".ass"
cues, _ = convert_one(src, dest, args, report)
total += cues
if len(args.inputs) > 1:
report.info(f"{total} cues across {len(args.inputs)} file(s)")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+99
View File
@@ -0,0 +1,99 @@
#!/usr/bin/env python3
"""Unified entry point for the NCMR songs-videos helper tools.
Dispatches to a sibling script, forwards all remaining arguments, and keeps the
exit status. Use whichever feels natural — this dispatcher, or the script
directly.
python3 scripts/tools.py --help
python3 scripts/tools.py hwprobe
python3 scripts/tools.py tools lrc2srt song.lrc -o song.srt
The same thing, called directly:
python3 scripts/hwprobe.py
"""
from __future__ import annotations
import os
import runpy
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
#: name -> (module filename, one-line summary)
TOOLS: dict[str, tuple[str, str]] = {
"hwprobe": ("hwprobe.py", "which ffmpeg encoders this machine can actually use"),
"ffmpeg": ("ffmpeg_tools.py", "mux, bake subtitles, build still-image videos, strip audio"),
"lrc2srt": ("lrc2srt.py", "convert LRC lyrics to SRT"),
"srt2ass": ("srt2ass.py", "convert SRT subtitles to ASS"),
"bom": ("bom_gen.py", "generate the untracked-asset bill of materials"),
}
ALIASES = {
"ffmpeg_tools": "ffmpeg",
"ffmpeg-tools": "ffmpeg",
"bom_gen": "bom",
"bom-gen": "bom",
"lrc": "lrc2srt",
"srt": "srt2ass",
}
def usage(stream=sys.stdout) -> None:
print("usage: tools.py <tool> [options] [-- <tool options>]", file=stream)
print(" tools.py --help | --list", file=stream)
print(file=stream)
print("tools:", file=stream)
width = max(len(name) for name in TOOLS)
for name, (_filename, summary) in TOOLS.items():
print(f" {name.ljust(width)} {summary}", file=stream)
print(file=stream)
print("Every tool also runs standalone: python3 scripts/<name>.py --help", file=stream)
def main(argv: list[str] | None = None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
if not argv or argv[0] in ("-h", "--help", "help", "list", "--list"):
usage(stream=sys.stderr if not argv else sys.stdout)
return 0 if argv else 2
name = ALIASES.get(argv[0], argv[0])
if name not in TOOLS:
print(f"tools.py: unknown tool {argv[0]!r}", file=sys.stderr)
print(file=sys.stderr)
usage(stream=sys.stderr)
return 2
filename, _summary = TOOLS[name]
path = os.path.join(HERE, filename)
if not os.path.isfile(path):
print(f"tools.py: {filename} is missing from {HERE}", file=sys.stderr)
return 1
remaining = argv[1:]
if remaining and remaining[0] == "--":
remaining = remaining[1:]
sys.argv = [path, *remaining]
# run_path gives the child the same process, so its exit code is ours.
try:
runpy.run_path(path, run_name="__main__")
except SystemExit as exc:
code = exc.code
if code is None:
return 0
if isinstance(code, int):
return code
print(code, file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())