Compare commits
17 Commits
b73e14cdd1
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| be81469c7c | |||
| 4bc2346480 | |||
| 2c6e227c82 | |||
| afe397da3f | |||
| a1fbd64f87 | |||
| 68c12caef9 | |||
| 51b878486f | |||
| 78e8a7decf | |||
| 475783556f | |||
| a7823f1747 | |||
| 18de75ee28 | |||
| edb5f85a75 | |||
| d25e693a4a | |||
| 74b70967a1 | |||
| 0de9c41c7d | |||
| 191077af1c | |||
| c86afdc75a |
+53
@@ -0,0 +1,53 @@
|
||||
# Binary input assets (per README guideline)
|
||||
# Audio
|
||||
*.mp3
|
||||
*.wav
|
||||
*.flac
|
||||
*.aac
|
||||
*.ogg
|
||||
*.wma
|
||||
*.m4a
|
||||
|
||||
# Images
|
||||
*.png
|
||||
*.jpg
|
||||
*.jpeg
|
||||
*.gif
|
||||
*.webp
|
||||
*.svg
|
||||
*.bmp
|
||||
*.ico
|
||||
|
||||
# Fonts
|
||||
*.ttf
|
||||
*.otf
|
||||
*.woff
|
||||
*.woff2
|
||||
|
||||
# Material video clips
|
||||
*.mp4
|
||||
*.mov
|
||||
*.avi
|
||||
*.mkv
|
||||
*.webm
|
||||
|
||||
# Video output (Remotion out/ + any rendered videos)
|
||||
node_modules
|
||||
dist
|
||||
out
|
||||
.env
|
||||
|
||||
# Python (helper scripts)
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
.venv/
|
||||
|
||||
# Package lock — deliberately NOT tracked.
|
||||
# Every dependency in package.json is pinned to an exact version (no ^ or ~), so
|
||||
# a lockfile has no floating ranges to guard against. What it would add instead is
|
||||
# trouble across machines: the lock captures the resolved tree including
|
||||
# platform-specific optional binaries (e.g. @rollup/rollup-linux-x64-gnu), so a
|
||||
# Linux-generated lock is wrong on Windows and vice versa, and `npm ci` refuses
|
||||
# outright when the lock and manifest disagree. Revisit this if a dependency is
|
||||
# ever loosened to a range or picks up meaningful transitive drift.
|
||||
package-lock.json
|
||||
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"useTabs": false,
|
||||
"bracketSpacing": true,
|
||||
"tabWidth": 2
|
||||
}
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
# Archive tags
|
||||
|
||||
Tags whose only purpose is to keep a commit reachable after the branch that
|
||||
carried it was deleted. Nothing here is active code.
|
||||
|
||||
| Tag | Kept because |
|
||||
| --- | --- |
|
||||
| `archive/feat-bom-generator` | The branch held one commit (`8f860e0`, "BOM generator v2 with hashes") that never landed on `master`. The capability — a content hash per asset — was ported forward to `scripts/bom_gen.py --hash` in `2c6e227`, and the TypeScript file it patched (`scripts/bom-gen.ts`) was deleted in `afe397d`. The tag remains so the original commit is reachable if the port is ever questioned. |
|
||||
|
||||
Tags are pushed explicitly (`git push origin archive/<name>`); they are not
|
||||
picked up by a plain `git push`.
|
||||
@@ -1,24 +1,89 @@
|
||||
# Video-related files for NCMR songs
|
||||
|
||||
This repo houses the video-related, code-like files for NCMR music videos.
|
||||
This repo houses the code-like files for NCMR music videos. Binary assets live elsewhere.
|
||||
|
||||
## Structure
|
||||
|
||||
> **Repo state: workspace under construction.** `master` is *not* locked in yet —
|
||||
> it is still being assembled as the shared template. Until it is locked, any
|
||||
> branching strategy is fine and template/tooling work can land on `master`
|
||||
> directly. The branch conventions below describe the intended end state, not a
|
||||
> rule currently in force.
|
||||
|
||||
### `master` branch — shared template
|
||||
|
||||
This branch holds the common scaffolding:
|
||||
|
||||
- **Remotion project** — entry point, root component, and a placeholder composition
|
||||
- **`scripts/`** — Python toolchain (stdlib only) for subtitles, ffmpeg and asset manifests
|
||||
- **Build config** — TypeScript, ESLint, Prettier, Remotion studio
|
||||
|
||||
### `src/` vs `scripts/`
|
||||
|
||||
Two directories, two jobs:
|
||||
|
||||
- **`src/`** — Remotion input only. Everything here is compiled by `tsc`/Remotion
|
||||
and linted by `npm run lint`.
|
||||
- **`scripts/`** — operator tooling. Python, standard library only, run directly
|
||||
(`python3 scripts/tools.py --help`). Nothing here is part of the Remotion build.
|
||||
|
||||
Keep the boundary sharp: anything the video build compiles belongs in `src/`, and
|
||||
anything you run by hand belongs in `scripts/`.
|
||||
|
||||
### Song branches
|
||||
|
||||
For each song, create a branch `<song-name>/master` (and optionally feature branches
|
||||
like `<song-name>/synced-lyrics-remotion`):
|
||||
|
||||
```
|
||||
git checkout -b <song-name>/master
|
||||
```
|
||||
|
||||
Copy the shared template into the song branch, then add song-specific files:
|
||||
|
||||
```
|
||||
mkdir -p <song-name>
|
||||
# Place your song-specific compositions, assets manifests, etc.
|
||||
```
|
||||
|
||||
Once `master` is locked in it remains protected and mostly immutable. Until then,
|
||||
song branches are a rehearsal for that end state, not a requirement.
|
||||
|
||||
## Guideline
|
||||
|
||||
### gitignored (not tracked here)
|
||||
- binary input assets (audio, image, fonts, material video clips if any)
|
||||
- video output
|
||||
### Git-ignored (not tracked here)
|
||||
|
||||
For untracked input assets, a bill of materials listing everything and explaining what each is will be mandatory.
|
||||
- Binary input assets (audio, images, fonts, material video clips)
|
||||
- Video output (`out/`, rendered files)
|
||||
- `package-lock.json` — see below
|
||||
|
||||
### branching rule
|
||||
For untracked input assets, a **bill of materials** listing everything and explaining
|
||||
what each file is will be mandatory. Keep it as `BOM.md` alongside your song assets.
|
||||
|
||||
This repo has specialised local branching rules and will supersede any generic git operation conventions.
|
||||
### Why there is no `package-lock.json`
|
||||
|
||||
- `master` serves as the minimal template and remain protected and mostly immutable
|
||||
- [Remotion](https://www.remotion.dev/) project scaffolding
|
||||
- `ffmpeg` helper script for baking in `ass` styled subtitles or bypassing the Remotion route (static image + audio)
|
||||
- legacy lrc->srt conversion script
|
||||
Deliberate, not an oversight. Work happens on both Windows and Linux, and two
|
||||
things make a lockfile a liability here rather than a safeguard:
|
||||
|
||||
- for each song, a song root named `<song-name>/master` should be created, for the very least
|
||||
- usually, working on a song root incrementally is the norm and totally acceptable
|
||||
- branches for parallel features (e.g. `<song-name>/synced-lyrics-remotion`, `<song-name>/overlay-ass`) may be created.
|
||||
1. **There is nothing for it to pin.** Every dependency in `package.json` is
|
||||
pinned to an exact version — no `^`, no `~`. A plain `npm install` on any
|
||||
machine resolves to the same versions the lockfile would have recorded.
|
||||
2. **A lockfile is platform-specific.** It records the resolved tree, including
|
||||
the optional native binaries for whichever OS generated it (`@rollup/rollup-linux-x64-gnu`,
|
||||
`@rollup/rollup-win32-x64-msvc`). Committed from Linux it is wrong on Windows,
|
||||
and `npm ci` refuses outright rather than adapting, since its whole purpose is
|
||||
to reproduce the recorded tree exactly.
|
||||
|
||||
So it would be strict about the wrong thing: breaking installs across machines
|
||||
while guarding versions that were never floating. Revisit this if a dependency is
|
||||
ever loosened to a range, or picks up transitive drift that exact pinning cannot
|
||||
control — at that point the lockfile earns its friction.
|
||||
|
||||
### Branching
|
||||
|
||||
- `master` — minimal template (protected once locked in)
|
||||
- `<song>/master` — song-specific work
|
||||
- `<song>/<feature>` — parallel features for a song
|
||||
|
||||
Once `master` is locked, this repo's branching rules supersede any generic git
|
||||
operation conventions. Before that point, use whatever is convenient.
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
# Remotion Quickstart
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Node.js 18+
|
||||
- ffmpeg installed on your system
|
||||
|
||||
## Getting started
|
||||
|
||||
```bash
|
||||
npm install # install dependencies
|
||||
npm run dev # start Remotion Studio
|
||||
```
|
||||
|
||||
Remotion Studio opens at **http://localhost:3000** in your browser. If 3000 is already in use, Remotion automatically bumps to 3001, 3002, etc. — watch the terminal output for the actual port number. You can also pin a specific port:
|
||||
|
||||
```bash
|
||||
npx remotion studio --port=3001
|
||||
```
|
||||
|
||||
## Commands
|
||||
|
||||
| Command | What it does |
|
||||
|----------------------|----------------------------------------------|
|
||||
| `npm run dev` | Starts Remotion Studio (live preview UI) |
|
||||
| `npm run build` | Bundles the project for production |
|
||||
| `npx remotion render`| Renders a video from the command line |
|
||||
| `npx remotion upgrade` | Upgrades all Remotion packages to latest |
|
||||
|
||||
## Remotion Studio
|
||||
|
||||
`npm run dev` starts the Studio — a browser-based editor where you can:
|
||||
|
||||
- Preview compositions in real time
|
||||
- Scrub through frames
|
||||
- Change props and see changes instantly
|
||||
- Render videos from the browser
|
||||
|
||||
The Studio auto-reloads on file changes. No manual refresh needed.
|
||||
|
||||
## Project structure
|
||||
|
||||
```
|
||||
.
|
||||
├── src/ # Remotion input — compiled by tsc/Remotion
|
||||
│ ├── index.ts # Entry point — calls registerRoot()
|
||||
│ ├── Root.tsx # React root component
|
||||
│ └── Composition.tsx # Composition definitions (placeholder)
|
||||
├── scripts/ # Operator tooling — Python, stdlib only
|
||||
│ ├── tools.py # Dispatcher: python3 scripts/tools.py --help
|
||||
│ ├── ffmpeg_tools.py # mux / bake / still / strip / probe
|
||||
│ ├── lrc2srt.py # LRC → SRT
|
||||
│ ├── srt2ass.py # SRT → ASS (the step before baking)
|
||||
│ ├── bom_gen.py # Bill of materials for untracked assets
|
||||
│ └── hwprobe.py # What this machine can actually encode
|
||||
├── package.json # Remotion + React + Tailwind deps
|
||||
├── remotion.config.ts # CLI config (rspack, jpeg output format)
|
||||
└── tsconfig.json # TypeScript (strict, ES2018)
|
||||
```
|
||||
|
||||
`src/` and `scripts/` do not mix: the former is what the video build compiles, the
|
||||
latter is what you run by hand.
|
||||
|
||||
## Compositions
|
||||
|
||||
Compositions are defined in `src/Composition.tsx`. Each has:
|
||||
|
||||
- **id** — unique name
|
||||
- **durationInFrames** — length in frames
|
||||
- **fps** — frames per second
|
||||
- **width/height** — resolution
|
||||
|
||||
The placeholder is 1280×720 at 30fps, 60 frames (2 seconds).
|
||||
|
||||
## Rendering a video
|
||||
|
||||
```bash
|
||||
npx remotion render src/index.ts MyComp out/video.mp4
|
||||
```
|
||||
|
||||
Replace `MyComp` with your composition ID. The output goes to `out/`.
|
||||
|
||||
## Next steps
|
||||
|
||||
- [Remotion docs](https://www.remotion.dev/docs)
|
||||
- [Compositions guide](https://www.remotion.dev/docs/the-introduction-video)
|
||||
- [Props guide](https://www.remotion.dev/docs/props)
|
||||
@@ -0,0 +1,3 @@
|
||||
import { config } from "@remotion/eslint-config-flat";
|
||||
|
||||
export default config;
|
||||
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"name": "ncmr-songs-videos",
|
||||
"version": "0.1.0",
|
||||
"description": "Video production templates for NCMR songs — Remotion project, ffmpeg helpers, and subtitle tools",
|
||||
"private": true,
|
||||
"license": "UNLICENSED",
|
||||
"dependencies": {
|
||||
"@remotion/cli": "4.0.522",
|
||||
"@remotion/tailwind-v4": "4.0.522",
|
||||
"react": "19.2.3",
|
||||
"react-dom": "19.2.3",
|
||||
"remotion": "4.0.522",
|
||||
"tailwindcss": "4.0.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@remotion/eslint-config-flat": "4.0.522",
|
||||
"@types/react": "19.2.7",
|
||||
"@types/web": "0.0.166",
|
||||
"eslint": "9.19.0",
|
||||
"prettier": "3.8.1",
|
||||
"typescript": "5.9.3"
|
||||
},
|
||||
"scripts": {
|
||||
"dev": "remotion studio",
|
||||
"build": "remotion bundle",
|
||||
"upgrade": "remotion upgrade",
|
||||
"lint": "eslint src && tsc"
|
||||
},
|
||||
"sideEffects": [
|
||||
"*.css"
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
/**
|
||||
* Note: When using the Node.JS APIs, the config file
|
||||
* doesn't apply. Instead, pass options directly to the APIs.
|
||||
*
|
||||
* All configuration options: https://remotion.dev/docs/config
|
||||
*/
|
||||
|
||||
import { Config } from "@remotion/cli/config";
|
||||
import { enableTailwind } from '@remotion/tailwind-v4';
|
||||
|
||||
Config.setRspack(true);
|
||||
Config.setVideoImageFormat("jpeg");
|
||||
Config.setOverwriteOutput(true);
|
||||
Config.overrideBundlerConfig(enableTailwind);
|
||||
@@ -0,0 +1,217 @@
|
||||
# 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 `nvenc` is unavailable 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. `hwprobe.py`
|
||||
reports what this host can actually encode, so you can tell at a glance whether a
|
||||
recipe will need a hardware-specific encoder.
|
||||
|
||||
---
|
||||
|
||||
## 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.
|
||||
|
||||
Add `--hash` to record a short content hash per file:
|
||||
|
||||
```sh
|
||||
python3 scripts/tools.py bom --dir . --hash
|
||||
```
|
||||
|
||||
That is how you answer "has this asset changed since the BOM was written?" —
|
||||
sizes stay identical when you re-render a take at the same length, but the hash
|
||||
will not. Hashing reads every asset in full, so it is off by default; when you
|
||||
only want the inventory, leave it out.
|
||||
|
||||
---
|
||||
|
||||
## Conventions
|
||||
|
||||
- `master` is the shared template. It is intended to freeze once the workspace is
|
||||
finalised; until then, tooling work can land on it directly. Per-song work goes
|
||||
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`.
|
||||
@@ -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
|
||||
@@ -0,0 +1,516 @@
|
||||
"""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 hashlib
|
||||
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 = ""
|
||||
digest: str = "" # short content hash, empty when --hash was not requested
|
||||
extra: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def directory(self) -> str:
|
||||
return os.path.dirname(self.path).split("/")[0] or "."
|
||||
|
||||
|
||||
def file_digest(full: str, *, length: int = 12, chunk: int = 1 << 20) -> str:
|
||||
"""Short BLAKE2b content hash, streamed so large assets do not land in RAM.
|
||||
|
||||
Assets here are audio and video files that can run to hundreds of megabytes,
|
||||
so this reads in chunks rather than slurping the whole file. BLAKE2b is in
|
||||
the standard library and is faster than md5 with no collision worries.
|
||||
``length`` bytes of digest = 3 * length hex characters.
|
||||
"""
|
||||
h = hashlib.blake2b(digest_size=length)
|
||||
with open(full, "rb") as fh:
|
||||
while bytes_read := fh.read(chunk):
|
||||
h.update(bytes_read)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def walk_assets(
|
||||
root: str,
|
||||
dirs: tuple[tuple[str, str], ...] = ASSET_DIRS,
|
||||
*,
|
||||
hashes: bool = False,
|
||||
) -> list[Asset]:
|
||||
"""Collect files under each asset directory, sorted by path.
|
||||
|
||||
Hashes are opt-in: they cost a full read of every asset, which is wasted
|
||||
work when you only want the inventory.
|
||||
"""
|
||||
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),
|
||||
digest=file_digest(full) if hashes else "",
|
||||
)
|
||||
)
|
||||
|
||||
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}")
|
||||
if asset.digest:
|
||||
lines.append(f" hash: {asset.digest}")
|
||||
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}")
|
||||
if asset.digest:
|
||||
lines.append(f" hash: {asset.digest}")
|
||||
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}")
|
||||
if asset.digest:
|
||||
lines.append(f" hash: {asset.digest}")
|
||||
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("")
|
||||
# Only carry a hash column when hashes were asked for, so the common
|
||||
# inventory stays narrow enough to read.
|
||||
with_hash = any(asset.digest for asset in grouping)
|
||||
rows = [
|
||||
(
|
||||
f"`{asset.path}`",
|
||||
human_size(asset.size),
|
||||
*( (f"`{asset.digest}`",) if with_hash else () ),
|
||||
asset.description or f"_{_DESCRIPTION_PLACEHOLDER}_",
|
||||
)
|
||||
for asset in grouping
|
||||
]
|
||||
headers = ("file", "size", "hash", "description") if with_hash else ("file", "size", "description")
|
||||
lines.append(text_table(headers, 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(
|
||||
"--hash",
|
||||
dest="hashes",
|
||||
action="store_true",
|
||||
help="record a short content hash per file (reads every asset; slower)",
|
||||
)
|
||||
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, hashes=args.hashes)
|
||||
|
||||
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())
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -0,0 +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 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,
|
||||
main_wrapper,
|
||||
)
|
||||
else: # pragma: no cover
|
||||
from ._common import (
|
||||
Die,
|
||||
Report,
|
||||
add_common_arguments,
|
||||
main_wrapper,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 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
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -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())
|
||||
@@ -0,0 +1,98 @@
|
||||
#!/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",
|
||||
"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())
|
||||
@@ -0,0 +1,25 @@
|
||||
import { CalculateMetadataFunction, Composition } from "remotion";
|
||||
|
||||
type Props = {};
|
||||
|
||||
const calculateMetadata: CalculateMetadataFunction<Props> = () => {
|
||||
return {};
|
||||
};
|
||||
|
||||
export const MyComposition = () => {
|
||||
return (
|
||||
<Composition
|
||||
id="MyComp"
|
||||
component={MyComponent}
|
||||
durationInFrames={60}
|
||||
fps={30}
|
||||
width={1280}
|
||||
height={720}
|
||||
calculateMetadata={calculateMetadata}
|
||||
/>
|
||||
);
|
||||
};
|
||||
|
||||
export const MyComponent: React.FC<Props> = () => {
|
||||
return null;
|
||||
};
|
||||
@@ -0,0 +1,10 @@
|
||||
import "./index.css";
|
||||
import { MyComposition } from "./Composition";
|
||||
|
||||
export const RemotionRoot: React.FC = () => {
|
||||
return (
|
||||
<>
|
||||
<MyComposition />
|
||||
</>
|
||||
);
|
||||
};
|
||||
@@ -0,0 +1 @@
|
||||
@import "tailwindcss";
|
||||
@@ -0,0 +1,4 @@
|
||||
import { registerRoot } from "remotion";
|
||||
import { RemotionRoot } from "./Root";
|
||||
|
||||
registerRoot(RemotionRoot);
|
||||
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2018",
|
||||
"module": "Preserve",
|
||||
"moduleResolution": "Bundler",
|
||||
"jsx": "react-jsx",
|
||||
"strict": true,
|
||||
"noEmit": true,
|
||||
"lib": ["es2015"],
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"forceConsistentCasingInFileNames": true,
|
||||
"noUnusedLocals": true
|
||||
},
|
||||
"exclude": ["remotion.config.ts"]
|
||||
}
|
||||
Reference in New Issue
Block a user