diff --git a/.gitignore b/.gitignore index 208ee49a..6696ab2e 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ erl_crash.dump .DS_Store rickroll.wav .lexical +octopus/recordings/* diff --git a/octopus/CLAUDE.md b/octopus/CLAUDE.md new file mode 100644 index 00000000..43c994c2 --- /dev/null +++ b/octopus/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/octopus/config/config.exs b/octopus/config/config.exs index 232c7878..b9c9f339 100644 --- a/octopus/config/config.exs +++ b/octopus/config/config.exs @@ -32,6 +32,27 @@ config :octopus, :firmware_broadcaster_remote_port, 1337 # Used by Octopus.Osc.Server - Open Sound Control for audio/visual applications config :octopus, :osc_server_port, 8000 +# ============================================================================= +# RECORDING +# ============================================================================= + +# Records the frames the mixer sends to the LED panels into an append-only file +# that can be converted to video. Disabled by default so it has zero overhead +# and can never influence the running installation unless explicitly enabled. +config :octopus, Octopus.Recording, + enabled: false, + output_dir: "recordings", + max_queue: 600, + # Where an auto-started/default recording is written: + # {:file, []} -> local file (default) + # {:remote, host: "10.0.0.5", port: 7000} -> stream to a TCP server + sink: {:file, []}, + # gzip the recording stream (built-in :zlib, no native deps). File targets + # gain a .gz suffix; encoders read .gz transparently. Lower gzip_level is + # cheaper on constrained CPUs (e.g. Raspberry Pi). + compress: false, + gzip_level: 6 + # Network addresses are now configured in Installation modules # See lib/octopus/installation/*.ex files diff --git a/octopus/docs/recording.md b/octopus/docs/recording.md new file mode 100644 index 00000000..51dc4d88 --- /dev/null +++ b/octopus/docs/recording.md @@ -0,0 +1,371 @@ +# Recording & Playback — User Guide + +This guide explains how to record what Octopus sends to the LED panels (and, +optionally, the radar sensor tracking data) and how to turn those recordings +into playable video. + +For the *why* and the internal design, see +[`recording_architecture.md`](./recording_architecture.md). + +--- + +## What you can record + +| Stream | Source | On-disk file | Becomes | +|--------|--------|--------------|---------| +| Panels (animations) | The mixer's outgoing frames | `panels.octorec` | one video per panel + a mixed video | +| Radar (motion) | Radar sensor tracks | `radar.jsonl` | a top-down "scope" video | + +A **session** records both together into one timestamped directory using a +shared clock, so the panel video and the radar scope line up frame-for-frame. + +``` +recordings/ + session-20260712-143000/ + panels.octorec + radar.jsonl +``` + +Recording is **disabled by default** and has zero overhead until you turn it +on. It is designed so it can never crash or slow down the running installation +(see the architecture doc for the guarantees). + +--- + +## Configuration + +`config/config.exs`: + +```elixir +config :octopus, Octopus.Recording, + enabled: false, # auto-start a session on boot + output_dir: "recordings",# base directory for sessions/files + max_queue: 600, # mailbox backlog before frames are dropped + sink: {:file, []} # default sink (see "Streaming to a remote server") +``` + +- `enabled: true` makes the app start a recording session automatically at + boot. Leave it `false` for on-demand recording. +- `output_dir` is relative to the app's working directory unless absolute. +- `max_queue` is a safety valve: if writing can't keep up (slow disk / slow + network) the recorder drops frames instead of growing memory. Dropped frames + just extend the previous frame's on-screen duration in playback. + +--- + +## Recording (runtime control) + +From an `iex` session attached to the app: + +```elixir +# Start a session (panels + radar-if-enabled) into output_dir/session-/ +Octopus.Recording.start() +# => {:ok, %{dir: "recordings/session-...", panels: "file:...", radar: "file:..."}} + +# Start into a specific base directory +Octopus.Recording.start(dir: "/tmp/rec") + +# Panels only (skip radar even if radar is enabled) +Octopus.Recording.start(radar: false) + +# Inspect status +Octopus.Recording.status() +# => %{active: true, dir: "...", panels: %{active: true, written: 1234, dropped: 0, ...}, +# radar: %{active: true, written: 88, dropped: 0, ...}} + +Octopus.Recording.recording?() # => true / false + +# Stop (flushes and closes both files) +Octopus.Recording.stop() +``` + +Notes: + +- `start/1` returns `{:error, :already_recording}` if a session is already + active. Call `stop/0` first. +- Radar is only recorded when `Octopus.Radar.enabled?()` is true (you can force + it off with `radar: false`). If radar is off, only `panels.octorec` is + written. +- `stop/0` is synchronous: once it returns, all frames received before it are + guaranteed to be written and the files are closed. + +### Recording a single stream (advanced) + +The session always writes files. If you want just one stream (e.g. only panels, +or to a custom sink), drive the low-level recorders directly: + +```elixir +Octopus.Recording.PanelRecorder.start_recording(dir: "/tmp/rec") +Octopus.Recording.PanelRecorder.stop_recording() + +Octopus.Recording.RadarRecorder.start_recording(dir: "/tmp/rec") +Octopus.Recording.RadarRecorder.stop_recording() +``` + +--- + +## Streaming to a remote server + +Instead of (or before) writing to disk, a stream can be sent to a TCP server. +The exact same bytes that would go into a file are sent over the socket, so the +receiver can just append them to a `.octorec` / `.jsonl` file and encode later. + +Start a single recorder with the remote sink: + +```elixir +Octopus.Recording.PanelRecorder.start_recording( + sink_mod: Octopus.Recording.Sink.Remote, + sink_opts: [host: "10.0.0.5", port: 7000] +) +``` + +Remote sink options: `:host` (string / charlist / IP tuple), `:port`, +`:connect_timeout` (default 5000 ms), `:send_timeout` (default 2000 ms). + +On the server, capture the stream to a file, then encode it: + +```sh +# netcat +nc -l 7000 > session.octorec + +# or socat (append-safe) +socat TCP-LISTEN:7000,reuseaddr OPEN:session.octorec,creat,append + +mix octopus.recording.encode session.octorec +``` + +> The **session** API (`Octopus.Recording.start/1`) always uses file sinks, +> because both streams share one directory. Remote streaming is a per-recorder +> feature (one stream = one connection). To stream both panels and radar, +> start each recorder on its own port. + +--- + +## Encoding to video + +Encoding is an **offline** step (a mix task) that shells out to `ffmpeg`. It +never runs inside the live app, so it can't affect the installation. + +Requires `ffmpeg` on your `PATH`: + +- macOS: `brew install ffmpeg` +- Debian/Ubuntu: `apt install ffmpeg` + +### Encode a whole session + +```sh +mix octopus.recording.encode recordings/session-20260712-143000 +``` + +Produces, inside the session directory: + +- `panel_00.mp4 … panel_NN.mp4` — one video per panel (native resolution, + upscaled with crisp nearest-neighbour pixels) +- `mixed.mp4` — all panels side by side in panel order (the circular + installation "unrolled" into a horizontal strip) +- `radar.mp4` — the top-down radar scope + +### Encode a single file + +```sh +mix octopus.recording.encode recordings/panels-20260712-143000.octorec +mix octopus.recording.encode recordings/radar-20260712-143000.jsonl +``` + +### Options + +| Option | Applies to | Default | Meaning | +|--------|-----------|---------|---------| +| `--out DIR` | all | dir named after the file (session: the session dir) | output directory | +| `--fps N` | all | `30` | constant output frame rate | +| `--scale N` | panels | `16` | integer nearest-neighbour upscale factor | +| `--size N` | radar | `256` | square scope size in pixels | +| `--no-panels` | panels | — | skip per-panel videos | +| `--no-mixed` | panels | — | skip the mixed video | +| `--ffmpeg PATH` | all | `ffmpeg` | ffmpeg executable | + +Examples: + +```sh +mix octopus.recording.encode rec.octorec --fps 60 --scale 24 --out /tmp/out +mix octopus.recording.encode rec.octorec --no-panels # mixed only +mix octopus.recording.encode session-dir --size 512 # bigger radar scope +``` + +Because both encoders resample to the same `--fps`, the `mixed.mp4` and +`radar.mp4` from one session have the same frame count and duration and can be +played/overlaid in sync. + +--- + +## Typical workflows + +**Local capture → video** + +```elixir +Octopus.Recording.start() +# ... let the animation run ... +Octopus.Recording.stop() # note the returned dir +``` +```sh +mix octopus.recording.encode recordings/session- +``` + +**Remote capture → video (on the server)** + +```elixir +Octopus.Recording.PanelRecorder.start_recording( + sink_mod: Octopus.Recording.Sink.Remote, sink_opts: [host: "server", port: 7000]) +``` +```sh +nc -l 7000 > capture.octorec # on the server +mix octopus.recording.encode capture.octorec +``` + +**Always-on recording in production** + +```elixir +config :octopus, Octopus.Recording, enabled: true +``` +A session auto-starts at boot; find it under `output_dir/session-/`. + +--- + +## Storage sizing + +The panel recording (`.octorec`) writes a **fixed-size record per frame**: + +``` +bytes/frame = 4 (timestamp) + num_panels × panel_width × panel_height × 3 (RGB) +``` + +For the current installation (12 panels of 8×8) that is: + +``` +4 + 12 × 8 × 8 × 3 = 4 + 2304 = 2308 bytes/frame +``` + +(The 26-byte header is one-time and negligible. Grayscale `WFrame`s are stored +as RGB, so they cost the same.) + +Size scales linearly with the frame rate the mixer actually emits — which is +**event-driven, up to ~60 fps**, not fixed: + +``` +MB/hour = bytes/frame × fps × 3600 ÷ 1,000,000 +``` + +| Frame rate | Per second | Per hour (MB) | Per hour (MiB) | +|-----------|-----------|---------------|----------------| +| 60 fps (peak) | ~135 KiB/s | ≈ 499 MB | ≈ 475 MiB | +| 30 fps | ~68 KiB/s | ≈ 249 MB | ≈ 238 MiB | +| 24 fps | ~54 KiB/s | ≈ 199 MB | ≈ 190 MiB | +| 10 fps | ~23 KiB/s | ≈ 83 MB | ≈ 79 MiB | +| ~1 fps (idle) | ~2.3 KiB/s | ≈ 8 MB | ≈ 8 MiB | + +Plan for roughly **0.5 GB/hour at full 60 fps**, ~250 MB/hour at 30 fps, and much +less when animations are mostly static (the mixer emits only ~1 idle frame per +second when nothing changes, and de-duplicated split-part frames don't count). + +Notes: + +- This is the **uncompressed on-disk `.octorec`** size — what you budget disk for + *during* recording. The encoded MP4s are far smaller. +- **Radar (`radar.jsonl`) is tiny** by comparison: a few hundred bytes per frame + at a handful of frames/second — on the order of a few MB/hour. +- Rule of thumb for other geometries: + `(4 + panels × w × h × 3) × fps × 3600` bytes per hour. +- If you run always-on (`enabled: true`), size the disk for the peak rate × + expected session length, or rotate/encode/delete sessions. + +--- + +## Compression + +Recordings can be gzip-compressed on the fly to save disk (and network) space. +Compression uses Erlang's built-in `:zlib`, so there are **no native +dependencies** — it works on a Raspberry Pi out of the box. + +Enable it globally: + +```elixir +config :octopus, Octopus.Recording, + compress: true, + gzip_level: 6 # 0..9; lower = faster / less CPU +``` + +Or per recording: + +```elixir +Octopus.Recording.start(dir: "/tmp/rec", compress: true) +``` + +When enabled: + +- File targets gain a `.gz` suffix (e.g. `panels-.octorec.gz`). +- The remote sink streams gzip bytes too (compose transparently). +- The encoder reads `.gz` recordings automatically — no extra flags: + + ```sh + mix octopus.recording.encode recordings/panels-20260712-143000.octorec.gz + ``` + +- Standard gzip, so `gunzip`/`zcat` work on the files as well. + +### How much it saves + +Highly content-dependent (see the measured numbers in +[`recording_architecture.md`](./recording_architecture.md#compression)): + +| Content | Typical result | +|---------|----------------| +| Busy full-colour animation | ~1.3× smaller (~20–35%) | +| Normal scenes (motion + dark areas) | ~2–5× smaller | +| Text / sparse / mostly dark | 10× – 1000× smaller | +| Full-frame random noise (unrealistic) | ~1× (no gain) | + +### Raspberry Pi notes + +- CPU cost is small next to the ~135 KiB/s data rate; if the Pi is busy, lower + `gzip_level` (e.g. `1`–`4`) for cheaper compression at a slightly worse ratio. +- No cross-compiled dependencies are involved (`:zlib` ships with OTP). +- A gzip stream is only finalized when the recording is **stopped**; the sink + also flushes periodically so a hard power-loss loses at most a few seconds. + Always `stop/0` cleanly when you can. + +--- + +## Troubleshooting + +- **`ffmpeg was not found`** — install ffmpeg or pass `--ffmpeg /path/to/ffmpeg`. +- **`Recording contains no frames`** — nothing was recorded (e.g. the app wasn't + producing frames, or radar wasn't enabled for `radar.jsonl`). +- **`status` shows `dropped > 0`** — the sink couldn't keep up (slow disk or slow + remote). Lower the frame rate at the source or check the sink; the recording + is still valid, just missing some frames. +- **`radar.jsonl` is empty / only metadata** — radar wasn't enabled + (`Octopus.Radar.enabled?()` was false) or no targets were tracked. +- **Disk usage** — panel frames are uncompressed until encoded (~0.5 GB/hour at + 60 fps for a 12×8×8 installation). Enable [Compression](#compression) to cut + this substantially, and see [Storage sizing](#storage-sizing). Keep sessions + bounded and encode/delete when done. +- **`Could not decompress ...`** — a `.gz` recording is truncated (e.g. the app + was hard-killed before `stop/0`) or isn't actually gzip. + +--- + +## File layout reference + +``` +recordings/ + session-/ + panels.octorec # binary panel frames (see architecture doc) + radar.jsonl # one JSON object per radar frame + # after encoding: + panel_00.mp4 ... # one per panel + mixed.mp4 # all panels, side by side + radar.mp4 # top-down scope +``` + +Standalone (low-level recorder) files are named +`panels-.octorec` and `radar-.jsonl` in `output_dir`. diff --git a/octopus/docs/recording_architecture.md b/octopus/docs/recording_architecture.md new file mode 100644 index 00000000..7d50f88e --- /dev/null +++ b/octopus/docs/recording_architecture.md @@ -0,0 +1,372 @@ +# Recording Subsystem — Design & Architecture + +This document describes the recording subsystem: what it does, why it is built +the way it is, and the important details for anyone extending or debugging it. + +For day-to-day usage see [`recording.md`](./recording.md). + +--- + +## Goals & constraints + +The installation produces pixel animations that are sent to LED panels over +UDP/protobuf, and radar sensors track motion that influences those animations. +We want to **record the animations sent to the panels** and **record the radar +data**, both for later **visualization** (playable video), not for +packet/firmware debugging. + +Hard requirements that shaped every decision: + +1. **Never crash or influence the running application.** Recording is + observational. A recording failure must degrade the recording only — never + the mixer, the broadcaster, the apps, or the sensors. +2. **Visualization, not forensics.** We capture the logical, displayable frame + (not the hardware wire format), and we don't record audio. +3. **A movie at the end.** Recordings must convert to standard video. Panels → + one video per panel plus a mixed video; radar → a top-down scope video. +4. **Append-only files, no database.** Simple, cheap to write, easy to convert. +5. **Also streamable to a remote server**, not only to a local file. +6. **The number of input devices is not fixed** — panel count is read from the + installation at record time and stored in the recording. + +--- + +## Data flow & tap points + +### Panels (animations → LEDs) + +``` +Apps ──► Mixer ──► (compose / transition / mask) ──► RGBFrame / WFrame + │ + ├─ PubSub topic "mixer": {:mixer, {:frame, frame}} ◄── PanelRecorder taps here + │ + └─ Untangle (hardware wire order) ─► Protobuf.split_and_encode ─► Broadcaster ─► UDP ─► panels +``` + +The recorder subscribes to the mixer's `"mixer"` PubSub topic and records the +`{:mixer, {:frame, frame}}` messages. This is deliberately **upstream of +`Untangle` and `split_and_encode`**, so the recorded frame is: + +- the **logical** frame (not the hardware-scrambled pixel order), and +- **full resolution and un-split** (not the two UDP packets). + +This is the same feed the simulators consume, so it is a proven, side-effect-free +tap. The pixel byte layout the mixer produces (see `Octopus.Mixer.canvas_to_frame/4`) +is **panel-major, then row-major within each panel**: + +``` +for panel <- 0..num_panels-1, y <- 0..panel_height-1, x <- 0..panel_width-1 -> {r,g,b} +``` + +which is exactly why splitting into per-panel videos is a trivial fixed-stride +slice. + +Two quirks handled by the recorder: + +- The `{:mixer, {:frame, ...}}` message fires **once per UDP split part**, so + each frame arrives twice back-to-back. The recorder de-duplicates + byte-identical frames that arrive within 5 ms of each other. +- Frames are **event-timed** (emitted when the mixer renders, plus a ~1 s idle + frame), not at a fixed rate. Timing is preserved as per-frame timestamps and + resampled to a constant rate at encode time. + +### Radar (sensors → animations) + +``` +Radar sensors ──► Octopus.Radar.Sensor ──► Transform (local → global frame) + │ + └─ PubSub topic "radar:hlk6001": {:radar_frame, device_id, %Frame{}} ◄── RadarRecorder taps here +``` + +Track positions are already mapped into the **installation global frame** +(meters, origin at the installation center) by `Octopus.Radar.Transform` before +publishing, so all sensors share one coordinate system and merge naturally into +a single scope. + +> The firmware proximity sensors (`ProximityEvent` via the broadcaster) are +> intentionally **not** recorded — only the radar layer is relevant. + +--- + +## Safety design (how "never influence the app" is guaranteed) + +- **Passive subscriber.** Recorders only ever *receive* PubSub broadcasts. + `Phoenix.PubSub.broadcast` does not link and does not wait for subscribers, so + the mixer/sensors never block on, or fail because of, a recorder. +- **Subscribe only while active.** When not recording, nothing is subscribed — + zero message traffic, zero overhead. Disabled = truly off. +- **No calls into the recorder from the hot path.** The producers never call the + recorder; communication is one-way via PubSub. +- **Bounded mailbox with drop-on-overload.** The only risk a passive subscriber + poses is an unbounded mailbox if it can't keep up. Each recorder checks its own + `message_queue_len` and drops frames once it exceeds `max_queue`, so memory + can't balloon. Dropped frames only shorten the recording. +- **Crash isolation.** Every incoming frame is processed inside + `try/rescue/catch`; a bad frame or sink error degrades to a dropped frame (or a + clean stop on sink error), never a crash. The whole subsystem lives under + `Octopus.Recording.Supervisor` (`:one_for_one`), so even an unexpected crash + restarts only the recorder. +- **Bounded network I/O.** The remote sink uses a connect timeout (unreachable + server fails the *start*, doesn't hang) and a per-write send timeout with + `send_timeout_close`, so a stalled server produces a bounded error rather than + an indefinite block. Combined with drop-on-overload, a slow network only + degrades the recording. +- **Additive-only touch to the app.** The single change to existing runtime code + is a new `Octopus.Mixer.unsubscribe/0` (mirrors `subscribe/0`) — no behavioral + change — plus adding `Octopus.Recording.Supervisor` to the supervision tree. + +--- + +## Shared timeline & alignment + +Both recorders stamp each frame with `offset_ms`, measured from a **monotonic** +origin captured at session start (`System.monotonic_time(:millisecond)`), and +store the wall-clock start (`System.system_time(:millisecond)`) in their header. +Monotonic time is used so frame timing is stable and immune to NTP jumps. + +`Octopus.Recording.Session` starts the panel and radar recorders with the **same +`start_mono_ms` and `started_at_ms`**, so their offsets share one zero. That's +what lets the encoders (which resample to the same fps) emit `mixed.mp4` and +`radar.mp4` with matching frame counts and durations, aligned frame-for-frame. + +The radar recorder uses `Frame.received_at` (already monotonic ms) for its +timestamps, so the recorded time reflects when the frame actually arrived, not +when the recorder happened to process it. + +--- + +## File formats + +### `.octorec` (panels) — `Octopus.Recording.Format` + +Append-only binary. All integers big-endian. Grayscale (`WFrame`) frames are +normalized to RGB (`r = g = b = w`) on write so the stream is uniform and the +encoder never branches on frame type. + +Header (26 bytes, once): + +| Field | Type | Notes | +|-------|------|-------| +| magic | 8 bytes | `"OCTOREC1"` | +| version | u8 | format version (1) | +| kind | u8 | `0` = rgb (only value emitted) | +| num_panels | u16 | from the installation at record time | +| panel_width | u16 | | +| panel_height | u16 | | +| reserved | u16 | `0` | +| started_at_ms | u64 | wall-clock ms at start | + +Record (`4 + frame_bytes` bytes, repeated): + +| Field | Type | Notes | +|-------|------|-------| +| offset_ms | u32 | monotonic ms since start | +| pixels | `frame_bytes` | RGB, panel-major / row-major | + +where `frame_bytes = num_panels * panel_width * panel_height * 3`. Panel `n` +occupies the slice `[n * panel_bytes, (n+1) * panel_bytes)` with +`panel_bytes = panel_width * panel_height * 3`, and each slice is already a +ready-to-use RGB image. + +**Why binary + fixed size:** at up to 60 fps this is the cheapest possible write +(append raw bytes, no per-frame parsing), and fixed geometry means the encoder +can slice panels with pure offset math. A database row per frame would be far too +much overhead; the file is the natural container. + +### `radar.jsonl` (radar) — `Octopus.Recording.RadarFormat` + +Line-delimited JSON. Radar is low-volume and variable-length (a frame has a +variable number of tracks), so a text format is the pragmatic choice: easy to +inspect, append, and parse. + +Metadata line (first line): + +```json +{"v": 1, "started_at_ms": 1700000000000, "world_radius_m": 8.0} +``` + +Frame line (one per radar frame): + +```json +{"t": 123, "dev": 1, "n": 42, + "tracks": [{"id": 7, "x": 1.2, "y": -0.5, "z": 2.0, "vx": 0.1, "vy": 0.0, "vz": 0.0}]} +``` + +`t` is `offset_ms` (shared timeline), `dev` the sensor device id, `n` the device +frame number, and track coordinates are in the global frame (meters, m/s). + +--- + +## Module structure + +``` +Octopus.Recording facade: config + start/stop/status (session-level) +Octopus.Recording.Supervisor isolated :one_for_one subtree + ├─ Octopus.Recording.PanelRecorder GenServer: taps mixer, writes .octorec + ├─ Octopus.Recording.RadarRecorder GenServer: taps radar, writes .jsonl + └─ Octopus.Recording.Session GenServer: shared clock + session dir; owns boot auto-start + +Octopus.Recording.Format .octorec encode/parse (pure) +Octopus.Recording.RadarFormat radar JSONL encode/parse (pure) + +Octopus.Recording.Sink behaviour: open/1, write/2, close/1, describe/1 +Octopus.Recording.Sink.File local file (:raw + :delayed_write) +Octopus.Recording.Sink.Remote TCP stream (bounded connect/send timeouts) + +Octopus.Recording.Encoder .octorec → per-panel + mixed video (ffmpeg) +Octopus.Recording.RadarEncoder .jsonl → top-down scope video (ffmpeg) +Mix.Tasks.Octopus.Recording.Encode offline CLI; accepts file or session dir +``` + +Supervisor child order matters: the recorders start before `Session` so the +session can drive (and auto-start) them once they're up. + +### The `Sink` seam + +Recorders write opaque `iodata` through a `Sink`. This is the extension point +that made remote streaming a small, self-contained addition: `Sink.File` and +`Sink.Remote` implement the same 4-callback behaviour, and the recorder is +agnostic to which is in use. New transports (e.g. an HTTP uploader) only need to +implement the behaviour. + +--- + +## Encoding design + +Both encoders are **offline** (a mix task), never part of the running app, so +`ffmpeg` and its CPU cost never touch the installation. + +Shared approach: + +1. **Resample** the event-timed records onto a constant fps using + hold-last-frame semantics (step time in `1000/fps` ms increments; emit the + most recent frame at or before each tick). This converts irregular timing into + a standard constant-frame-rate movie. +2. Write raw `rgb24` frames to a temp file and hand them to `ffmpeg` + (`-f rawvideo -pix_fmt rgb24 ...`), producing H.264 `yuv420p` MP4s. Temp files + are always cleaned up. + +**Panels** (`Encoder`): + +- Per-panel video: slice panel `n`'s block each frame → native `pw x ph` image, + upscaled by `--scale` with nearest-neighbour (crisp pixels). +- Mixed video: transpose the panel-major buffer into a row-major **strip** + (`num_panels * panel_width` × `panel_height`) — the circular installation + unrolled left-to-right in panel order. + +**Radar** (`RadarEncoder`): + +- At each tick, the scene is the **union of the most recent frame from each + sensor** (grouped by `dev`), so all sensors appear together. +- Each track is drawn as a colour-coded dot (colour by track id) on a top-down + view: origin centered, `+x` right, `+y` up, with a boundary ring and centre + marker. World extent defaults to the recording's `world_radius_m`. + +Pure helpers (`resample/2`, `panel_frame/3`, `strip_frame/2`, `scope_frames/2`, +`world_to_px/4`, `render/3`) are separated from the `ffmpeg` shell-out so they +can be unit-tested without `ffmpeg`. + +--- + +## Configuration reference + +```elixir +config :octopus, Octopus.Recording, + enabled: false, # Session auto-starts on boot when true + output_dir: "recordings", # base dir for sessions and standalone files + max_queue: 600, # per-recorder mailbox backlog → drop threshold + sink: {:file, []} # default sink for the low-level recorders +``` + +`sink` accepts `{:file, opts}` (opts may set `:dir` or a fixed `:path`) or +`{:remote, host:, port:, ...}`. Note the **Session always uses file sinks** +(both streams share a directory); the `sink` spec drives the low-level recorders +when started directly without an explicit `:sink_mod`. + +--- + +## Testing + +- Pure format/encoder logic is unit-tested (header round-trips, W→RGB + normalization, resampling, panel slicing, strip transpose, radar scene merge, + `world_to_px`). +- Recorders are tested end-to-end by broadcasting synthetic `{:mixer, {:frame, + ...}}` / `{:radar_frame, ...}` messages and asserting the resulting files. + Because a live mixer may emit blank idle frames onto the same topic, panel + assertions filter to the test's own distinctive frames. +- The remote sink is tested against an in-test TCP listener (bytes arrive + verbatim; unreachable server errors cleanly). +- `ffmpeg` end-to-end encode tests are tagged `:ffmpeg` and run with + `mix test --include ffmpeg`; they no-op when `ffmpeg` is absent. + +--- + +## Compression + +Recordings can be gzip-compressed via a composable sink, +`Octopus.Recording.Sink.Gzip`, which wraps an inner sink (file or remote) and +compresses the stream with Erlang's built-in `:zlib`. This is deliberately +**dependency-free** so it runs on a Raspberry Pi without any native/cross- +compiled libraries (`zstd` was rejected for exactly this reason). Enable with +`compress: true` (config or per `start/1`); `gzip_level` (0..9) trades ratio for +CPU. + +Design points: + +- **Composability.** `Sink.Gzip.wrap/4` turns a resolved `{sink_mod, sink_opts}` + into a gzip-wrapped one, adding a `.gz` suffix for file targets. It works + identically over the file and remote sinks. +- **Crash safety.** A gzip stream is only fully valid once finalized on + `close/1` (called on stop and on normal supervised shutdown). To bound loss + from a hard VM crash, the stream is `:sync`-flushed every `flush_every` writes + (default 200); a sync flush keeps the compression dictionary, so the ratio + impact is negligible. +- **Transparent decode.** The encoders detect a `.gz` extension and + `:zlib.gunzip` the file before parsing, so `mix octopus.recording.encode` + needs no extra flags. + +### Measured savings + +Measured on 900-frame (~30 s) recordings for a 12x8x8 installation +(2,077,226 bytes raw), gzip level 6: + +| Content | gzip savings | with temporal delta (XOR)+gzip | +|---------|--------------|--------------------------------| +| Busy full-colour animation | ~20–35% (1.3–1.5×) | ~59% (2.5×) | +| Text / sparse / mostly dark | 99%+ (100–800×) | 99.98% (6000×+) | +| Full-frame random noise | ~0% | ~0% | + +The big lever is **temporal redundancy** (consecutive frames are nearly +identical). Generic gzip only partly exploits it because its 32 KB window spans +only ~14 frames; a temporal delta pre-pass (XOR each frame against the previous) +turns most bytes to zero and roughly doubles the ratio on busy content. A delta +pre-pass is a possible future addition to the format (it would need a format +flag and reversal in the encoder); it is **not** implemented today. + +--- + +## Known limitations & gotchas + +- **`ffmpeg` is required for encoding** (not for recording). +- **Compressed recordings are only finalized on a clean stop.** A hard VM crash + can truncate the trailing gzip block; periodic sync-flushing bounds the loss + but the very tail may be unreadable by one-shot `:zlib.gunzip`. +- **Radar recording requires the radar layer to be enabled** + (`Octopus.Radar.enabled?()`); otherwise `radar.jsonl` contains only metadata. +- **Sessions are file-only.** Streaming both panels and radar to a remote server + means starting each low-level recorder on its own connection/port. +- **Disk usage.** `.octorec` is uncompressed; bound session length and encode/ + delete when done. (~hundreds of KB/s for a full installation at 60 fps.) +- **Frame de-dup window** is 5 ms; two genuinely-distinct identical frames closer + than that would be collapsed (visually irrelevant — timing is preserved). +- **The mixed layout is a horizontal strip** (unrolled circle). A true ring + layout would be a rendering change in `Encoder`/`RadarEncoder`. + +--- + +## Possible next steps + +- A combined overlay clip (radar scope composited beside/over `mixed.mp4`) as a + single deliverable. +- An HTTP/object-storage `Sink` implementation. +- A player that streams a session back into the simulator. diff --git a/octopus/lib/mix/tasks/octopus.recording.encode.ex b/octopus/lib/mix/tasks/octopus.recording.encode.ex new file mode 100644 index 00000000..93644ef6 --- /dev/null +++ b/octopus/lib/mix/tasks/octopus.recording.encode.ex @@ -0,0 +1,89 @@ +defmodule Mix.Tasks.Octopus.Recording.Encode do + @shortdoc "Encode a .octorec panel recording into per-panel and mixed videos" + + @moduledoc """ + Convert a panel recording produced by `Octopus.Recording` into video. + + mix octopus.recording.encode RECORDING.octorec [options] + + Gzip-compressed recordings (`.octorec.gz`) are accepted and decompressed + automatically. + + By default this writes one video per panel plus a single mixed video (all + panels laid out side by side) into a directory named after the recording. + + ## Options + + * `--out DIR` - output directory (default: sibling dir named after the file) + * `--fps N` - constant output frame rate (default: 30) + * `--scale N` - integer nearest-neighbour upscale factor (default: 16) + * `--no-panels` - skip per-panel videos + * `--no-mixed` - skip the mixed video + * `--ffmpeg PATH` - ffmpeg executable to use (default: "ffmpeg") + + ## Examples + + mix octopus.recording.encode recordings/panels-20260712-143000.octorec + mix octopus.recording.encode rec.octorec --fps 60 --scale 24 --out /tmp/out + mix octopus.recording.encode rec.octorec --no-panels + """ + + use Mix.Task + + alias Octopus.Recording.Encoder + + @switches [ + out: :string, + fps: :integer, + scale: :integer, + panels: :boolean, + mixed: :boolean, + ffmpeg: :string + ] + + @impl Mix.Task + def run(argv) do + {opts, args} = OptionParser.parse!(argv, strict: @switches) + + input = + case args do + [input | _] -> + input + + [] -> + Mix.raise("Missing recording file. Usage: mix octopus.recording.encode FILE.octorec") + end + + unless File.regular?(input) do + Mix.raise("Recording file not found: #{input}") + end + + Mix.shell().info("Encoding #{input} ...") + + case Encoder.encode(input, opts) do + {:ok, outputs} -> + Mix.shell().info("Wrote #{length(outputs)} file(s):") + Enum.each(outputs, &Mix.shell().info(" #{&1}")) + + {:error, :ffmpeg_not_found} -> + Mix.raise(""" + ffmpeg was not found on your PATH. + + Install it (e.g. `brew install ffmpeg` on macOS, `apt install ffmpeg` on + Debian/Ubuntu) or pass an explicit path with --ffmpeg. + """) + + {:error, :empty_recording} -> + Mix.raise("Recording contains no frames: #{input}") + + {:error, :invalid_header} -> + Mix.raise("Not a valid .octorec recording: #{input}") + + {:error, {:gunzip_failed, _}} -> + Mix.raise("Could not decompress #{input} (truncated or not a gzip file?)") + + {:error, reason} -> + Mix.raise("Failed to encode recording: #{inspect(reason)}") + end + end +end diff --git a/octopus/lib/octopus/application.ex b/octopus/lib/octopus/application.ex index 72e83c8a..e2ee018e 100644 --- a/octopus/lib/octopus/application.ex +++ b/octopus/lib/octopus/application.ex @@ -48,6 +48,7 @@ defmodule Octopus.Application do Octopus.Mixer, Octopus.ButtonServer, Octopus.Sunlight, + Octopus.Recording.Supervisor, # WebApp {Finch, name: Octopus.Finch}, diff --git a/octopus/lib/octopus/mixer.ex b/octopus/lib/octopus/mixer.ex index 784aa6df..6972afa1 100644 --- a/octopus/lib/octopus/mixer.ex +++ b/octopus/lib/octopus/mixer.ex @@ -118,10 +118,16 @@ defmodule Octopus.Mixer do `duration_ms` is the total fade time (half out, half in). When `duration_ms` is 0 or transitions are disabled, `on_black` runs immediately. """ - def run_transition(duration_ms, on_black) when is_integer(duration_ms) and duration_ms >= 0 and is_function(on_black, 0) do + def run_transition(duration_ms, on_black) + when is_integer(duration_ms) and duration_ms >= 0 and is_function(on_black, 0) do GenServer.cast(__MODULE__, {:run_transition, duration_ms, on_black}) end + @doc "Unsubscribes the calling process from the mixer topic." + def unsubscribe do + Phoenix.PubSub.unsubscribe(Octopus.PubSub, @pubsub_topic) + end + def init(:ok) do # Subscribe to app events AppManager.subscribe() @@ -313,7 +319,8 @@ defmodule Octopus.Mixer do {state, :noop} state.transition == nil -> - {%State{state | transition: {:out, @transition_duration, selected_app}}, :start_transition} + {%State{state | transition: {:out, @transition_duration, selected_app}}, + :start_transition} match?({:out, _, _}, state.transition) -> # Retarget an in-flight fade-out (e.g. stop old app then immediately select new one). @@ -402,7 +409,10 @@ defmodule Octopus.Mixer do {:noreply, state} end - def handle_info(:transition, %State{transition: {:out, time, {:callback, on_black, half}}} = state) + def handle_info( + :transition, + %State{transition: {:out, time, {:callback, on_black, half}}} = state + ) when time <= 0 do on_black.() @@ -413,7 +423,10 @@ defmodule Octopus.Mixer do {:noreply, state} end - def handle_info(:transition, %State{transition: {:out, time, {:callback, on_black, half}}} = state) do + def handle_info( + :transition, + %State{transition: {:out, time, {:callback, on_black, half}}} = state + ) do state = %State{ state | transition: {:out, time - @transition_frame_time, {:callback, on_black, half}} diff --git a/octopus/lib/octopus/recording.ex b/octopus/lib/octopus/recording.ex new file mode 100644 index 00000000..8e64fcd2 --- /dev/null +++ b/octopus/lib/octopus/recording.ex @@ -0,0 +1,97 @@ +defmodule Octopus.Recording do + @moduledoc """ + Public facade for recording the animations sent to the LED panels. + + Recording captures the mixer's outgoing frames into an append-only file (or, + later, a remote stream) that can be converted into a video for playback. See + `Octopus.Recording.PanelRecorder` for the recorder itself and + `Octopus.Recording.Format` for the on-disk format. + + ## Configuration + + config :octopus, Octopus.Recording, + enabled: false, # auto-start a recording on boot + output_dir: "recordings",# where generated recording files are written + max_queue: 600, # mailbox backlog before frames are dropped + sink: {:file, []}, # default sink; see below + compress: false, # gzip the recording stream (see below) + gzip_level: 6 # zlib level 0..9 when compress: true + + The `:sink` spec selects where an auto-started (or default) recording is + written: + + * `{:file, opts}` - append to a local file (`opts` may set `:dir` or a + fixed `:path`). This is the default. + * `{:remote, opts}` - stream to a TCP server; `opts` requires `:host` and + `:port` (see `Octopus.Recording.Sink.Remote`). + + When `:compress` is true the stream is gzip-compressed (via built-in `:zlib`, + no native deps) before hitting the sink; file targets gain a `.gz` suffix. The + encoders read `.gz` recordings transparently. Lower `:gzip_level` values are + cheaper on constrained CPUs (e.g. a Raspberry Pi). + + ## Runtime control + + Octopus.Recording.start() # start recording using the default sink + Octopus.Recording.start(dir: "/tmp/rec") + Octopus.Recording.start(sink_mod: Octopus.Recording.Sink.Remote, + sink_opts: [host: "10.0.0.5", port: 7000]) + Octopus.Recording.status() + Octopus.Recording.stop() + """ + + alias Octopus.Recording.PanelRecorder + + @default_output_dir "recordings" + @default_max_queue 600 + + @doc "The recording configuration keyword list." + @spec config() :: keyword() + def config, do: Application.get_env(:octopus, __MODULE__, []) + + @doc "Whether recording should auto-start on boot." + @spec enabled?() :: boolean() + def enabled?, do: config()[:enabled] == true + + @doc "Directory generated recording files are written to." + @spec output_dir() :: String.t() + def output_dir, do: config()[:output_dir] || @default_output_dir + + @doc "Mailbox backlog at which the recorder starts dropping frames." + @spec max_queue() :: pos_integer() + def max_queue, do: config()[:max_queue] || @default_max_queue + + @doc """ + The configured default sink spec, used for auto-start and when `start/1` is + called without an explicit `:sink_mod`. Defaults to `{:file, []}`. + """ + @spec sink_spec() :: {:file, keyword()} | {:remote, keyword()} + def sink_spec, do: config()[:sink] || {:file, []} + + @doc "Whether recordings should be gzip-compressed. Defaults to false." + @spec compress?() :: boolean() + def compress?, do: config()[:compress] == true + + @doc "zlib compression level (0..9) used when `compress?/0` is true. Defaults to 6." + @spec gzip_level() :: 0..9 + def gzip_level, do: config()[:gzip_level] || 6 + + @doc """ + Start recording. See `Octopus.Recording.PanelRecorder.start_recording/1` for + options. Returns `{:ok, target}` or `{:error, reason}`. + """ + @spec start(keyword()) :: {:ok, String.t()} | {:error, term()} + def start(opts \\ []), do: PanelRecorder.start_recording(opts) + + @doc "Stop the current recording." + @spec stop() :: :ok | {:error, :not_recording} + def stop, do: PanelRecorder.stop_recording() + + @doc "Return the recorder status map." + @spec status() :: map() + def status, do: PanelRecorder.status() + + @doc "Whether a recording is currently active." + @spec recording?() :: boolean() + def recording?, do: match?(%{active: true}, status()) +end diff --git a/octopus/lib/octopus/recording/encoder.ex b/octopus/lib/octopus/recording/encoder.ex new file mode 100644 index 00000000..7ef83f5f --- /dev/null +++ b/octopus/lib/octopus/recording/encoder.ex @@ -0,0 +1,268 @@ +defmodule Octopus.Recording.Encoder do + @moduledoc """ + Converts a `.octorec` panel recording (see `Octopus.Recording.Format`) into + playable video using `ffmpeg`. + + Two kinds of output are produced: + + * **Per-panel videos** — one video per LED panel (`panel_00.mp4`, ...), + each the native `panel_width x panel_height` resolution upscaled with + nearest-neighbour so pixels stay crisp. + * **A mixed video** (`mixed.mp4`) — all panels laid out side by side in + panel order (the circular installation "unrolled" into a horizontal + strip), i.e. width `num_panels * panel_width`, height `panel_height`. + + Recordings are event-timed (frames are written whenever the mixer emits one, + not at a fixed rate), so records are resampled onto a constant frame rate + using hold-last-frame semantics before encoding. + + The pixel/resampling helpers (`resample/2`, `panel_frame/3`, `strip_frame/2`) + are pure and independently testable; only `encode/2` shells out to `ffmpeg`. + """ + + require Logger + + alias Octopus.Recording.Format + + @default_fps 30 + @default_scale 16 + + @type geometry :: + {num_panels :: pos_integer(), panel_width :: pos_integer(), + panel_height :: pos_integer()} + + @doc """ + Encode `input_path` (a `.octorec` file) into videos. + + Options: + + * `:out` - output directory (default: sibling directory named after the + recording file). + * `:fps` - constant output frame rate (default `#{@default_fps}`). + * `:scale` - integer upscale factor, nearest-neighbour (default `#{@default_scale}`). + * `:panels` - emit per-panel videos (default `true`). + * `:mixed` - emit the mixed strip video (default `true`). + * `:ffmpeg` - ffmpeg executable (default `"ffmpeg"`). + + Returns `{:ok, [output_path]}` or `{:error, reason}`. + """ + @spec encode(String.t(), keyword()) :: {:ok, [String.t()]} | {:error, term()} + def encode(input_path, opts \\ []) do + fps = Keyword.get(opts, :fps, @default_fps) + scale = Keyword.get(opts, :scale, @default_scale) + ffmpeg = Keyword.get(opts, :ffmpeg, "ffmpeg") + do_panels? = Keyword.get(opts, :panels, true) + do_mixed? = Keyword.get(opts, :mixed, true) + out_dir = Keyword.get(opts, :out) || default_out_dir(input_path) + + with :ok <- ensure_ffmpeg(ffmpeg), + {:ok, binary} <- read_recording(input_path), + {:ok, header, records} <- Format.parse(binary), + :ok <- ensure_records(records) do + geom = {header.num_panels, header.panel_width, header.panel_height} + frames = resample(records, fps) + + File.mkdir_p!(out_dir) + raw_dir = Path.join(System.tmp_dir!(), "octorec-raw-#{System.unique_integer([:positive])}") + File.mkdir_p!(raw_dir) + + try do + outputs = + [] + |> maybe_encode_panels(do_panels?, frames, geom, raw_dir, out_dir, fps, scale, ffmpeg) + |> maybe_encode_mixed(do_mixed?, frames, geom, raw_dir, out_dir, fps, scale, ffmpeg) + + {:ok, Enum.reverse(outputs)} + after + File.rm_rf(raw_dir) + end + end + end + + @doc "Whether the given ffmpeg executable is available on the system." + @spec ffmpeg_available?(String.t()) :: boolean() + def ffmpeg_available?(ffmpeg \\ "ffmpeg") do + System.find_executable(ffmpeg) != nil + end + + # Read a recording, transparently decompressing a .gz file. + defp read_recording(path) do + with {:ok, bin} <- File.read(path) do + if String.ends_with?(path, ".gz") do + try do + {:ok, :zlib.gunzip(bin)} + rescue + error -> {:error, {:gunzip_failed, error}} + end + else + {:ok, bin} + end + end + end + + ## Pure pixel / timing helpers (testable without ffmpeg) + + @doc """ + Resample event-timed records onto a constant `fps`, holding the most recent + frame for each output tick. Returns a list of frame pixel binaries. + """ + @spec resample([{non_neg_integer(), binary()}], pos_integer()) :: [binary()] + def resample([], _fps), do: [] + + def resample(records, fps) when is_integer(fps) and fps > 0 do + dt = max(div(1000, fps), 1) + {last_offset, _} = List.last(records) + + times = + if last_offset <= 0 do + [0] + else + Enum.to_list(0..last_offset//dt) + end + + {frames, _} = + Enum.map_reduce(times, records, fn t, recs -> + recs = advance(recs, t) + {current(recs), recs} + end) + + frames + end + + # Advance to the last record whose offset is <= t (records are ascending). + defp advance([{_o1, _d1}, {o2, _d2} = next | rest], t) when o2 <= t, + do: advance([next | rest], t) + + defp advance(recs, _t), do: recs + + defp current([{_o, data} | _]), do: data + + @doc """ + Extract panel `panel_index`'s pixel block from a full frame. The result is a + ready-to-use `panel_width x panel_height` RGB image (row-major). + """ + @spec panel_frame(binary(), non_neg_integer(), geometry()) :: binary() + def panel_frame(data, panel_index, {_n, pw, ph}) do + block = pw * ph * 3 + binary_part(data, panel_index * block, block) + end + + @doc """ + Build the mixed "strip" frame: all panels concatenated horizontally in panel + order, producing a `(num_panels * panel_width) x panel_height` RGB image. + """ + @spec strip_frame(binary(), geometry()) :: binary() + def strip_frame(data, {n, pw, ph}) do + row_bytes = pw * 3 + + for y <- 0..(ph - 1), p <- 0..(n - 1), into: <<>> do + offset = (p * ph + y) * pw * 3 + binary_part(data, offset, row_bytes) + end + end + + ## Encoding steps + + defp maybe_encode_panels(outputs, false, _frames, _geom, _raw, _out, _fps, _scale, _ffmpeg), + do: outputs + + defp maybe_encode_panels( + outputs, + true, + frames, + {n, pw, ph} = geom, + raw_dir, + out_dir, + fps, + scale, + ffmpeg + ) do + pad = n |> Kernel.-(1) |> max(1) |> Integer.to_string() |> String.length() + + Enum.reduce(0..(n - 1), outputs, fn p, acc -> + raw = Path.join(raw_dir, "panel_#{p}.raw") + write_raw(raw, frames, &panel_frame(&1, p, geom)) + + out = Path.join(out_dir, "panel_#{String.pad_leading(Integer.to_string(p), pad, "0")}.mp4") + :ok = run_ffmpeg(ffmpeg, raw, pw, ph, fps, scale, out) + [out | acc] + end) + end + + defp maybe_encode_mixed(outputs, false, _frames, _geom, _raw, _out, _fps, _scale, _ffmpeg), + do: outputs + + defp maybe_encode_mixed( + outputs, + true, + frames, + {n, pw, ph} = geom, + raw_dir, + out_dir, + fps, + scale, + ffmpeg + ) do + raw = Path.join(raw_dir, "mixed.raw") + write_raw(raw, frames, &strip_frame(&1, geom)) + + out = Path.join(out_dir, "mixed.mp4") + :ok = run_ffmpeg(ffmpeg, raw, n * pw, ph, fps, scale, out) + [out | outputs] + end + + defp write_raw(path, frames, transform) do + File.open!(path, [:write, :raw, :binary], fn io -> + Enum.each(frames, fn data -> IO.binwrite(io, transform.(data)) end) + end) + end + + defp run_ffmpeg(ffmpeg, raw, width, height, fps, scale, out) do + args = [ + "-y", + "-f", + "rawvideo", + "-pixel_format", + "rgb24", + "-video_size", + "#{width}x#{height}", + "-framerate", + "#{fps}", + "-i", + raw, + "-vf", + "scale=iw*#{scale}:ih*#{scale}:flags=neighbor,pad=ceil(iw/2)*2:ceil(ih/2)*2", + "-pix_fmt", + "yuv420p", + out + ] + + case System.cmd(ffmpeg, args, stderr_to_stdout: true) do + {_output, 0} -> + :ok + + {output, code} -> + Logger.error("[recording] ffmpeg failed (#{code}): #{output}") + raise "ffmpeg failed with exit code #{code} for #{out}" + end + end + + ## Helpers + + defp ensure_ffmpeg(ffmpeg) do + if ffmpeg_available?(ffmpeg) do + :ok + else + {:error, :ffmpeg_not_found} + end + end + + defp ensure_records([]), do: {:error, :empty_recording} + defp ensure_records(_records), do: :ok + + defp default_out_dir(input_path) do + dir = Path.dirname(input_path) + name = Path.basename(input_path, Path.extname(input_path)) + Path.join(dir, name) + end +end diff --git a/octopus/lib/octopus/recording/format.ex b/octopus/lib/octopus/recording/format.ex new file mode 100644 index 00000000..7efd4495 --- /dev/null +++ b/octopus/lib/octopus/recording/format.ex @@ -0,0 +1,174 @@ +defmodule Octopus.Recording.Format do + @moduledoc """ + On-disk (and on-the-wire) container format for panel recordings. + + A recording is an append-only stream: a single fixed-size header followed + by any number of fixed-size frame records. Because the geometry is captured + once in the header, every record is just a monotonic timestamp plus the raw + RGB pixel bytes for the whole installation. + + Pixel bytes use the exact same layout the mixer produces for a frame + (`Octopus.Mixer.canvas_to_frame/4`): **panel-major, then row-major within + each panel**. For `num_panels` panels of `panel_width x panel_height`, one + frame is `num_panels * panel_width * panel_height * 3` bytes. Panel `n` + therefore occupies the contiguous slice + `[n * panel_bytes, (n + 1) * panel_bytes)` where + `panel_bytes = panel_width * panel_height * 3`, and each panel slice is a + ready-to-use RGB image. This is what lets a converter emit one video stream + per panel (and a mixed stream) without any hardware de-tangling. + + All multi-byte integers are big-endian. Grayscale (`WFrame`) frames are + normalized to RGB (`r = g = b = w`) before writing so the stream is uniform + and the converter never needs to branch on frame type. + + ## Header (#{26} bytes) + + magic 8 bytes "OCTOREC1" + version u8 format version + kind u8 0 = rgb (only value currently emitted) + num_panels u16 + panel_width u16 + panel_height u16 + reserved u16 always 0 + started_at_ms u64 wall-clock ms (System.system_time/1) at start + + ## Record (`4 + frame_bytes` bytes, repeated) + + offset_ms u32 System.monotonic_time ms since recording start + pixels frame_bytes bytes (RGB, panel-major/row-major) + """ + + @magic "OCTOREC1" + @version 1 + @kind_rgb 0 + @header_size 26 + + @type header :: %{ + version: non_neg_integer(), + kind: non_neg_integer(), + num_panels: non_neg_integer(), + panel_width: non_neg_integer(), + panel_height: non_neg_integer(), + started_at_ms: non_neg_integer() + } + + @doc "The magic bytes every recording starts with." + @spec magic() :: binary() + def magic, do: @magic + + @doc "Current format version." + @spec version() :: non_neg_integer() + def version, do: @version + + @doc "Size of the fixed header in bytes." + @spec header_size() :: non_neg_integer() + def header_size, do: @header_size + + @doc """ + Number of pixel bytes in a single frame for the given geometry (RGB). + """ + @spec frame_bytes(non_neg_integer(), non_neg_integer(), non_neg_integer()) :: non_neg_integer() + def frame_bytes(num_panels, panel_width, panel_height), + do: num_panels * panel_width * panel_height * 3 + + @doc "Encode the fixed-size header binary." + @spec header(pos_integer(), pos_integer(), pos_integer(), non_neg_integer()) :: binary() + def header(num_panels, panel_width, panel_height, started_at_ms) do + << + @magic::binary, + @version::8, + @kind_rgb::8, + num_panels::16, + panel_width::16, + panel_height::16, + 0::16, + started_at_ms::64 + >> + end + + @doc """ + Encode a single frame record. + + `rgb_data` must already be RGB pixel bytes of the expected frame size; use + `normalize/3` to coerce raw mixer frame data into that shape. + """ + @spec record(non_neg_integer(), binary()) :: binary() + def record(offset_ms, rgb_data) when is_integer(offset_ms) and is_binary(rgb_data) do + <> + end + + @doc """ + Coerce raw mixer frame data into RGB pixel bytes of `expected_rgb` size. + + * data already the RGB size -> returned as-is + * data the grayscale size (1/3 of RGB) -> expanded to `r = g = b = w` + * anything else -> `:error` (caller should drop the frame) + """ + @spec normalize(binary(), non_neg_integer(), non_neg_integer()) :: + {:ok, binary()} | :error + def normalize(data, expected_rgb, expected_w) when is_binary(data) do + case byte_size(data) do + ^expected_rgb -> {:ok, data} + ^expected_w -> {:ok, expand_w(data)} + _ -> :error + end + end + + @doc "Parse the header from the front of a recording binary." + @spec parse_header(binary()) :: {:ok, header(), binary()} | {:error, :invalid_header} + def parse_header( + <<@magic, version::8, kind::8, num_panels::16, panel_width::16, panel_height::16, + _reserved::16, started_at_ms::64, rest::binary>> + ) do + header = %{ + version: version, + kind: kind, + num_panels: num_panels, + panel_width: panel_width, + panel_height: panel_height, + started_at_ms: started_at_ms + } + + {:ok, header, rest} + end + + def parse_header(_), do: {:error, :invalid_header} + + @doc """ + Parse a complete recording binary into `{header, records}` where each record + is `{offset_ms, rgb_data}`. Intended for tests and offline tooling; streaming + readers should use `parse_header/1` plus their own chunked loop for large + files. + """ + @spec parse(binary()) :: {:ok, header(), [{non_neg_integer(), binary()}]} | {:error, term()} + def parse(binary) when is_binary(binary) do + with {:ok, header, rest} <- parse_header(binary) do + fb = frame_bytes(header.num_panels, header.panel_width, header.panel_height) + + case parse_records(fb, rest, []) do + {:ok, records} -> {:ok, header, records} + {:error, _} = err -> err + end + end + end + + defp parse_records(_fb, <<>>, acc), do: {:ok, Enum.reverse(acc)} + + defp parse_records(fb, bin, acc) do + case bin do + <> -> + parse_records(fb, rest, [{offset, data} | acc]) + + _ -> + {:error, :truncated_record} + end + end + + defp expand_w(data) do + for <>, into: <<>>, do: <> + end + + defp clamp_u32(n) when n < 0, do: 0 + defp clamp_u32(n) when n > 0xFFFFFFFF, do: 0xFFFFFFFF + defp clamp_u32(n), do: n +end diff --git a/octopus/lib/octopus/recording/panel_recorder.ex b/octopus/lib/octopus/recording/panel_recorder.ex new file mode 100644 index 00000000..23886eb8 --- /dev/null +++ b/octopus/lib/octopus/recording/panel_recorder.ex @@ -0,0 +1,374 @@ +defmodule Octopus.Recording.PanelRecorder do + @moduledoc """ + Records the frames the mixer sends to the LED panels into an append-only + recording stream (see `Octopus.Recording.Format`). + + ## Safety + + This recorder is designed so it can never crash or influence the running + installation: + + * It is a passive `Phoenix.PubSub` subscriber of the mixer's frame topic. + Broadcasts are asynchronous and fire-and-forget, so the mixer and + broadcaster never wait for, or are affected by, the recorder. + * It only subscribes while actively recording, so when disabled there is + zero message traffic and zero overhead. + * Every incoming frame is processed inside a `try/rescue/catch`; a bad + frame or a sink error degrades to a dropped frame (and, for sink errors, + a clean stop), never a crash. + * It guards its own mailbox: if messages pile up faster than they can be + written (slow disk / slow remote sink) it drops frames instead of growing + memory without bound. + * It lives in an isolated supervision subtree, so even an unexpected crash + only restarts the recorder. + + ## Tap point + + Subscribes to `Octopus.Mixer` and records `{:mixer, {:frame, frame}}` + messages. These carry the logical, full-resolution frame *before* hardware + de-tangling and UDP splitting, i.e. exactly what should be visualized. + + The same frame is broadcast once per UDP split part, so identical frames that + arrive within `#{5}` ms of each other are de-duplicated; dropped duplicates + simply extend the previous frame's on-screen duration in playback. + """ + + use GenServer + require Logger + + alias Octopus.Installation + alias Octopus.Recording + alias Octopus.Recording.{Format, Sink} + alias Octopus.Protobuf.{RGBFrame, WFrame} + + @dedup_window_ms 5 + + defmodule State do + @moduledoc false + defstruct active: false, + sink_mod: nil, + sink: nil, + path: nil, + start_mono_ms: nil, + started_at_ms: nil, + num_panels: nil, + panel_width: nil, + panel_height: nil, + frame_bytes: nil, + w_bytes: nil, + max_queue: 600, + last_rgb: nil, + last_offset_ms: nil, + written: 0, + dropped: 0 + end + + ## Client API + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @doc """ + Start a recording. + + Options: + + * `:sink_mod` - module implementing `Octopus.Recording.Sink` + (default `Octopus.Recording.Sink.File`). + * `:sink_opts` - options passed to the sink's `open/1`. For the file sink + a `:path` is used; when omitted a timestamped path under the configured + output directory is generated. + * `:dir` - convenience for the file sink: output directory for the + generated filename. + """ + @spec start_recording(keyword()) :: {:ok, String.t()} | {:error, term()} + def start_recording(opts \\ []) do + GenServer.call(__MODULE__, {:start_recording, opts}) + end + + @doc "Stop the current recording, flushing and closing the sink." + @spec stop_recording() :: :ok | {:error, :not_recording} + def stop_recording do + GenServer.call(__MODULE__, :stop_recording) + end + + @doc "Return a status map describing the recorder." + @spec status() :: map() + def status do + GenServer.call(__MODULE__, :status) + end + + ## Server callbacks + + @impl true + def init(_opts) do + {:ok, %State{max_queue: Recording.max_queue()}, {:continue, :maybe_autostart}} + end + + @impl true + def handle_continue(:maybe_autostart, %State{} = state) do + if Recording.enabled?() do + case do_start(state, []) do + {:ok, new_state, path} -> + Logger.info("[recording] Auto-started panel recording -> #{path}") + {:noreply, new_state} + + {:error, reason} -> + Logger.warning("[recording] Auto-start failed: #{inspect(reason)}. Staying idle.") + {:noreply, state} + end + else + {:noreply, state} + end + end + + @impl true + def handle_call({:start_recording, opts}, _from, %State{active: true} = state) do + _ = opts + {:reply, {:error, :already_recording}, state} + end + + def handle_call({:start_recording, opts}, _from, %State{} = state) do + case do_start(state, opts) do + {:ok, new_state, path} -> {:reply, {:ok, path}, new_state} + {:error, reason} -> {:reply, {:error, reason}, state} + end + end + + def handle_call(:stop_recording, _from, %State{active: false} = state) do + {:reply, {:error, :not_recording}, state} + end + + def handle_call(:stop_recording, _from, %State{} = state) do + {:reply, :ok, do_stop(state)} + end + + def handle_call(:status, _from, %State{} = state) do + {:reply, status_map(state), state} + end + + @impl true + def handle_info({:mixer, {:frame, frame}}, %State{active: true} = state) do + {:noreply, maybe_record(frame, state)} + end + + # Ignore all other mixer traffic (config changes, and frames while inactive). + def handle_info({:mixer, _}, %State{} = state), do: {:noreply, state} + def handle_info(_msg, %State{} = state), do: {:noreply, state} + + @impl true + def terminate(_reason, %State{active: true} = state) do + _ = do_stop(state) + :ok + end + + def terminate(_reason, _state), do: :ok + + ## Recording lifecycle + + defp do_start(%State{} = state, opts) do + num_panels = Installation.num_panels() + panel_width = Installation.panel_width() + panel_height = Installation.panel_height() + + started_at_ms = System.system_time(:millisecond) + {sink_mod, sink_opts} = resolve_sink(opts, started_at_ms) + + with {:ok, sink} <- sink_mod.open(sink_opts), + header = Format.header(num_panels, panel_width, panel_height, started_at_ms), + {:ok, sink} <- sink_mod.write(sink, header) do + Octopus.Mixer.subscribe() + + state = %State{ + state + | active: true, + sink_mod: sink_mod, + sink: sink, + path: sink_mod.describe(sink), + start_mono_ms: System.monotonic_time(:millisecond), + started_at_ms: started_at_ms, + num_panels: num_panels, + panel_width: panel_width, + panel_height: panel_height, + frame_bytes: Format.frame_bytes(num_panels, panel_width, panel_height), + w_bytes: num_panels * panel_width * panel_height, + last_rgb: nil, + last_offset_ms: nil, + written: 0, + dropped: 0 + } + + {:ok, state, sink_mod.describe(sink)} + end + rescue + error -> {:error, error} + end + + defp do_stop(%State{sink_mod: sink_mod, sink: sink} = state) do + Octopus.Mixer.unsubscribe() + + if sink_mod && sink do + _ = safe_close(sink_mod, sink) + end + + Logger.info( + "[recording] Stopped panel recording (#{state.written} frames written, #{state.dropped} dropped)" + ) + + %State{max_queue: state.max_queue} + end + + # Resolve which sink to use and the options to open it with. An explicit + # `:sink_mod` in the start options wins; otherwise the configured `:sink` + # spec is used (defaulting to a file sink). The result is wrapped in gzip when + # compression is requested. + defp resolve_sink(opts, started_at_ms) do + {mod, sink_opts} = base_sink(opts, started_at_ms) + compress? = Keyword.get(opts, :compress, Recording.compress?()) + Sink.Gzip.wrap(mod, sink_opts, compress?, Recording.gzip_level()) + end + + defp base_sink(opts, started_at_ms) do + case Keyword.fetch(opts, :sink_mod) do + {:ok, Sink.File} -> + {Sink.File, file_open_opts(opts, [], started_at_ms)} + + {:ok, mod} -> + {mod, Keyword.get(opts, :sink_opts, [])} + + :error -> + case Recording.sink_spec() do + {:remote, remote_opts} -> + {Sink.Remote, remote_opts} + + {:file, file_opts} -> + {Sink.File, file_open_opts(opts, file_opts, started_at_ms)} + + _ -> + {Sink.File, file_open_opts(opts, [], started_at_ms)} + end + end + end + + defp file_open_opts(opts, file_opts, started_at_ms) do + sink_opts = Keyword.get(opts, :sink_opts, []) + + path = + cond do + p = Keyword.get(sink_opts, :path) -> p + p = Keyword.get(file_opts, :path) -> p + true -> Path.join(file_dir(opts, file_opts), generated_filename(started_at_ms)) + end + + [path: path] + end + + defp file_dir(opts, file_opts) do + Keyword.get(opts, :dir) || Keyword.get(file_opts, :dir) || Recording.output_dir() + end + + defp generated_filename(started_at_ms) do + stamp = + started_at_ms + |> DateTime.from_unix!(:millisecond) + |> Calendar.strftime("%Y%m%d-%H%M%S") + + "panels-#{stamp}.octorec" + end + + ## Frame handling + + defp maybe_record(frame, %State{} = state) do + if overloaded?(state) do + %State{state | dropped: state.dropped + 1} + else + record_frame(frame, state) + end + rescue + error -> + Logger.warning("[recording] Dropping frame after error: #{inspect(error)}") + %State{state | dropped: state.dropped + 1} + catch + kind, reason -> + Logger.warning("[recording] Dropping frame after #{kind}: #{inspect(reason)}") + %State{state | dropped: state.dropped + 1} + end + + defp overloaded?(%State{max_queue: max}) do + case Process.info(self(), :message_queue_len) do + {:message_queue_len, len} -> len > max + _ -> false + end + end + + defp record_frame(frame, %State{} = state) do + with {:ok, data} <- frame_data(frame), + {:ok, rgb} <- Format.normalize(data, state.frame_bytes, state.w_bytes) do + offset_ms = max(System.monotonic_time(:millisecond) - state.start_mono_ms, 0) + + if duplicate?(state, rgb, offset_ms) do + state + else + write_record(state, offset_ms, rgb) + end + else + _ -> %State{state | dropped: state.dropped + 1} + end + end + + defp frame_data(%RGBFrame{data: data}) when is_binary(data), do: {:ok, data} + defp frame_data(%WFrame{data: data}) when is_binary(data), do: {:ok, data} + defp frame_data(%{data: data}) when is_binary(data), do: {:ok, data} + defp frame_data(_), do: :error + + defp duplicate?(%State{last_rgb: last, last_offset_ms: last_offset}, rgb, offset_ms) + when is_binary(last) and is_integer(last_offset) do + offset_ms - last_offset < @dedup_window_ms and rgb == last + end + + defp duplicate?(_state, _rgb, _offset_ms), do: false + + defp write_record(%State{sink_mod: sink_mod, sink: sink} = state, offset_ms, rgb) do + record = Format.record(offset_ms, rgb) + + case sink_mod.write(sink, record) do + {:ok, sink} -> + %State{ + state + | sink: sink, + last_rgb: rgb, + last_offset_ms: offset_ms, + written: state.written + 1 + } + + {:error, reason} -> + Logger.error("[recording] Sink write failed: #{inspect(reason)}. Stopping recording.") + do_stop(state) + end + end + + defp safe_close(sink_mod, sink) do + sink_mod.close(sink) + rescue + error -> Logger.warning("[recording] Sink close error: #{inspect(error)}") + end + + defp status_map(%State{active: false} = state) do + %{active: false, written: state.written, dropped: state.dropped} + end + + defp status_map(%State{} = state) do + %{ + active: true, + sink: state.path, + written: state.written, + dropped: state.dropped, + started_at_ms: state.started_at_ms, + num_panels: state.num_panels, + panel_width: state.panel_width, + panel_height: state.panel_height, + frame_bytes: state.frame_bytes + } + end +end diff --git a/octopus/lib/octopus/recording/sink.ex b/octopus/lib/octopus/recording/sink.ex new file mode 100644 index 00000000..5eb067c5 --- /dev/null +++ b/octopus/lib/octopus/recording/sink.ex @@ -0,0 +1,32 @@ +defmodule Octopus.Recording.Sink do + @moduledoc """ + Behaviour for a recording output target. + + A sink is where the raw recording byte stream (header + frame records, see + `Octopus.Recording.Format`) is written. The recorder is transport-agnostic: + it opens a sink, writes opaque `iodata` to it, and closes it. This lets the + same recording pipeline write to a local file today and stream to a remote + server later without touching the recorder. + + Implementations must be non-blocking enough not to endanger the recorder's + mailbox. The recorder additionally guards against overload by dropping frames + when its mailbox grows too large, but a sink should still avoid unbounded + blocking (e.g. use buffered/delayed writes for files, bounded timeouts for + network sinks). + """ + + @typedoc "Opaque per-sink state threaded through `write/2` and `close/1`." + @type state :: term() + + @doc "Open the sink. Returns the initial sink state." + @callback open(opts :: keyword()) :: {:ok, state()} | {:error, term()} + + @doc "Append bytes to the sink, returning the updated state." + @callback write(state(), iodata()) :: {:ok, state()} | {:error, term()} + + @doc "Close the sink, flushing any buffered data." + @callback close(state()) :: :ok + + @doc "Human-readable description of the sink target (for status/logging)." + @callback describe(state()) :: String.t() +end diff --git a/octopus/lib/octopus/recording/sink/file.ex b/octopus/lib/octopus/recording/sink/file.ex new file mode 100644 index 00000000..0a380c3f --- /dev/null +++ b/octopus/lib/octopus/recording/sink/file.ex @@ -0,0 +1,38 @@ +defmodule Octopus.Recording.Sink.File do + @moduledoc """ + `Octopus.Recording.Sink` that appends the recording stream to a local file. + + Opens the file in `:raw` + `:delayed_write` mode so writes are buffered by + the runtime and flushed in batches. This keeps per-frame writes cheap (no + syscall per frame at 60 fps) and avoids blocking the recorder. + """ + + @behaviour Octopus.Recording.Sink + + @impl true + def open(opts) do + path = Keyword.fetch!(opts, :path) + + with :ok <- File.mkdir_p(Path.dirname(path)), + {:ok, io} <- File.open(path, [:write, :binary, :raw, :delayed_write]) do + {:ok, %{io: io, path: path}} + end + end + + @impl true + def write(%{io: io} = state, iodata) do + case IO.binwrite(io, iodata) do + :ok -> {:ok, state} + other -> {:error, other} + end + end + + @impl true + def close(%{io: io}) do + _ = File.close(io) + :ok + end + + @impl true + def describe(%{path: path}), do: "file:" <> path +end diff --git a/octopus/lib/octopus/recording/sink/gzip.ex b/octopus/lib/octopus/recording/sink/gzip.ex new file mode 100644 index 00000000..a1cf4d16 --- /dev/null +++ b/octopus/lib/octopus/recording/sink/gzip.ex @@ -0,0 +1,124 @@ +defmodule Octopus.Recording.Sink.Gzip do + @moduledoc """ + A composable `Octopus.Recording.Sink` that gzip-compresses the stream and + forwards the compressed bytes to an inner sink (file or remote). + + Compression uses Erlang's built-in `:zlib`, so there are **no native + dependencies** — it works out of the box on a Raspberry Pi (and anywhere OTP + runs). Output is a standard gzip stream, so the resulting files can be read by + `gunzip`/`zcat` and are decoded transparently by the encoders. + + LED-panel recordings are highly redundant, so gzip typically shrinks them + substantially (often several-fold, and 10x+ for dark/sparse content); only + pathological full-frame noise stays near 1:1. + + ## Crash safety + + A gzip stream is only fully valid once finalized on `close/1` (which the + recorder calls on stop and on normal supervised shutdown). To bound data loss + from a hard VM crash, the stream is `:sync`-flushed every `:flush_every` + writes (default #{200}); a sync flush keeps the compression dictionary, so the + ratio impact is small. + + ## Options + + * `:inner_mod` - the wrapped sink module (required) + * `:inner_opts` - options passed to the inner sink's `open/1` (default `[]`) + * `:level` - zlib compression level 0..9 (default #{6}); lower is faster and + cheaper on constrained CPUs + * `:flush_every` - sync-flush cadence in writes; `0` disables periodic + flushing (default #{200}) + """ + + @behaviour Octopus.Recording.Sink + + alias Octopus.Recording.Sink + + @default_level 6 + @default_flush_every 200 + # 15-bit window + 16 selects a gzip (not zlib) wrapper. + @gzip_window_bits 31 + + @impl true + def open(opts) do + inner_mod = Keyword.fetch!(opts, :inner_mod) + inner_opts = Keyword.get(opts, :inner_opts, []) + level = Keyword.get(opts, :level, @default_level) + flush_every = Keyword.get(opts, :flush_every, @default_flush_every) + + with {:ok, inner} <- inner_mod.open(inner_opts) do + z = :zlib.open() + :ok = :zlib.deflateInit(z, level, :deflated, @gzip_window_bits, 8, :default) + + {:ok, %{z: z, inner_mod: inner_mod, inner: inner, flush_every: flush_every, writes: 0}} + end + end + + @impl true + def write(state, iodata) do + writes = state.writes + 1 + flush = if flush?(state.flush_every, writes), do: :sync, else: :none + compressed = :zlib.deflate(state.z, iodata, flush) + state = %{state | writes: writes} + + if IO.iodata_length(compressed) == 0 do + {:ok, state} + else + case state.inner_mod.write(state.inner, compressed) do + {:ok, inner} -> {:ok, %{state | inner: inner}} + {:error, _} = err -> err + end + end + end + + @impl true + def close(state) do + final = :zlib.deflate(state.z, [], :finish) + _ = :zlib.deflateEnd(state.z) + _ = :zlib.close(state.z) + + inner = + if IO.iodata_length(final) > 0 do + case state.inner_mod.write(state.inner, final) do + {:ok, inner} -> inner + _ -> state.inner + end + else + state.inner + end + + state.inner_mod.close(inner) + end + + @impl true + def describe(state), do: "gzip+" <> state.inner_mod.describe(state.inner) + + @doc """ + Wrap `{sink_mod, sink_opts}` in gzip when `compress?` is true, otherwise return + it unchanged. For a file sink the target path gains a `.gz` suffix. + """ + @spec wrap(module(), keyword(), boolean(), non_neg_integer()) :: {module(), keyword()} + def wrap(sink_mod, sink_opts, compress?, level \\ @default_level) + + def wrap(sink_mod, sink_opts, false, _level), do: {sink_mod, sink_opts} + + def wrap(sink_mod, sink_opts, true, level) do + {__MODULE__, [inner_mod: sink_mod, inner_opts: gz_path(sink_mod, sink_opts), level: level]} + end + + defp flush?(0, _writes), do: false + defp flush?(every, writes), do: rem(writes, every) == 0 + + defp gz_path(Sink.File, opts) do + case Keyword.fetch(opts, :path) do + {:ok, path} -> Keyword.put(opts, :path, ensure_gz(path)) + :error -> opts + end + end + + defp gz_path(_mod, opts), do: opts + + defp ensure_gz(path) do + if String.ends_with?(path, ".gz"), do: path, else: path <> ".gz" + end +end diff --git a/octopus/lib/octopus/recording/sink/remote.ex b/octopus/lib/octopus/recording/sink/remote.ex new file mode 100644 index 00000000..aa191668 --- /dev/null +++ b/octopus/lib/octopus/recording/sink/remote.ex @@ -0,0 +1,90 @@ +defmodule Octopus.Recording.Sink.Remote do + @moduledoc """ + `Octopus.Recording.Sink` that streams the recording to a remote server over + a plain TCP connection. + + The exact same byte stream that would be written to a file (header followed + by frame records, see `Octopus.Recording.Format`) is sent over the socket, so + the receiving end can simply append it to a `.octorec` file and later run the + encoder on it, or convert on the fly. + + ## Safety + + Network I/O must never endanger the running installation. Two bounds keep it + safe: + + * A connect timeout on `open/1` — an unreachable server fails the + *recording start*, it does not hang. + * A send timeout (with `send_timeout_close: true`) on every write — a + stalled server causes a bounded `{:error, :timeout}` that the recorder + turns into a clean stop, rather than an indefinite block. + + Because the recorder is a passive PubSub subscriber and drops frames when its + mailbox backs up, a slow network only degrades the recording (dropped + frames); the mixer and broadcaster are never affected. + + ## Options + + * `:host` - hostname string, charlist, or IPv4/IPv6 tuple (required) + * `:port` - TCP port (required) + * `:connect_timeout` - ms to wait for the connection (default 5000) + * `:send_timeout` - ms to wait for each write (default 2000) + """ + + @behaviour Octopus.Recording.Sink + + @default_connect_timeout 5_000 + @default_send_timeout 2_000 + + @impl true + def open(opts) do + host = Keyword.fetch!(opts, :host) + port = Keyword.fetch!(opts, :port) + connect_timeout = Keyword.get(opts, :connect_timeout, @default_connect_timeout) + send_timeout = Keyword.get(opts, :send_timeout, @default_send_timeout) + + tcp_opts = [ + :binary, + packet: :raw, + active: false, + send_timeout: send_timeout, + send_timeout_close: true + ] + + case :gen_tcp.connect(resolve_host(host), port, tcp_opts, connect_timeout) do + {:ok, socket} -> + {:ok, %{socket: socket, host: host, port: port}} + + {:error, reason} -> + {:error, {:connect_failed, reason}} + end + end + + @impl true + def write(%{socket: socket} = state, iodata) do + case :gen_tcp.send(socket, iodata) do + :ok -> {:ok, state} + {:error, reason} -> {:error, reason} + end + end + + @impl true + def close(%{socket: socket}) do + _ = :gen_tcp.close(socket) + :ok + end + + @impl true + def describe(%{host: host, port: port}), do: "tcp://#{format_host(host)}:#{port}" + + defp resolve_host(host) when is_tuple(host), do: host + defp resolve_host(host) when is_binary(host), do: String.to_charlist(host) + defp resolve_host(host) when is_list(host), do: host + + defp format_host(host) when is_tuple(host) do + host |> :inet.ntoa() |> to_string() + end + + defp format_host(host) when is_binary(host), do: host + defp format_host(host) when is_list(host), do: to_string(host) +end diff --git a/octopus/lib/octopus/recording/supervisor.ex b/octopus/lib/octopus/recording/supervisor.ex new file mode 100644 index 00000000..6f47b15a --- /dev/null +++ b/octopus/lib/octopus/recording/supervisor.ex @@ -0,0 +1,25 @@ +defmodule Octopus.Recording.Supervisor do + @moduledoc """ + Isolated supervision subtree for the recording subsystem. + + Started unconditionally by the application, but its children are cheap and + idle unless a recording is active. Keeping recording under its own supervisor + (with a `:one_for_one` strategy) guarantees that a recorder restart can never + affect the mixer, broadcaster, or apps. + """ + + use Supervisor + + def start_link(opts) do + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + def init(_opts) do + children = [ + Octopus.Recording.PanelRecorder + ] + + Supervisor.init(children, strategy: :one_for_one) + end +end diff --git a/octopus/test/octopus/recording/encoder_test.exs b/octopus/test/octopus/recording/encoder_test.exs new file mode 100644 index 00000000..cc14b116 --- /dev/null +++ b/octopus/test/octopus/recording/encoder_test.exs @@ -0,0 +1,117 @@ +defmodule Octopus.Recording.EncoderTest do + use ExUnit.Case, async: true + + alias Octopus.Recording.{Encoder, Format} + + describe "resample/2" do + test "holds the most recent frame at each fixed tick" do + records = [{0, "A"}, {100, "B"}, {250, "C"}] + # fps 10 -> dt 100ms -> ticks at 0, 100, 200 + assert Encoder.resample(records, 10) == ["A", "B", "B"] + end + + test "single record yields a single frame" do + assert Encoder.resample([{0, "A"}], 30) == ["A"] + end + + test "empty records yield no frames" do + assert Encoder.resample([], 30) == [] + end + end + + test "panel_frame/3 slices a panel's contiguous block" do + geom = {3, 2, 2} + block = 2 * 2 * 3 + data = for b <- 0..(3 * block - 1), into: <<>>, do: <> + + assert Encoder.panel_frame(data, 0, geom) == binary_part(data, 0, block) + assert Encoder.panel_frame(data, 1, geom) == binary_part(data, block, block) + assert Encoder.panel_frame(data, 2, geom) == binary_part(data, 2 * block, block) + end + + test "strip_frame/2 transposes panel-major data into a row-major strip" do + # 2 panels, 2x2. Each pixel's 3 bytes carry its index 0..7. + geom = {2, 2, 2} + data = for i <- 0..7, into: <<>>, do: <> + + px = fn i -> <> end + + # Row 0: p0(x0,x1), p1(x0,x1) = 0,1,4,5 ; Row 1: 2,3,6,7 + expected = + px.(0) <> px.(1) <> px.(4) <> px.(5) <> px.(2) <> px.(3) <> px.(6) <> px.(7) + + assert Encoder.strip_frame(data, geom) == expected + end + + @tag :ffmpeg + test "encode/2 produces per-panel and mixed videos" do + unless Encoder.ffmpeg_available?() do + # Keep the suite independent of ffmpeg being installed. + IO.puts("skipping: ffmpeg not available") + else + num_panels = 3 + pw = 2 + ph = 2 + frame_bytes = Format.frame_bytes(num_panels, pw, ph) + + frame_a = :binary.copy(<<10>>, frame_bytes) + frame_b = :binary.copy(<<200>>, frame_bytes) + + binary = + Format.header(num_panels, pw, ph, 0) <> + Format.record(0, frame_a) <> + Format.record(100, frame_b) + + base = Path.join(System.tmp_dir!(), "octorec-enc-#{System.unique_integer([:positive])}") + input = base <> ".octorec" + out_dir = base <> "_out" + File.write!(input, binary) + + on_exit(fn -> + File.rm(input) + File.rm_rf(out_dir) + end) + + assert {:ok, outputs} = Encoder.encode(input, out: out_dir, fps: 10, scale: 4) + + assert length(outputs) == num_panels + 1 + assert Path.join(out_dir, "mixed.mp4") in outputs + + for output <- outputs do + assert File.regular?(output) + assert File.stat!(output).size > 0 + end + end + end + + @tag :ffmpeg + test "encode/2 transparently reads a gzip-compressed .octorec.gz" do + unless Encoder.ffmpeg_available?() do + IO.puts("skipping: ffmpeg not available") + else + num_panels = 3 + pw = 2 + ph = 2 + frame_bytes = Format.frame_bytes(num_panels, pw, ph) + + binary = + Format.header(num_panels, pw, ph, 0) <> + Format.record(0, :binary.copy(<<10>>, frame_bytes)) <> + Format.record(100, :binary.copy(<<200>>, frame_bytes)) + + base = Path.join(System.tmp_dir!(), "octorec-gz-#{System.unique_integer([:positive])}") + input = base <> ".octorec.gz" + out_dir = base <> "_out" + File.write!(input, :zlib.gzip(binary)) + + on_exit(fn -> + File.rm(input) + File.rm_rf(out_dir) + end) + + assert {:ok, outputs} = Encoder.encode(input, out: out_dir, fps: 10, scale: 4) + assert length(outputs) == num_panels + 1 + assert Enum.all?(outputs, &(File.regular?(&1) and File.stat!(&1).size > 0)) + end + end +end diff --git a/octopus/test/octopus/recording/format_test.exs b/octopus/test/octopus/recording/format_test.exs new file mode 100644 index 00000000..27572260 --- /dev/null +++ b/octopus/test/octopus/recording/format_test.exs @@ -0,0 +1,77 @@ +defmodule Octopus.Recording.FormatTest do + use ExUnit.Case, async: true + + alias Octopus.Recording.Format + + test "header round-trips through parse_header/1" do + header = Format.header(12, 8, 8, 1_700_000_000_000) + + assert byte_size(header) == Format.header_size() + assert {:ok, parsed, <<>>} = Format.parse_header(header) + + assert parsed == %{ + version: Format.version(), + kind: 0, + num_panels: 12, + panel_width: 8, + panel_height: 8, + started_at_ms: 1_700_000_000_000 + } + end + + test "parse_header/1 rejects non-recording binaries" do + assert {:error, :invalid_header} = Format.parse_header("not a recording") + end + + test "frame_bytes/3 is num_panels * width * height * 3" do + assert Format.frame_bytes(12, 8, 8) == 12 * 8 * 8 * 3 + end + + describe "normalize/3" do + test "passes RGB-sized data through unchanged" do + rgb = :binary.copy(<<7>>, 12 * 8 * 8 * 3) + assert {:ok, ^rgb} = Format.normalize(rgb, 12 * 8 * 8 * 3, 12 * 8 * 8) + end + + test "expands grayscale data to r = g = b = w" do + w = <<10, 20, 30>> + assert {:ok, <<10, 10, 10, 20, 20, 20, 30, 30, 30>>} = Format.normalize(w, 9, 3) + end + + test "returns :error for unexpected sizes" do + assert :error = Format.normalize(<<1, 2, 3, 4, 5>>, 9, 3) + end + end + + test "records round-trip through parse/1 in order" do + num_panels = 2 + pw = 2 + ph = 2 + frame_bytes = Format.frame_bytes(num_panels, pw, ph) + + a = :binary.copy(<<1>>, frame_bytes) + b = :binary.copy(<<2>>, frame_bytes) + + binary = + Format.header(num_panels, pw, ph, 42) <> + Format.record(0, a) <> + Format.record(33, b) + + assert {:ok, header, records} = Format.parse(binary) + assert header.num_panels == num_panels + assert records == [{0, a}, {33, b}] + end + + test "parse/1 reports truncated trailing records" do + binary = Format.header(2, 2, 2, 0) <> <<0::32, 1, 2, 3>> + assert {:error, :truncated_record} = Format.parse(binary) + end + + test "record/2 clamps out-of-range offsets into u32" do + <> = Format.record(-5, <<>>) + assert offset == 0 + + <> = Format.record(0xFFFFFFFF + 100, <<>>) + assert big == 0xFFFFFFFF + end +end diff --git a/octopus/test/octopus/recording/panel_recorder_test.exs b/octopus/test/octopus/recording/panel_recorder_test.exs new file mode 100644 index 00000000..9f5a39fa --- /dev/null +++ b/octopus/test/octopus/recording/panel_recorder_test.exs @@ -0,0 +1,178 @@ +defmodule Octopus.Recording.PanelRecorderTest do + use ExUnit.Case, async: false + + alias Octopus.{Installation, Recording} + alias Octopus.Recording.Format + alias Octopus.Recording.Sink + alias Octopus.Protobuf.{RGBFrame, WFrame} + + @mixer_topic "mixer" + + setup do + # These are part of the running application; make sure they exist even if + # the app was not started for the test run. + unless Process.whereis(Octopus.PubSub) do + start_supervised!({Phoenix.PubSub, name: Octopus.PubSub}) + end + + unless Process.whereis(Recording.PanelRecorder) do + start_supervised!(Recording.PanelRecorder) + end + + # Ensure we begin idle regardless of what previous tests did. + _ = Recording.stop() + on_exit(fn -> _ = Recording.stop() end) + + :ok + end + + test "records mixer frames to an append-only file, dedups split parts and normalizes W frames" do + dir = Path.join(System.tmp_dir!(), "octorec-#{System.unique_integer([:positive])}") + on_exit(fn -> File.rm_rf(dir) end) + + num_panels = Installation.num_panels() + pw = Installation.panel_width() + ph = Installation.panel_height() + rgb_size = num_panels * pw * ph * 3 + w_size = num_panels * pw * ph + + frame_a = %RGBFrame{data: :binary.copy(<<11>>, rgb_size)} + frame_b = %RGBFrame{data: :binary.copy(<<22>>, rgb_size)} + wframe = %WFrame{data: :binary.copy(<<50>>, w_size)} + + expected_a = frame_a.data + expected_b = frame_b.data + expected_w = :binary.copy(<<50>>, rgb_size) + + assert {:ok, "file:" <> path} = Recording.start(dir: dir) + + # A duplicate immediately follows A (as the mixer does per UDP split part) + # and must be de-duplicated. + broadcast(frame_a) + broadcast(frame_a) + broadcast(frame_b) + broadcast(wframe) + + # stop/0 is a synchronous call to the recorder; all frames broadcast before + # it are guaranteed to have been processed by the time it returns. + assert :ok = Recording.stop() + + assert {:ok, header, records} = Format.parse(File.read!(path)) + assert header.num_panels == num_panels + assert header.panel_width == pw + assert header.panel_height == ph + + # Filter to the frames this test produced (the running mixer may also emit + # blank idle frames onto the same topic). + distinctive = + records + |> Enum.map(fn {_offset, data} -> data end) + |> Enum.filter(&(&1 in [expected_a, expected_b, expected_w])) + + assert distinctive == [expected_a, expected_b, expected_w] + end + + test "status reflects active/idle state" do + dir = Path.join(System.tmp_dir!(), "octorec-#{System.unique_integer([:positive])}") + on_exit(fn -> File.rm_rf(dir) end) + + assert %{active: false} = Recording.status() + + assert {:ok, _target} = Recording.start(dir: dir) + status = Recording.status() + assert status.active == true + assert status.num_panels == Installation.num_panels() + + assert :ok = Recording.stop() + assert %{active: false} = Recording.status() + end + + test "start/1 twice returns already_recording" do + dir = Path.join(System.tmp_dir!(), "octorec-#{System.unique_integer([:positive])}") + on_exit(fn -> File.rm_rf(dir) end) + + assert {:ok, _} = Recording.start(dir: dir) + assert {:error, :already_recording} = Recording.start(dir: dir) + end + + test "compress: true writes a gzip file that decompresses to a valid recording" do + dir = Path.join(System.tmp_dir!(), "octorec-#{System.unique_integer([:positive])}") + on_exit(fn -> File.rm_rf(dir) end) + + num_panels = Installation.num_panels() + rgb_size = num_panels * Installation.panel_width() * Installation.panel_height() * 3 + frame_a = %RGBFrame{data: :binary.copy(<<91>>, rgb_size)} + frame_b = %RGBFrame{data: :binary.copy(<<200>>, rgb_size)} + + assert {:ok, "gzip+file:" <> path} = Recording.start(dir: dir, compress: true) + assert String.ends_with?(path, ".octorec.gz") + + broadcast(frame_a) + broadcast(frame_b) + assert :ok = Recording.stop() + + raw = path |> File.read!() |> :zlib.gunzip() + assert {:ok, header, records} = Format.parse(raw) + assert header.num_panels == num_panels + + distinctive = + records + |> Enum.map(fn {_o, d} -> d end) + |> Enum.filter(&(&1 in [frame_a.data, frame_b.data])) + + assert distinctive == [frame_a.data, frame_b.data] + end + + test "streams the recording to a remote TCP server" do + {:ok, listen} = + :gen_tcp.listen(0, [:binary, packet: :raw, active: false, reuseaddr: true]) + + {:ok, port} = :inet.port(listen) + test_pid = self() + + _acceptor = + spawn_link(fn -> + {:ok, sock} = :gen_tcp.accept(listen, 2000) + send(test_pid, {:received, recv_all(sock, <<>>)}) + end) + + num_panels = Installation.num_panels() + pw = Installation.panel_width() + ph = Installation.panel_height() + rgb_size = num_panels * pw * ph * 3 + + frame_a = %RGBFrame{data: :binary.copy(<<77>>, rgb_size)} + frame_b = %RGBFrame{data: :binary.copy(<<123>>, rgb_size)} + + assert {:ok, "tcp://127.0.0.1:" <> _} = + Recording.start(sink_mod: Sink.Remote, sink_opts: [host: {127, 0, 0, 1}, port: port]) + + broadcast(frame_a) + broadcast(frame_b) + + # Closing the socket on stop lets the server's recv loop finish. + assert :ok = Recording.stop() + + assert_receive {:received, data}, 2000 + assert {:ok, header, records} = Format.parse(data) + assert header.num_panels == num_panels + + distinctive = + records + |> Enum.map(fn {_offset, d} -> d end) + |> Enum.filter(&(&1 in [frame_a.data, frame_b.data])) + + assert distinctive == [frame_a.data, frame_b.data] + end + + defp broadcast(frame) do + Phoenix.PubSub.broadcast(Octopus.PubSub, @mixer_topic, {:mixer, {:frame, frame}}) + end + + defp recv_all(sock, acc) do + case :gen_tcp.recv(sock, 0, 2000) do + {:ok, data} -> recv_all(sock, acc <> data) + {:error, _} -> acc + end + end +end diff --git a/octopus/test/octopus/recording/sink_gzip_test.exs b/octopus/test/octopus/recording/sink_gzip_test.exs new file mode 100644 index 00000000..0af0968a --- /dev/null +++ b/octopus/test/octopus/recording/sink_gzip_test.exs @@ -0,0 +1,59 @@ +defmodule Octopus.Recording.Sink.GzipTest do + use ExUnit.Case, async: true + + alias Octopus.Recording.Sink + + test "gzip sink produces a standard gzip file that round-trips the bytes" do + path = Path.join(System.tmp_dir!(), "gzip-sink-#{System.unique_integer([:positive])}.bin.gz") + on_exit(fn -> File.rm(path) end) + + payload = :binary.copy("the quick brown fox ", 5000) + + {:ok, sink} = + Sink.Gzip.open(inner_mod: Sink.File, inner_opts: [path: path], level: 6, flush_every: 3) + + {:ok, sink} = Sink.Gzip.write(sink, "HEADER") + + sink = + Enum.reduce(1..10, sink, fn _, s -> + {:ok, s} = Sink.Gzip.write(s, payload) + s + end) + + assert :ok = Sink.Gzip.close(sink) + + compressed = File.read!(path) + expected = "HEADER" <> :binary.copy(payload, 10) + + # Standard gzip: readable by :zlib.gunzip, and much smaller than the input. + assert :zlib.gunzip(compressed) == expected + assert byte_size(compressed) < byte_size(expected) + end + + describe "wrap/4" do + test "returns the sink unchanged when compression is disabled" do + assert Sink.Gzip.wrap(Sink.File, [path: "/x/y.octorec"], false) == + {Sink.File, [path: "/x/y.octorec"]} + end + + test "wraps a file sink and adds a .gz suffix to the path" do + assert {Sink.Gzip, opts} = Sink.Gzip.wrap(Sink.File, [path: "/x/y.octorec"], true, 4) + assert opts[:inner_mod] == Sink.File + assert opts[:inner_opts][:path] == "/x/y.octorec.gz" + assert opts[:level] == 4 + end + + test "does not double up the .gz suffix" do + assert {Sink.Gzip, opts} = Sink.Gzip.wrap(Sink.File, [path: "/x/y.octorec.gz"], true) + assert opts[:inner_opts][:path] == "/x/y.octorec.gz" + end + + test "wraps a remote sink without touching its options" do + assert {Sink.Gzip, opts} = + Sink.Gzip.wrap(Sink.Remote, [host: "h", port: 1], true) + + assert opts[:inner_mod] == Sink.Remote + assert opts[:inner_opts] == [host: "h", port: 1] + end + end +end diff --git a/octopus/test/octopus/recording/sink_remote_test.exs b/octopus/test/octopus/recording/sink_remote_test.exs new file mode 100644 index 00000000..dc7bf0ac --- /dev/null +++ b/octopus/test/octopus/recording/sink_remote_test.exs @@ -0,0 +1,46 @@ +defmodule Octopus.Recording.Sink.RemoteTest do + use ExUnit.Case, async: true + + alias Octopus.Recording.Sink.Remote + + test "streams written bytes to a TCP server verbatim" do + {:ok, listen} = + :gen_tcp.listen(0, [:binary, packet: :raw, active: false, reuseaddr: true]) + + {:ok, port} = :inet.port(listen) + test_pid = self() + + _acceptor = + spawn_link(fn -> + {:ok, sock} = :gen_tcp.accept(listen, 2000) + send(test_pid, {:received, recv_all(sock, <<>>)}) + end) + + assert {:ok, sink} = Remote.open(host: {127, 0, 0, 1}, port: port) + assert Remote.describe(sink) == "tcp://127.0.0.1:#{port}" + + assert {:ok, sink} = Remote.write(sink, "HEADER") + assert {:ok, sink} = Remote.write(sink, <<0, 1, 2, 3>>) + assert :ok = Remote.close(sink) + + assert_receive {:received, data}, 2000 + assert data == "HEADER" <> <<0, 1, 2, 3>> + end + + test "open/1 returns an error when the server is unreachable" do + # Grab a port, then immediately release it so nothing is listening. + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + :ok = :gen_tcp.close(listen) + + assert {:error, {:connect_failed, _reason}} = + Remote.open(host: {127, 0, 0, 1}, port: port, connect_timeout: 1000) + end + + defp recv_all(sock, acc) do + case :gen_tcp.recv(sock, 0, 2000) do + {:ok, data} -> recv_all(sock, acc <> data) + {:error, _} -> acc + end + end +end