From be78eec4021dbfe52ea2fc9eed1a903c64f14ea7 Mon Sep 17 00:00:00 2001 From: Michael B Reiser Date: Sun, 6 Sep 2026 13:46:50 -0400 Subject: [PATCH] feat(bridge): behavior_v2 run-log format, --convert CLI, log_control ack + hello levels (PR 1 of runlog-behavior-v2-plan) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bridge side of docs/development/runlog-behavior-v2-plan.md (Part 1). Nothing in the Studio or dashboard changes; old Studios never send behavior_v2 and ignore the new ack messages. - behavior_v2 (new default): the browser's arena_command echo (76 % of a v1 file) is written as ["a", t_off, dt, hex, status, rx_off(, error)] with ms offsets from the schema line's t0. Frame arrays are unchanged; runner / session / run_metadata / config lines stay verbatim objects. Lossless by construction: compact_arena_command verifies every invariant the expansion relies on (12-key set, dir, int t/rx_ms, spaced lowercase hex head, len, echo = command byte and ok = status===0 — or all three null on timeout) and writes a non-fitting echo verbatim instead. - bridge.py --convert IN OUT [--to v1|v2]: offline v1 ⇄ v2 re-encoding, direction auto-detected, .jsonl.gz on either side, strict (unknown arena key raises). Files without a v1 schema (pre-#140, full level) get a v2 schema with cols:null inserted; the reverse drops it. websockets import is now optional so --convert runs without the server deps. - log_control → log_control_ack {enabled, level, requested, file} with the level ACTUALLY in force (an unknown requested level is ignored, not applied); hello → hello_ack {bridge, levels, level, logging}. --log-level flag (--log-frames = alias for full). BRIDGE_VERSION 3.0. - tests/test-bridge-behavior.py: +99 checks — encode/decode invariants on real line samples (ok / status-1 / timeout / non-null error / other cmd), strict vs lenient fallbacks, file round trips (.jsonl + .gz), legacy path, LogWriter output per level, dispatcher acks. - scripts/runlog-v2-corpus.py: the corpus gate — round trip + known-shape check + size table over every log in a course-repo clone. 164/164 pass on cshl-2026-course origin/main (1281.6 MB v1 → 655.6 MB v2 → 211.3 MB v2.gz). Co-Authored-By: Claude Fable 5.1 --- docs/development/runlog-behavior-v2-plan.md | 49 +- fictrac-bridge/README.md | 76 ++- fictrac-bridge/bridge.py | 537 ++++++++++++++++++-- scripts/runlog-v2-corpus.py | 240 +++++++++ tests/test-bridge-behavior.py | 250 ++++++++- 5 files changed, 1076 insertions(+), 76 deletions(-) create mode 100755 scripts/runlog-v2-corpus.py diff --git a/docs/development/runlog-behavior-v2-plan.md b/docs/development/runlog-behavior-v2-plan.md index 04dfb2f..8d79b1d 100644 --- a/docs/development/runlog-behavior-v2-plan.md +++ b/docs/development/runlog-behavior-v2-plan.md @@ -1,8 +1,10 @@ -# Run-log format `behavior_v2` + large-file commit path — PLAN (not started) +# Run-log format `behavior_v2` + large-file commit path — PLAN (PR 1 bridge shipped; PRs 2–3 pending) Owner: Michael. Drafted 2026-09-06 from the analysis of rig03-sr run `rydc2tql` -(40 s trials, 51.4 MB, failed to auto-commit). Status: **approved direction, no code -yet**. Implementation lands as three PRs (bridge, Studio, dashboard) in that order. +(40 s trials, 51.4 MB, failed to auto-commit). Status: **PR 1 (bridge) implemented +2026-09-06** — see "PR 1 implementation notes" at the end; PR 2 (Studio) and PR 3 +(dashboard/readers) not started. Implementation lands as three PRs (bridge, Studio, +dashboard) in that order. ## Problem @@ -183,3 +185,44 @@ hour-long runs — hence Part 2. - Convert the existing course-repo logs to v2+gz, or leave history as is? - Default log level after the release: `behavior_v2` everywhere, including course benches (recommend yes; readers handle both). + +## PR 1 implementation notes (bridge, 2026-09-06) + +What shipped in `fictrac-bridge/bridge.py` (BRIDGE_VERSION 3.0), and the facts the +Studio (PR 2) and reader (PR 3) work must build on: + +- **The compact line is exactly** `["a", t_off, dt, hex, status, rx_off]` with a 7th + element only when v1 `error` is non-null. `t0` is the `ms` of the session line the + file opened with (`_open` emits session + schema together). The v1 object is + restored in the original key order `type, event, t, dt, len, head, status, echo, + ok, error, dir, rx_ms`. +- **Correction to the plan's derivation rule:** on a timeout the real v1 lines have + `status: null, echo: null, ok: null` (js/arena-session.js `_logCommand` sets all + three from the decoded reply or leaves all three null). So `expandV2Line` must + emit `echo = ok = null` when `status` is null, and `echo = command byte` + (`head` byte 1), `ok = (status === 0)` otherwise — NOT `ok: false` on timeout. + The corpus has 33 such lines (rig2 `spzae5dn`), all with a non-null `error`. +- **Lossless by construction, not by assumption:** `compact_arena_command` verifies + every invariant it later relies on (fixed 12-key set, `dir`, int `t`/`rx_ms`, + spaced lowercase hex `head` with `len` = byte count, `echo`/`ok` consistency). + A line that does not fit is written **verbatim** by the live bridge (stderr + warning) and **raises** under `--convert`/the corpus gate. Readers must therefore + accept a v1-shaped `arena_command` object inside a v2 file (e.g. a bulk command + whose `head` carries the ` …` truncation marker). +- **Files without a v1 schema line** (pre-#140 logs, `full` level) convert to v2 with + a schema line inserted after the first session line carrying `"cols": null` + (no positional frame rows in this file); the reverse drops it. Readers: `cols` + may be null. +- **Level negotiation:** `hello` → `hello_ack {bridge, levels:[behavior_v2, + behavior_v1, full], level, logging}`; `log_control` → `log_control_ack {enabled, + level, requested, file}` where `level` is the one actually in force (an unknown + `requested` is ignored, not applied). Old Studios ignore unknown message types + (`fictrac-bridge-client.js` dispatches only `frame`/`log_export_result`). Old + bridges never reply to `hello` — the Studio should treat "no hello_ack" as + "behavior_v1-only bridge". +- **`run_metadata.log_format`** (Part 1 §3) is a Studio-side field (the Studio + composes that line and now knows the acked level); the bridge does not inject it, + so the v1↔v2 round trip stays exact. +- **Corpus result (164 logs, 1.28 GB, origin/main of cshl-2026-course):** all pass; + totals in PR 1's description (1281.6 MB v1 → 655.6 MB v2 → 211.3 MB v2.gz at gzip + level 6). The 51 MB `rydc2tql` run → 20.4 MB v2 → 6.2 MB v2.gz. diff --git a/fictrac-bridge/README.md b/fictrac-bridge/README.md index 97b4ca5..adf21af 100644 --- a/fictrac-bridge/README.md +++ b/fictrac-bridge/README.md @@ -116,14 +116,24 @@ bridge → browser: {"type":"frame", "index":, "seq":, "t":, hardware clock — NANOSECONDS on our rigs — normalized here via FT_TS_NS_PER_MS. `ms` is the bridge wall-clock (display axis); `ft` is the velocity time base (per-frame differences, drop-safe). + {"type":"hello_ack", "bridge":, "levels":[…], "level":, + "logging":} + (reply to hello — the log levels this bridge can write, so the + browser can tell a stale bridge before a run; an old bridge + never replies to hello) + {"type":"log_control_ack", "enabled":, "level":, + "requested":, "file":} + (reply to log_control — `level` is the level ACTUALLY in force; + an unknown requested level is ignored and this is how you know) {"type":"log_export_result", "name":, "content":} (reply to log_export; {"error":} when nothing was written) browser → bridge: {"type":"hello", "client":"arena_console", "v":1} (on connect) {"type":"config", "fictrac_port":, "gain":, "offset":, "frames":} (any subset) - {"type":"log_control", "enabled":, "level":"behavior_v1"|"full"} + {"type":"log_control", "enabled":, + "level":"behavior_v2"|"behavior_v1"|"full"} (open the log file; level - picks the frame-row format) + picks the log format) {"type":"log", "event":, ...arbitrary fields, "ms":} {"type":"log_export"} (close the active log, stream it back whole) ``` @@ -135,25 +145,57 @@ re-binds the FicTrac input when `fictrac_port` changes. `log_control{enabled:tru **starts a new timestamped log file** and re-zeroes the behavior_v1 `ms`/`ft` clocks (false closes it; `--log-dir` picks where on-demand files land, default CWD). The log is **uniform NDJSON** — one JSON value per line; a reader parses each line -and dispatches on `Array.isArray` (frame array vs event object). While logging is -active the bridge records: - -- a one-time schema line `{"type":"frame_schema","level":"behavior_v1", - "cols":["ms","fc","idx","ft","x","y","hd"]}`, then **every** FicTrac record it +and dispatches on `Array.isArray` (positional array vs event object), then on +`arr[0]` (`"a"` = arena echo, a number = frame). While logging is active the bridge +records: + +- a one-time schema line — `{"type":"frame_schema","level":"behavior_v2", + "cols":["ms","fc","idx","ft","x","y","hd"],"arena_cols":["t_off","dt","hex", + "status","rx_off"],"t0":}` (default) or the `behavior_v1` form without + `arena_cols`/`t0` — then **every** FicTrac record it receives (before WS coalescing) as the positional array `[ms, fc, idx, ft, x, y, hd]` — `ms` bridge-relative ms, `fc` FicTrac frame counter (col 1), `idx` displayed arena index, `ft` FicTrac timestamp (col 22) as relative ms (**not** col-24 dt, which can't recover elapsed time across a dropped frame), `x`/`y`/`hd` integrated position + heading (rad, 5-decimal). The live scope + offline dashboard recompute every derived channel (turning/forward/side/speed/dir) from this via - `js/kinematics.js`. The **browser picks the level** per run via `log_control`'s - `level` (Arena Studio's runner asserts the level chosen in File ▾ → Run logging, - default `behavior_v1`, overriding `--log-frames`) — `--log-frames` only sets the - launch default. `full` logs the whole 25-column record - (`{"type":"fictrac_frame", ..., "fictrac":[…25…]}`) for debug/archival. -- inbound browser `log` messages (e.g. `{"event":"arena_command", ...}` for every - Web Serial command, or Arena Studio's `{"event":"run_metadata", ...}` header - line at recorded-run start), each stamped with `dir` and `rx_ms`. + `js/kinematics.js`. The frame array is identical in `behavior_v1` and `behavior_v2`. + The **browser picks the level** per run via `log_control`'s `level` (Arena + Studio's runner asserts the level chosen in File ▾ → Run logging, overriding + `--log-level`); the bridge answers with `log_control_ack` naming the level it + will actually write. `full` logs the whole 25-column record + (`{"type":"fictrac_frame", ..., "fictrac":[…25…]}`) for debug/archival, with no + schema line. +- inbound browser `log` messages (e.g. Arena Studio's `{"event":"run_metadata", ...}` + header line at recorded-run start), each stamped with `dir` and `rx_ms`, as + verbatim JSON objects. +- the browser's `{"event":"arena_command", ...}` echo of every Web Serial command + (one per closed-loop 0x70 frame command, ~100 Hz — 76 % of a `behavior_v1` file's + bytes). Under **`behavior_v2`** each becomes the compact array + `["a", t_off, dt, hex, status, rx_off]` (+ a 7th `error` string when non-null): + `t_off`/`rx_off` are ms offsets from the schema line's `t0`, `hex` is the `head` + bytes without spaces, `status` is the reply status or `null` on timeout. The + constant/derivable v1 fields (`type`, `event`, `dir`, `len`, `echo` = the command + byte, `ok` = `status === 0`; all three of `status`/`echo`/`ok` are `null` when no + reply decoded) are restored on expansion — **lossless**, verified per line: an + echo that does not fit the fixed shape is written verbatim instead. Measured on + the course corpus (164 logs, 1.28 GB): v2 is 0.51× the v1 bytes overall and 0.38× + on closed-loop P3 runs; v2.gz is 0.17× overall. + `behavior_v1` writes the echo as the full object (the pre-2026-09 format). + +**Converting existing files** (migration + testing readers on real data before a rig +produces v2), no sockets needed: + +```bash +pixi run bridge -- --convert runlogs/rig1/run.jsonl run.v2.jsonl.gz # v1 → v2 (+gzip) +pixi run bridge -- --convert run.v2.jsonl.gz run.v1.jsonl # and back +``` + +Direction is auto-detected from the `frame_schema` line (`--to v1|v2` forces it); +`.gz` on either side is handled. The conversion is strict — an `arena_command` with +an unexpected key set aborts instead of dropping a field. `tests/test-bridge-behavior.py` +holds the round-trip unit tests; `scripts/runlog-v2-corpus.py` runs the same round +trip over every log in a course-repo clone and prints the size table. `log_export` (Arena Studio's course pipeline) **closes** the active log — guaranteeing complete, flushed content — and streams the whole file back to the @@ -191,7 +233,9 @@ sends it automatically when you load a Mode-3 pattern. | `--gain` | `1.8` | Degrees of heading per frame index (360/200); negative reverses. Re-settable live. | | `--offset` | `0.0` | Heading offset in degrees. | | `--log PATH` | on demand | Append log events (JSONL). If unset, opened when the browser enables logging. | -| `--log-frames` | off | Log the FULL 25-column FicTrac record per frame (debug/archival) instead of the default compact `behavior_v1` array `[ms,fc,idx,ft,x,y,hd]`. | +| `--log-level {behavior_v2,behavior_v1,full}` | `behavior_v2` | Launch default for the log format; the browser's `log_control` overrides it per run (acknowledged in `log_control_ack`). | +| `--log-frames` | off | Alias for `--log-level full` (the 25-column FicTrac record per frame, debug/archival). | +| `--convert IN OUT [--to v1\|v2]` | — | Offline: re-encode a run log v1 ⇄ v2 (`.jsonl` or `.jsonl.gz` either side) and exit. | ## Replaying a recorded FicTrac log diff --git a/fictrac-bridge/bridge.py b/fictrac-bridge/bridge.py index b6ec277..a703ab0 100644 --- a/fictrac-bridge/bridge.py +++ b/fictrac-bridge/bridge.py @@ -22,32 +22,53 @@ "x":, "y":, "hd":} (the `behavior_v1` fields drive the live oscilloscope; the legacy index/seq/t keys are kept for back-compatibility) + {"type":"hello_ack", "bridge":, "levels":[...], + "level":, "logging":} + (reply to hello — advertises the log levels this bridge can + write so the browser can tell a stale bridge from a current one) + {"type":"log_control_ack", "enabled":, "level":, + "requested":, "file":} + (reply to log_control — `level` is the level ACTUALLY in force; + an unknown requested level is ignored, and this is how the + browser finds out) {"type":"log_export_result", "name":, "content":} (reply to log_export; {"error":} when nothing was written) browser → bridge: {"type":"hello", "client":"arena_console", "v":1} (on connect) {"type":"config", "fictrac_port":, "gain":, "offset":, "frames":} (any subset) - {"type":"log_control", "enabled":, "level":"behavior_v1"|"full"} - (open/close the log file; level picks the frame-row format, - overriding --log-frames — the runner asserts behavior_v1) + {"type":"log_control", "enabled":, + "level":"behavior_v2"|"behavior_v1"|"full"} + (open/close the log file; level picks the log format, + overriding --log-level — the runner asserts it per run) {"type":"log", "event":, ...arbitrary, "ms":} {"type":"log_export"} (close the active log, stream it back whole) The FicTrac → frame-index policy lives in frame_index_from_fictrac(); edit that one function to change closed-loop behaviour. -FRAME LOGGING has two levels (issue #140), both uniform NDJSON — a reader does one -JSON.parse() per line and dispatches on Array.isArray (frame array vs event object): - - behavior_v1 (DEFAULT): a one-time {"type":"frame_schema","level":"behavior_v1", - "cols":["ms","fc","idx","ft","x","y","hd"]} header, then each frame as the - positional array [ms, fc, idx, ft, x, y, hd]. Compact behavioral state the live - scope + offline dashboard recompute all derived channels from (see js/kinematics.js). - ft = FicTrac col-22 timestamp as relative ms — NOT col-24 dt, which cannot - recover elapsed time across a frame dropped before logging (Frank, #143). - col 22 is the camera hardware clock in ns on our rigs, normalized to ms here - (FT_TS_NS_PER_MS); downstream dt is per-frame ft differences (variable-rate safe). - - full (--log-frames): the whole 25-column record under a "fictrac" key (debug/archival). -Session/runner events stay JSON objects on their own lines. `minimal` and gzip are deferred. +LOG LEVELS — three, all uniform NDJSON: a reader does one JSON.parse() per line and +dispatches on Array.isArray (positional array vs event object), then on arr[0]: + - behavior_v2 (DEFAULT, docs/development/runlog-behavior-v2-plan.md Part 1): a one-time + {"type":"frame_schema","level":"behavior_v2","cols":[...], + "arena_cols":["t_off","dt","hex","status","rx_off"],"t0":} header, then + each FicTrac frame as the SAME positional array as behavior_v1 (below), and each + browser `arena_command` echo (one per closed-loop 0x70 frame command, ~100 Hz — 76 % + of a v1 file) as the compact array ["a", t_off, dt, hex, status, rx_off(, error)] + with ms offsets from t0 (= the session line's ms). LOSSLESS: expand_arena_command() + rebuilds the exact v1 object (see compact_arena_command for the invariants that are + verified per line; an echo that does not fit is written verbatim, never dropped). + Runner / session / run_metadata / config lines stay verbatim JSON objects. + - behavior_v1: the v2 frame_schema minus arena_cols/t0, positional frame arrays + [ms, fc, idx, ft, x, y, hd], and arena_command echoes as full JSON objects. + Compact behavioral state the live scope + offline dashboard recompute all derived + channels from (see js/kinematics.js). ft = FicTrac col-22 timestamp as relative + ms — NOT col-24 dt, which cannot recover elapsed time across a frame dropped before + logging (Frank, #143). col 22 is the camera hardware clock in ns on our rigs, + normalized to ms here (FT_TS_NS_PER_MS); downstream dt is per-frame ft differences. + - full (--log-level full / --log-frames): the whole 25-column record under a + "fictrac" key (debug/archival); no schema line; arena echoes verbatim. +`bridge.py --convert IN OUT` re-encodes an existing file v1 ⇄ v2 (direction auto- +detected, `.gz` in/out handled) — the migration + reader-test path. """ from __future__ import annotations @@ -62,8 +83,16 @@ import sys import time -from websockets.asyncio.server import serve -from websockets.exceptions import ConnectionClosed +try: + from websockets.asyncio.server import serve + from websockets.exceptions import ConnectionClosed +except ImportError: # --convert and the offline tests need none of the server + serve = None + + class ConnectionClosed(Exception): # type: ignore[no-redef] + pass + +import gzip # Inbound WebSocket message cap. The library default is 1 MiB — raised so a # multi-MB payload never kills the socket; sized to match the log_export @@ -75,13 +104,32 @@ # leads the startup banner. (An OLD bridge has no --version flag → argparse errors, # which is itself the tell.) "behavior_v1" here means frames carry ms/fc/idx/ft/x/y/hd # with `ft` normalized ns→ms — i.e. the live scope + dashboard will work. -BRIDGE_VERSION = "2.0 · behavior_v1 (ns→ms ft, x/y/hd frames)" +BRIDGE_VERSION = "3.0 · behavior_v2 (compact arena echo, log_control ack)" -# behavior_v1 — the default logged frame schema (issue #140). Positional-array -# rows in this column order; the live scope + offline dashboard recompute every -# derived channel (turning/forward/side/speed/dir) from this compact state. +# behavior_v1 — the logged frame schema (issue #140), UNCHANGED in behavior_v2. +# Positional-array rows in this column order; the live scope + offline dashboard +# recompute every derived channel (turning/forward/side/speed/dir) from this state. BEHAVIOR_V1_COLS = ["ms", "fc", "idx", "ft", "x", "y", "hd"] +# Log levels the bridge can write, most-preferred first. Advertised in hello_ack +# so a browser can detect a stale bridge; the first entry is the fresh-process +# default (log_control overrides it per run). +LOG_LEVELS = ("behavior_v2", "behavior_v1", "full") +DEFAULT_LOG_LEVEL = LOG_LEVELS[0] + +# behavior_v2 compact arena-command echo: ["a", t_off, dt, hex, status, rx_off] with +# an optional 7th element carrying a non-null `error` string. Offsets are ms from +# the schema line's t0. Declared in the v2 frame_schema line as `arena_cols`. +BEHAVIOR_V2_ARENA_COLS = ["t_off", "dt", "hex", "status", "rx_off"] +ARENA_TAG = "a" +ARENA_DIR = "browser→bridge" +# The exact v1 arena_command object (js/arena-session.js _logCommand + write_inbound +# stamps). Key ORDER is what the browser/bridge wrote; the SET is what must match +# for a line to be compacted — an extra key means a newer producer, so the line is +# kept verbatim (or raises under strict conversion) rather than losing the field. +ARENA_COMMAND_KEYS = ("type", "event", "t", "dt", "len", "head", "status", "echo", "ok", "error", "dir", "rx_ms") +_ARENA_COMMAND_KEYSET = frozenset(ARENA_COMMAND_KEYS) + # FicTrac col-22 is the camera's hardware-clock timestamp. Our rigs run identical # cameras + software that emit it in NANOSECONDS (FicTrac's docs nominally call it # ms, but this hardware clock is ns). behavior_v1's `ft` is defined as @@ -123,6 +171,293 @@ def behavior_v1_row(fields: list[float], index: int, rel_ms: int, ft0: float | N "hd": round(fields[16], 5), } +# ───────────────────────────────────────────────────────────────────────────── +# behavior_v2 line format — pure functions (no clocks, no I/O; offline-tested by +# tests/test-bridge-behavior.py and gated over the whole course corpus by +# scripts/runlog-v2-corpus.py). Both the live LogWriter and --convert use these. +# ───────────────────────────────────────────────────────────────────────────── +class RunlogFormatError(ValueError): + """A line does not fit the format it claims (raised only under strict conversion; + the live writer falls back to a verbatim object instead).""" + + +def _is_int(v) -> bool: + return isinstance(v, int) and not isinstance(v, bool) + + +def _is_number(v) -> bool: + return isinstance(v, (int, float)) and not isinstance(v, bool) + + +def _hex_bytes(head: str) -> list[str] | None: + """Split a v1 `head` ("03 70 2e 00") into 2-digit lowercase hex bytes, or None + when it is not exactly that shape (e.g. the ' …' truncation marker, uppercase).""" + if not isinstance(head, str) or not head: + return None + parts = head.split(" ") + for p in parts: + if len(p) != 2 or p.lower() != p or any(c not in "0123456789abcdef" for c in p): + return None + return parts + + +def frame_schema_line(level: str, cols: list[str] | None, t0: int | None = None) -> dict: + """The one-time schema header for a behavior_v1 / behavior_v2 log.""" + if level == "behavior_v2": + return { + "type": "frame_schema", + "level": "behavior_v2", + "cols": cols, + "arena_cols": list(BEHAVIOR_V2_ARENA_COLS), + "t0": t0, + } + return {"type": "frame_schema", "level": level, "cols": cols} + + +def arena_compact_reason(obj, t0: int) -> str | None: + """Why `obj` can NOT be written as a compact "a" array (None = it can). + + These are the invariants expand_arena_command() relies on to rebuild the v1 + object exactly: fixed key set, dir, int t/rx_ms, numeric dt, spaced lowercase + hex head with len == byte count, status int-or-null, echo == the command byte + (head[1]) and ok == (status == 0) — or ALL THREE null when there was no reply + (timeout / undecodable), error str-or-null.""" + if not isinstance(obj, dict): + return "not an object" + if obj.get("type") != "log" or obj.get("event") != "arena_command": + return "not an arena_command" + keys = set(obj) + if keys != _ARENA_COMMAND_KEYSET: + return f"unexpected key set (diff {sorted(keys ^ _ARENA_COMMAND_KEYSET)})" + if obj["dir"] != ARENA_DIR: + return f"dir {obj['dir']!r}" + if not _is_int(obj["t"]) or not _is_int(obj["rx_ms"]) or not _is_int(t0): + return "t/rx_ms/t0 not integers" + if not _is_number(obj["dt"]): + return "dt not a number" + parts = _hex_bytes(obj["head"]) + if parts is None: + return f"head {obj['head']!r} not spaced lowercase hex bytes" + if obj["len"] != len(parts): + return f"len {obj['len']!r} != {len(parts)} head bytes" + status, echo, ok, error = obj["status"], obj["echo"], obj["ok"], obj["error"] + if status is None: + if echo is not None or ok is not None: + return "status null but echo/ok not null" + else: + if not _is_int(status): + return f"status {status!r} not int/null" + if len(parts) < 2 or echo != int(parts[1], 16): + return f"echo {echo!r} != command byte" + if not isinstance(ok, bool) or ok != (status == 0): + return f"ok {ok!r} != (status == 0)" + if error is not None and not isinstance(error, str): + return f"error {error!r} not str/null" + return None + + +def compact_arena_command(obj: dict, t0: int, strict: bool = False): + """v1 arena_command object → ["a", t_off, dt, hex, status, rx_off(, error)]. + + Returns None when the object does not fit (the live writer then emits it + verbatim); with strict=True raises RunlogFormatError instead.""" + reason = arena_compact_reason(obj, t0) + if reason is not None: + if strict: + raise RunlogFormatError(f"arena_command not compactable: {reason}") + return None + arr = [ + ARENA_TAG, + obj["t"] - t0, + obj["dt"], + obj["head"].replace(" ", ""), + obj["status"], + obj["rx_ms"] - t0, + ] + if obj["error"] is not None: + arr.append(obj["error"]) + return arr + + +def is_arena_array(value) -> bool: + """True for a behavior_v2 compact arena echo (vs a frame array, whose [0] is ms).""" + return isinstance(value, list) and len(value) in (6, 7) and value[0] == ARENA_TAG + + +def expand_arena_command(arr: list, t0: int) -> dict: + """["a", t_off, dt, hex, status, rx_off(, error)] → the exact v1 object, in the + key order the browser + bridge wrote it. Raises RunlogFormatError if malformed.""" + if not is_arena_array(arr): + raise RunlogFormatError(f"not a compact arena array: {arr!r}") + _, t_off, dt, hexs, status, rx_off = arr[:6] + error = arr[6] if len(arr) == 7 else None + if not _is_int(t_off) or not _is_int(rx_off) or not _is_int(t0): + raise RunlogFormatError("t_off/rx_off/t0 not integers") + if not _is_number(dt): + raise RunlogFormatError("dt not a number") + if not isinstance(hexs, str) or len(hexs) % 2 or _hex_bytes(" ".join(hexs[i : i + 2] for i in range(0, len(hexs), 2)) or "") is None: + raise RunlogFormatError(f"hex {hexs!r} malformed") + if status is not None and not _is_int(status): + raise RunlogFormatError(f"status {status!r} not int/null") + if error is not None and not isinstance(error, str): + raise RunlogFormatError(f"error {error!r} not str") + parts = [hexs[i : i + 2] for i in range(0, len(hexs), 2)] + if status is not None and len(parts) < 2: + raise RunlogFormatError("status present but no command byte") + return { + "type": "log", + "event": "arena_command", + "t": t0 + t_off, + "dt": dt, + "len": len(parts), + "head": " ".join(parts), + "status": status, + "echo": None if status is None else int(parts[1], 16), + "ok": None if status is None else status == 0, + "error": error, + "dir": ARENA_DIR, + "rx_ms": t0 + rx_off, + } + + +def _is_schema(obj) -> bool: + return isinstance(obj, dict) and obj.get("type") == "frame_schema" + + +def detect_format(objs) -> str: + """'behavior_v2' | 'behavior_v1' | 'full' | 'legacy' for a parsed log (list of + JSON values). The schema line decides; without one, "a" arrays mean v2, + fictrac_frame objects carrying the 25-column `fictrac` array mean the `full` + level, anything else is a pre-#140 log.""" + saw_full = False + for o in objs: + if _is_schema(o): + return str(o.get("level")) + if is_arena_array(o): + return "behavior_v2" + if isinstance(o, dict) and o.get("type") == "fictrac_frame" and "fictrac" in o: + saw_full = True + return "full" if saw_full else "legacy" + + +def convert_v1_to_v2(objs, strict: bool = True) -> list: + """Re-encode a parsed v1 (or legacy/full) log as behavior_v2. Lossless: only the + frame_schema line changes and arena_command objects become "a" arrays; every + other line is passed through untouched. t0 = the first session line's `ms`. + A file without a v1 schema line (pre-#140 / full level) gets a v2 schema line + inserted after its first session line with "cols": null (no positional frame + rows in this file) — convert_v2_to_v1 drops that line again.""" + schema_idx = next((i for i, o in enumerate(objs) if _is_schema(o)), None) + cols = None + if schema_idx is not None: + sch = objs[schema_idx] + if sch.get("level") == "behavior_v2": + raise RunlogFormatError("already behavior_v2") + if sch.get("level") != "behavior_v1" or set(sch) != {"type", "level", "cols"}: + raise RunlogFormatError(f"unexpected v1 frame_schema {sch!r}") + cols = sch["cols"] + t0 = None + for o in objs: + if isinstance(o, dict) and o.get("type") == "session" and _is_int(o.get("ms")): + t0 = o["ms"] + break + if t0 is None: + first_a = next((o for o in objs if isinstance(o, dict) and o.get("event") == "arena_command"), None) + t0 = first_a["t"] if first_a and _is_int(first_a.get("t")) else 0 + if schema_idx is not None: + insert_at = schema_idx + else: + insert_at = 1 if objs and isinstance(objs[0], dict) and objs[0].get("type") == "session" else 0 + schema = frame_schema_line("behavior_v2", cols, t0) + out = [] + for i, o in enumerate(objs): + if i == schema_idx: + out.append(schema) + continue + if schema_idx is None and i == insert_at: + out.append(schema) + if isinstance(o, dict) and o.get("event") == "arena_command" and o.get("type") == "log": + arr = compact_arena_command(o, t0, strict=strict) + out.append(arr if arr is not None else o) + else: + out.append(o) + if schema_idx is None and insert_at >= len(objs): + out.append(schema) + return out + + +def convert_v2_to_v1(objs) -> list: + """Inverse of convert_v1_to_v2: "a" arrays → v1 arena_command objects, the v2 + schema → the v1 schema (or dropped when cols is null). Everything else verbatim.""" + schema_idx = next((i for i, o in enumerate(objs) if _is_schema(o)), None) + if schema_idx is None or objs[schema_idx].get("level") != "behavior_v2": + raise RunlogFormatError("not a behavior_v2 log (no behavior_v2 frame_schema line)") + sch = objs[schema_idx] + if set(sch) != {"type", "level", "cols", "arena_cols", "t0"}: + raise RunlogFormatError(f"unexpected v2 frame_schema keys {sorted(sch)}") + if sch["arena_cols"] != BEHAVIOR_V2_ARENA_COLS: + raise RunlogFormatError(f"unknown arena_cols {sch['arena_cols']!r}") + t0 = sch["t0"] + out = [] + for i, o in enumerate(objs): + if i == schema_idx: + if sch["cols"] is not None: + out.append(frame_schema_line("behavior_v1", sch["cols"])) + continue + if is_arena_array(o): + out.append(expand_arena_command(o, t0)) + else: + out.append(o) + return out + + +def canonical_json(obj) -> str: + """Key-sorted compact JSON — the equality used by the round-trip tests.""" + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def read_jsonl(path: str) -> list: + """Parse a .jsonl or .jsonl.gz log into a list of JSON values (blank lines skipped).""" + opener = gzip.open if path.endswith(".gz") else open + with opener(path, "rt", encoding="utf-8") as fh: + return [json.loads(line) for line in fh if line.strip()] + + +def dumps_jsonl(objs) -> str: + return "".join(json.dumps(o, separators=(",", ":")) + "\n" for o in objs) + + +def write_jsonl(path: str, objs) -> int: + """Write compact NDJSON (gzip when the name ends in .gz, mtime 0 so the bytes are + reproducible). Returns the number of bytes written.""" + data = dumps_jsonl(objs).encode("utf-8") + if path.endswith(".gz"): + with open(path, "wb") as raw, gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz: + gz.write(data) + return os.path.getsize(path) + with open(path, "wb") as fh: + fh.write(data) + return len(data) + + +def convert_file(src: str, dst: str, to: str | None = None) -> dict: + """--convert: re-encode a run log v1 ⇄ v2. `to` = 'v1' | 'v2' | None (auto: the + opposite of what `src` is). Returns a small report dict.""" + objs = read_jsonl(src) + fmt = detect_format(objs) + if to is None: + to = "v1" if fmt == "behavior_v2" else "v2" + if to == "v2": + if fmt == "behavior_v2": + raise RunlogFormatError(f"{src} is already behavior_v2") + out = convert_v1_to_v2(objs, strict=True) + elif to == "v1": + out = convert_v2_to_v1(objs) + else: + raise ValueError(f"--to must be v1 or v2, not {to!r}") + n = write_jsonl(dst, out) + return {"src": src, "dst": dst, "from": fmt, "to": to, "lines": len(out), "bytes": n} + # ───────────────────────────────────────────────────────────────────────────── # Processing policy — THE part you customise. @@ -217,12 +552,18 @@ class LogWriter: On-demand files land in --log-dir (default: the process CWD). """ - def __init__(self, path: str | None, log_frames: bool, log_dir: str | None = None) -> None: + def __init__(self, path: str | None, level: str | bool = DEFAULT_LOG_LEVEL, log_dir: str | None = None) -> None: self._explicit = path # fixed --log path (standalone), else None self._dir = log_dir or "" self._fh = None self._name: str | None = None # current-or-most-recent file (export target) - self.log_frames = log_frames + self._t0: int | None = None # behavior_v2 offset base = the open session line's ms + self._verbatim_warned = 0 + # `level` accepts the legacy log_frames bool (True → 'full') for callers that + # predate --log-level. + self.level = "full" if level is True else (DEFAULT_LOG_LEVEL if level is False else level) + if self.level not in LOG_LEVELS: + raise ValueError(f"unknown log level {level!r} (one of {LOG_LEVELS})") if self._dir: os.makedirs(self._dir, exist_ok=True) if path: @@ -232,21 +573,41 @@ def __init__(self, path: str | None, log_frames: bool, log_dir: str | None = Non def active(self) -> bool: return self._fh is not None - def set_level(self, level: str) -> None: - """Select the frame-logging level for the NEXT log file (browser-driven, - overriding the --log-frames launch flag): 'full' = 25-column record, - anything else = the compact behavior_v1 array.""" - self.log_frames = level == "full" + @property + def current_name(self) -> str | None: + """Basename of the current-or-most-recent log file (for log_control_ack).""" + return os.path.basename(self._name) if self._name else None + + @property + def log_frames(self) -> bool: + """Legacy alias: True when the level is `full` (25-column FicTrac records).""" + return self.level == "full" + + def set_level(self, level: str) -> bool: + """Select the log level for the NEXT log file (browser-driven, overriding the + --log-level launch flag). Returns False — and leaves the level unchanged — for + a level this bridge does not know; the dispatcher reports the level actually + in force back to the browser in log_control_ack (a stale bridge used to + ignore an unknown level silently).""" + if level not in LOG_LEVELS: + return False + self.level = level + return True def _open(self, name: str, event: str) -> None: self._fh = open(name, "a", buffering=1, encoding="utf-8") self._name = name - self._emit({"type": "session", "event": event, "file": name, "ms": now_ms()}) - # behavior_v1 logs lead with a one-time schema line so the positional - # frame arrays are self-describing (full mode stays keyed objects). - if not self.log_frames: - self._emit({"type": "frame_schema", "level": "behavior_v1", "cols": BEHAVIOR_V1_COLS}) - print(f"[log] writing to {name}", file=sys.stderr) + ms = now_ms() + self._t0 = ms + self._emit({"type": "session", "event": event, "file": name, "ms": ms}) + # behavior_v1/v2 logs lead with a one-time schema line so the positional + # arrays are self-describing (full mode stays keyed objects). v2 also + # carries the compact arena-echo layout + t0 (= this session line's ms). + if self.level == "behavior_v2": + self._emit(frame_schema_line("behavior_v2", BEHAVIOR_V1_COLS, ms)) + elif self.level == "behavior_v1": + self._emit(frame_schema_line("behavior_v1", BEHAVIOR_V1_COLS)) + print(f"[log] writing to {name} ({self.level})", file=sys.stderr) def start_new_log(self) -> None: """Begin a fresh timestamped log file — one per logging activation. @@ -294,17 +655,30 @@ def write_inbound(self, raw: str | bytes) -> None: obj = json.loads(raw) except json.JSONDecodeError: obj = {"type": "log", "event": "unparsed", "raw": raw} - obj.setdefault("dir", "browser→bridge") + obj.setdefault("dir", ARENA_DIR) obj.setdefault("rx_ms", now_ms()) + if self.level == "behavior_v2" and obj.get("event") == "arena_command": + # Compact echo (76 % of a v1 file). An echo that does not fit the fixed + # shape is written verbatim — lossless either way, never dropped. + arr = compact_arena_command(obj, self._t0) + if arr is not None: + self._emit(arr) + return + self._verbatim_warned += 1 + if self._verbatim_warned <= 3: + print( + f"[log] arena_command kept verbatim: {arena_compact_reason(obj, self._t0)}", + file=sys.stderr, + ) self._emit(obj) def write_frame(self, beh: dict, fields: list[float]) -> None: # Store EVERY received FicTrac frame whenever logging is active — regardless - # of whether the browser is applying frames. Default = behavior_v1 positional - # array; --log-frames = the full 25-field record (debug/archival). + # of whether the browser is applying frames. behavior_v1/v2 = the positional + # array (identical in both); full = the 25-field record (debug/archival). if not self._fh: return - if self.log_frames: + if self.level == "full": rec = {"type": "fictrac_frame", "seq": beh["fc"], "index": beh["idx"], "t": beh["ms"]} rec["fictrac"] = fields self._emit(rec) @@ -484,6 +858,16 @@ async def consume(queue: asyncio.Queue, pipeline: Pipeline) -> None: def make_dispatcher(pipeline: Pipeline, log: LogWriter, inputs: InputManager): """Build the async handler for inbound browser messages.""" + async def reply(websocket, msg: dict) -> None: + """Answer the ASKING client only (the hub's broadcast path coalesces to the + newest frame and would drop a one-shot reply).""" + if websocket is None: + return + try: + await websocket.send(json.dumps(msg)) + except ConnectionClosed: + pass + async def dispatch(raw, websocket=None) -> None: if isinstance(raw, bytes): raw = raw.decode("utf-8", errors="replace") @@ -492,9 +876,23 @@ async def dispatch(raw, websocket=None) -> None: except json.JSONDecodeError: log.write_inbound(raw) return - kind = obj.get("type") + kind = obj.get("type") if isinstance(obj, dict) else None - if kind == "config": + if kind == "hello": + # Advertise what this bridge can write so the browser can detect a stale + # bridge BEFORE a run (an old bridge never replies to hello at all). + log.write_inbound(raw) + await reply( + websocket, + { + "type": "hello_ack", + "bridge": BRIDGE_VERSION, + "levels": list(LOG_LEVELS), + "level": log.level, + "logging": log.active, + }, + ) + elif kind == "config": applied = {} if obj.get("gain") is not None: pipeline.gain = float(obj["gain"]) @@ -511,27 +909,37 @@ async def dispatch(raw, websocket=None) -> None: print(f"[cfg] applied {applied}", file=sys.stderr) log.write_inbound(raw) elif kind == "log_control": + requested = obj.get("level") if obj.get("enabled"): - if obj.get("level") in ("behavior_v1", "full"): - log.set_level(obj["level"]) # browser asserts the level per run - pipeline.reset_base() # zero behavior_v1 ms/ft at the run boundary + if requested is not None and not log.set_level(requested): + # Browser asserts the level per run; an unknown one is NOT applied. + # The ack below carries the level actually in force so the Studio + # can warn ("bridge too old for X") instead of finding out later. + print(f"[log] unknown log level {requested!r} requested; keeping {log.level}", file=sys.stderr) + pipeline.reset_base() # zero behavior ms/ft at the run boundary log.start_new_log() # fresh timestamped file per activation log.write_inbound(raw) else: log.write_inbound(raw) log.close() + await reply( + websocket, + { + "type": "log_control_ack", + "enabled": log.active, + "level": log.level, + "requested": requested, + "file": log.current_name, + }, + ) elif kind == "log_export": # Close + stream the whole log back to the ASKING client only. name, content = log.export_current() if name is not None: - reply = {"type": "log_export_result", "name": name, "content": content} + reply_msg = {"type": "log_export_result", "name": name, "content": content} else: - reply = {"type": "log_export_result", "error": "no log file has been written"} - if websocket is not None: - try: - await websocket.send(json.dumps(reply)) - except ConnectionClosed: - pass + reply_msg = {"type": "log_export_result", "error": "no log file has been written"} + await reply(websocket, reply_msg) print( f"[log] export → {name or 'nothing'}" + (f" ({len(content)} chars)" if content else ""), @@ -545,7 +953,9 @@ async def dispatch(raw, websocket=None) -> None: async def run(args: argparse.Namespace) -> None: - log = LogWriter(args.log, args.log_frames, args.log_dir) + if serve is None: + raise SystemExit("the `websockets` package is required to serve (pixi install); --convert works without it") + log = LogWriter(args.log, args.log_level, args.log_dir) queue: asyncio.Queue[str] = asyncio.Queue() # pipeline ↔ hub is a cycle (hub's dispatcher reconfigures the pipeline; the # pipeline publishes to the hub), so build the pipeline first and wire the hub in. @@ -569,7 +979,7 @@ async def run(args: argparse.Namespace) -> None: print( f"[ws] serving ws://{args.ws_host}:{args.ws_port} " f"(bridge {BRIDGE_VERSION}; proto={args.proto}, fictrac_port={args.in_port}, " - f"frames={args.frames}, gain={args.gain:g}, log={args.log or 'on-demand'})", + f"frames={args.frames}, gain={args.gain:g}, log={args.log or 'on-demand'}, level={log.level})", file=sys.stderr, ) await stop.wait() @@ -599,11 +1009,32 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("--offset", type=float, default=0.0, help="heading offset in degrees (default: 0.0)") p.add_argument("--log", default=None, help="append browser log events (JSONL) to this file (else opened on demand)") p.add_argument("--log-dir", default=None, help="directory for on-demand arena-log-*.jsonl files (default: CWD; created if missing)") - p.add_argument("--log-frames", action="store_true", help="log the FULL 25-column FicTrac record per frame (debug/archival); default logs the compact behavior_v1 array [ms,fc,idx,ft,x,y,hd]") + p.add_argument("--log-level", choices=LOG_LEVELS, default=None, help=f"launch default for the log format, overridden per run by the browser's log_control (default: {DEFAULT_LOG_LEVEL}). behavior_v2 = compact arena echoes; behavior_v1 = the pre-2026-09 format; full = the 25-column FicTrac record per frame") + p.add_argument("--log-frames", action="store_true", help="alias for --log-level full (debug/archival)") + p.add_argument("--convert", nargs=2, metavar=("IN", "OUT"), help="offline: re-encode a run log v1 ⇄ v2 (direction auto-detected; .jsonl or .jsonl.gz either side) and exit — no sockets") + p.add_argument("--to", choices=("v1", "v2"), default=None, help="with --convert: force the target format instead of auto-detecting") args = p.parse_args(argv) + if args.convert: + src, dst = args.convert + try: + rep = convert_file(src, dst, args.to) + except (RunlogFormatError, OSError, json.JSONDecodeError) as exc: + print(f"[convert] FAILED: {exc}", file=sys.stderr) + return 1 + print( + f"[convert] {rep['src']} ({rep['from']}) → {rep['dst']} ({rep['to']}): " + f"{rep['lines']} lines, {rep['bytes']} bytes", + file=sys.stderr, + ) + return 0 + if args.frames <= 0: p.error("--frames must be > 0") + if args.log_frames and args.log_level not in (None, "full"): + p.error("--log-frames conflicts with --log-level") + if args.log_level is None: + args.log_level = "full" if args.log_frames else DEFAULT_LOG_LEVEL try: asyncio.run(run(args)) diff --git a/scripts/runlog-v2-corpus.py b/scripts/runlog-v2-corpus.py new file mode 100755 index 0000000..c6248aa --- /dev/null +++ b/scripts/runlog-v2-corpus.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +"""runlog-v2-corpus.py — the behavior_v2 corpus gate (runlog-behavior-v2-plan.md Part 1 §4). + +For EVERY run log under a course-repo clone (default: the local cshl-2026-course +checkout), convert v1 → v2 → v1 with the bridge's strict converter and assert +canonical-JSON identity line by line; record v1 / v1.gz / v2 / v2.gz sizes; and fail +on any line whose (type, event) → key set is not in the known v1 inventory (the +older July logs, legacy `_a`/`_b` protocols and `full`-level logs are exactly the +variants a handful of hand-picked samples would miss). Prints a Markdown report +(summary + per-rig + the largest files + the full per-file table) for the PR +description. Exit status 1 on any failure. + +Not part of `pixi run test` (needs the clone). Run: + + pixi run python scripts/runlog-v2-corpus.py [ROOT] [--out report.md] [--workers 4] + +ROOT may be the clone itself or any directory tree containing *.jsonl / *.jsonl.gz. +""" +from __future__ import annotations + +import argparse +import gzip +import os +import sys +import time +from concurrent.futures import ProcessPoolExecutor + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "fictrac-bridge")) +import bridge # noqa: E402 + +DEFAULT_ROOT = os.path.expanduser("~/Documents/GitHub/cshl-2026-course") +GZIP_LEVEL = 6 # ≈ what the browser's CompressionStream('gzip') produces (Part 2) + +# Known v1 line shapes: (type, event) → every key ever observed for that shape +# (2026-09-06 inventory of 164 logs). A line with a key OUTSIDE this set is a new +# producer variant → the run FAILS so the converter/readers get looked at before +# anything is silently passed through. (Pass-through would still be lossless; the +# point is to notice.) +KNOWN_SHAPES = { + ("session", "logging_started"): {"type", "event", "file", "ms"}, + ("session", "logging_stopped"): {"type", "event", "ms"}, + ("session", "bridge_start"): {"type", "event", "file", "ms"}, + ("log_control", None): {"type", "enabled", "level", "dir", "rx_ms"}, + ("frame_schema", None): {"type", "level", "cols"}, + ("config", None): {"type", "fictrac_port", "gain", "offset", "frames", "dir", "rx_ms"}, + ("fictrac_frame", None): {"type", "seq", "index", "t", "fictrac"}, + ("hello", None): {"type", "client", "v", "dir", "rx_ms"}, + ("log", "arena_command"): set(bridge.ARENA_COMMAND_KEYS), + ("log", "run_metadata"): { + "type", "event", "rig_id", "run_id", "experimenter", "genotype", "age", "sex", "fly_number", + "notes", "protocol_filename", "protocol_sha256", "arena_config", "rig", "firmware", + "controller_id", "timestamp_start", "tool_version", "dir", "rx_ms", + }, + ("log", "runner"): { + "type", "event", "phase", "index", "total", "condition", "op", "value", "ledPercent", "on", + "durationSec", "params", "ledActivation", "status", "ok", "reason", "error", "dir", "rx_ms", + }, +} + + +def shape_of(o): + if isinstance(o, list): + return ("ARRAY", len(o)) + return (o.get("type"), o.get("event")) + + +def gz_len(data: bytes) -> int: + return len(gzip.compress(data, compresslevel=GZIP_LEVEL, mtime=0)) + + +def process(path: str) -> dict: + t = time.time() + r = {"path": path, "ok": True, "errors": [], "unknown_shapes": []} + try: + raw = open(path, "rb").read() + if path.endswith(".gz"): + raw = gzip.decompress(raw) + objs = bridge.read_jsonl(path) + r["fmt"] = bridge.detect_format(objs) + r["lines"] = len(objs) + # ── known-shape gate ──────────────────────────────────────────────── + seen = set() + n_arena = n_frames = 0 + for o in objs: + sh = shape_of(o) + if sh == ("ARRAY", 7) and not isinstance(o[0], str): + n_frames += 1 + continue + if sh == ("log", "arena_command"): + n_arena += 1 + if isinstance(o, list): + r["unknown_shapes"].append(("ARRAY", len(o), str(o)[:60])) + continue + allowed = KNOWN_SHAPES.get(sh) + extra = set(o) - allowed if allowed is not None else set(o) + if allowed is None or extra: + key = (sh, tuple(sorted(extra))) + if key not in seen: + seen.add(key) + r["unknown_shapes"].append((sh, sorted(extra))) + r["n_arena"], r["n_frames"], r["n_other"] = n_arena, n_frames, len(objs) - n_arena - n_frames + # ── round trip ────────────────────────────────────────────────────── + if r["fmt"] == "behavior_v2": + v2, back = objs, bridge.convert_v2_to_v1(objs) + v2 = bridge.convert_v1_to_v2(back, strict=True) + cmp_a, cmp_b = objs, v2 + else: + v2 = bridge.convert_v1_to_v2(objs, strict=True) + back = bridge.convert_v2_to_v1(v2) + cmp_a, cmp_b = objs, back + if len(cmp_a) != len(cmp_b): + r["errors"].append(f"line count {len(cmp_a)} → {len(cmp_b)}") + else: + for i, (a, b) in enumerate(zip(cmp_a, cmp_b)): + if bridge.canonical_json(a) != bridge.canonical_json(b): + r["errors"].append(f"line {i + 1} differs: {bridge.canonical_json(a)[:80]} vs {bridge.canonical_json(b)[:80]}") + if len(r["errors"]) > 5: + break + v2_bytes = bridge.dumps_jsonl(v2).encode("utf-8") + r["v1"] = len(raw) + r["v1_compact"] = len(bridge.dumps_jsonl(objs).encode("utf-8")) + r["v1_gz"] = gz_len(raw) + r["v2"] = len(v2_bytes) + r["v2_gz"] = gz_len(v2_bytes) + r["n_a"] = sum(1 for o in v2 if bridge.is_arena_array(o)) + if r["n_a"] != n_arena: + r["errors"].append(f"{n_arena - r['n_a']} arena_command lines were not compacted (verbatim)") + except Exception as exc: # noqa: BLE001 — report, don't die mid-corpus + r["errors"].append(f"{type(exc).__name__}: {exc}") + if r["unknown_shapes"] or r["errors"]: + r["ok"] = False + r["secs"] = time.time() - t + return r + + +def mb(n: float) -> str: + return f"{n / 1e6:.1f}" + + +def main(argv=None) -> int: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("root", nargs="?", default=DEFAULT_ROOT) + p.add_argument("--out", default=None, help="write the Markdown report here (default: stdout)") + p.add_argument("--workers", type=int, default=4) + p.add_argument("--top", type=int, default=12, help="how many largest files to list in the summary") + args = p.parse_args(argv) + + files = sorted( + os.path.join(d, f) + for d, _, fs in os.walk(args.root) + for f in fs + if (f.endswith(".jsonl") or f.endswith(".jsonl.gz")) and "/.git/" not in d + ) + if not files: + print(f"no run logs under {args.root}", file=sys.stderr) + return 2 + print(f"[corpus] {len(files)} files under {args.root}", file=sys.stderr) + t = time.time() + with ProcessPoolExecutor(max_workers=args.workers) as ex: + results = list(ex.map(process, files, chunksize=1)) + for r in results: + print(f"[corpus] {'ok ' if r['ok'] else 'FAIL'} {os.path.relpath(r['path'], args.root)} ({r['secs']:.1f}s)", file=sys.stderr) + elapsed = time.time() - t + + ok = [r for r in results if r["ok"]] + bad = [r for r in results if not r["ok"]] + sized = [r for r in results if "v2_gz" in r] + tot = {k: sum(r[k] for r in sized) for k in ("v1", "v1_compact", "v1_gz", "v2", "v2_gz", "lines", "n_arena", "n_frames", "n_other")} + + out = [] + w = out.append + w(f"## behavior_v2 corpus gate — {len(files)} run logs under `{args.root}`") + w("") + w(f"Round trip v1 → v2 → v1 (strict converter, canonical-JSON identity per line): **{len(ok)} / {len(results)} files pass**" + + (f", **{len(bad)} FAIL**" if bad else "") + f". Formats: " + + ", ".join(f"{n} {f}" for f, n in sorted(((f, sum(1 for r in results if r.get('fmt') == f)) for f in {r.get('fmt') for r in results}), key=lambda x: -x[1])) + + f". Wall time {elapsed:.0f} s ({args.workers} workers).") + w("") + w("| | lines | arena echoes | frames | other | v1 on disk | v1.gz | v2 | v2.gz | v2/v1 | v2.gz/v1 |") + w("|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|") + w(f"| **total** | {tot['lines']:,} | {tot['n_arena']:,} | {tot['n_frames']:,} | {tot['n_other']:,} | {mb(tot['v1'])} MB | {mb(tot['v1_gz'])} MB | {mb(tot['v2'])} MB | {mb(tot['v2_gz'])} MB | {tot['v2'] / tot['v1']:.2f} | {tot['v2_gz'] / tot['v1']:.3f} |") + w("") + w(f"gzip level {GZIP_LEVEL} (≈ the browser's CompressionStream). `v1 on disk` counts the July logs' non-compact whitespace; compact v1 would be {mb(tot['v1_compact'])} MB.") + w("") + # per-rig + rigs = {} + for r in sized: + rig = os.path.relpath(r["path"], args.root).split(os.sep) + rig = rig[1] if len(rig) > 2 else rig[0] + rigs.setdefault(rig, []).append(r) + w("### Per rig") + w("") + w("| rig | files | pass | v1 on disk | v2 | v2.gz | v2/v1 | v2.gz/v1 |") + w("|---|---:|---:|---:|---:|---:|---:|---:|") + for rig, rs in sorted(rigs.items()): + s = {k: sum(r[k] for r in rs) for k in ("v1", "v2", "v2_gz")} + w(f"| {rig} | {len(rs)} | {sum(1 for r in rs if r['ok'])} | {mb(s['v1'])} MB | {mb(s['v2'])} MB | {mb(s['v2_gz'])} MB | {s['v2'] / s['v1']:.2f} | {s['v2_gz'] / s['v1']:.3f} |") + w("") + w(f"### Largest {args.top} files") + w("") + w("| file | lines | arena | v1 on disk | v1.gz | v2 | v2.gz | v2/v1 |") + w("|---|---:|---:|---:|---:|---:|---:|---:|") + for r in sorted(sized, key=lambda r: -r["v1"])[: args.top]: + w(f"| {os.path.relpath(r['path'], args.root)} | {r['lines']:,} | {r['n_arena']:,} | {mb(r['v1'])} | {mb(r['v1_gz'])} | {mb(r['v2'])} | {mb(r['v2_gz'])} | {r['v2'] / r['v1']:.2f} |") + w("") + if bad: + w("### FAILURES") + w("") + for r in bad: + w(f"- `{os.path.relpath(r['path'], args.root)}`") + for e in r["errors"]: + w(f" - {e}") + for sh in r["unknown_shapes"]: + w(f" - unknown line shape: {sh}") + w("") + w("
All files (MB)") + w("") + w("| file | fmt | ok | lines | arena | frames | v1 | v1.gz | v2 | v2.gz | v2/v1 |") + w("|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|") + for r in results: + rel = os.path.relpath(r["path"], args.root) + if "v2_gz" not in r: + w(f"| {rel} | {r.get('fmt', '?')} | ✗ | | | | | | | | |") + continue + w(f"| {rel} | {r['fmt']} | {'✓' if r['ok'] else '✗'} | {r['lines']:,} | {r['n_arena']:,} | {r['n_frames']:,} | {mb(r['v1'])} | {mb(r['v1_gz'])} | {mb(r['v2'])} | {mb(r['v2_gz'])} | {r['v2'] / r['v1']:.2f} |") + w("") + w("
") + report = "\n".join(out) + "\n" + if args.out: + with open(args.out, "w", encoding="utf-8") as fh: + fh.write(report) + print(f"[corpus] report → {args.out}", file=sys.stderr) + else: + print(report) + print(f"[corpus] {len(ok)} / {len(results)} pass", file=sys.stderr) + return 1 if bad else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test-bridge-behavior.py b/tests/test-bridge-behavior.py index 94c6ab0..27b58df 100644 --- a/tests/test-bridge-behavior.py +++ b/tests/test-bridge-behavior.py @@ -1,13 +1,21 @@ #!/usr/bin/env python3 -"""Offline tests for fictrac-bridge/bridge.py behavior_v1_row() — the pure ns→ms -timestamp normalization + column mapping that the live scope AND the offline -analysis dashboard depend on. No sockets/WebSocket: the buggy logic is pure, so -this runs offline. Wired into `pixi run test`. +"""Offline tests for fictrac-bridge/bridge.py — no sockets/WebSocket, wired into +`pixi run test`: + 1. behavior_v1_row(): the pure ns→ms timestamp normalization + column mapping the + live scope AND the offline analysis dashboard depend on. + 2. behavior_v2 (docs/development/runlog-behavior-v2-plan.md Part 1): compact + arena-echo encode/decode invariants, v1→v2→v1 round trip on real line samples + (ok / status-1 reject / timeout / non-null error / unknown key → raise), the + legacy no-schema path, LogWriter output per level, and the log_control / + hello acknowledgements of the dispatcher. Run: python tests/test-bridge-behavior.py """ +import asyncio +import json import os import sys +import tempfile # import bridge.py (lives in fictrac-bridge/, not on the default path) sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "fictrac-bridge")) @@ -91,6 +99,240 @@ def rec(fc, x, y, hd, ts): check("missing col-22 → ft None", bridge.behavior_v1_row(short, index=1, rel_ms=3, ft0=None)["ft"], None) check("missing col-22 still maps hd", bridge.behavior_v1_row(short, index=1, rel_ms=3, ft0=None)["fc"], 7) + +# ═════════════════════════════════════════════════════════════════════════════ +# behavior_v2 — compact arena echo + round trip + ack (plan Part 1) +# ═════════════════════════════════════════════════════════════════════════════ +def canon(o): + return bridge.canonical_json(o) + + +def check_raises(name, fn, exc=bridge.RunlogFormatError): + global total, failures + total += 1 + try: + fn() + except exc: + print(f" PASS {name} (raised {exc.__name__})") + return + except Exception as e: # noqa: BLE001 + print(f" FAIL {name} — raised {type(e).__name__}: {e}") + else: + print(f" FAIL {name} — did not raise") + failures += 1 + + +T0 = 1788636439304 # a real logging_started.ms (rig03-sr run rydc2tql) +# Real v1 lines (verbatim shapes from the course corpus; values trimmed). +A_OK = json.loads('{"type":"log","event":"arena_command","t":1788636442353,"dt":7,"len":4,"head":"03 70 2e 00","status":0,"echo":112,"ok":true,"error":null,"dir":"browser→bridge","rx_ms":1788636442360}') +A_REJECT = json.loads('{"type":"log","event":"arena_command","t":1788636442400,"dt":4,"len":4,"head":"03 70 31 00","status":1,"echo":112,"ok":false,"error":null,"dir":"browser→bridge","rx_ms":1788636442405}') +A_TIMEOUT = json.loads('{"type":"log","event":"arena_command","t":1788636442467,"dt":505,"len":4,"head":"03 70 a6 00","status":null,"echo":null,"ok":null,"error":"response timeout after 500 ms (cmd 0x70)","dir":"browser→bridge","rx_ms":1788636442972}') +# Synthetic: a decoded reply AND an error string (not seen in the corpus; must survive). +A_ERR_WITH_STATUS = dict(A_REJECT, error="post-decode warning", t=1788636442500, rx_ms=1788636442504) +A_OTHER_CMD = json.loads('{"type":"log","event":"arena_command","t":1788636442600,"dt":9,"len":4,"head":"03 a0 01 00","status":0,"echo":160,"ok":true,"error":null,"dir":"browser→bridge","rx_ms":1788636442610}') + +print("=== behavior_v2: compact_arena_command / expand_arena_command ===") +check("ok line → 6-element array", bridge.compact_arena_command(A_OK, T0), ["a", 3049, 7, "03702e00", 0, 3056]) +check("status-1 reject → status kept", bridge.compact_arena_command(A_REJECT, T0)[4], 1) +arr_to = bridge.compact_arena_command(A_TIMEOUT, T0) +check("timeout → status null + error as 7th element", arr_to, ["a", 3163, 505, "0370a600", None, 3668, "response timeout after 500 ms (cmd 0x70)"]) +check("error with status → 7 elements", len(bridge.compact_arena_command(A_ERR_WITH_STATUS, T0)), 7) +for name, o in [("ok", A_OK), ("reject", A_REJECT), ("timeout", A_TIMEOUT), ("error+status", A_ERR_WITH_STATUS), ("other cmd 0xa0", A_OTHER_CMD)]: + back = bridge.expand_arena_command(bridge.compact_arena_command(o, T0), T0) + check(f"expand(compact({name})) == original (canonical)", canon(back), canon(o)) + check(f"expand({name}) preserves v1 key order", list(back), list(bridge.ARENA_COMMAND_KEYS)) +check("expanded timeout has echo/ok null (not derived)", (bridge.expand_arena_command(arr_to, T0)["echo"], bridge.expand_arena_command(arr_to, T0)["ok"]), (None, None)) +check("is_arena_array: compact line", bridge.is_arena_array(["a", 1, 2, "03700000", 0, 3]), True) +check("is_arena_array: behavior frame array is NOT", bridge.is_arena_array([5, 9052, 39, 0.0, 1.42578, -3.19222, 1.21378]), False) + +print("=== behavior_v2: lines that must NOT be compacted (verbatim fallback / strict raise) ===") +def not_compactable(name, o): + check(f"lenient: {name} → None", bridge.compact_arena_command(o, T0), None) + check_raises(f"strict: {name} raises", lambda: bridge.compact_arena_command(o, T0, strict=True)) +not_compactable("unknown extra key", dict(A_OK, extra=1)) +not_compactable("missing key", {k: v for k, v in A_OK.items() if k != "len"}) +not_compactable("echo != command byte", dict(A_OK, echo=113)) +not_compactable("ok inconsistent with status", dict(A_OK, ok=False)) +not_compactable("status null but ok set", dict(A_TIMEOUT, ok=False)) +not_compactable("len != head bytes", dict(A_OK, len=5)) +not_compactable("truncated head (' …')", dict(A_OK, head="03 8d 00 01 02 03 04 05 …", len=12)) +not_compactable("uppercase hex head", dict(A_OK, head="03 70 2E 00")) +not_compactable("float t", dict(A_OK, t=1788636442353.5)) +not_compactable("other dir", dict(A_OK, dir="bridge→browser")) +not_compactable("non-string error", dict(A_OK, error={"code": 1})) +not_compactable("not an arena_command", {"type": "log", "event": "runner", "phase": "x"}) +check_raises("expand: malformed array raises", lambda: bridge.expand_arena_command(["a", 1, 2, "037", 0, 3], T0)) +check_raises("expand: non-int offset raises", lambda: bridge.expand_arena_command(["a", 1.5, 2, "03700000", 0, 3], T0)) + +print("=== behavior_v2: v1 → v2 → v1 file round trip ===") +V1_FILE = [ + {"type": "session", "event": "logging_started", "file": "arena-log-x.jsonl", "ms": T0}, + {"type": "frame_schema", "level": "behavior_v1", "cols": bridge.BEHAVIOR_V1_COLS}, + json.loads('{"type":"log_control","enabled":true,"level":"behavior_v1","dir":"browser→bridge","rx_ms":1788636439305}'), + json.loads('{"type":"log","event":"run_metadata","rig_id":"rig03-sr","run_id":"rydc2tql","experimenter":"x","genotype":"g","notes":"","protocol_filename":"p.yaml","protocol_sha256":"00","arena_config":"G6_2x10","rig":"cshl_g6_2x10_ball","firmware":"v1","dir":"browser→bridge","rx_ms":1788636439306}'), + json.loads('{"type":"log","event":"runner","phase":"trial-running","index":0,"durationSec":40,"params":{"mode":3,"patternId":36,"frameRate":0,"gain":0,"initPos":0,"duration":0,"duty":0},"condition":"c","status":0,"ok":true,"dir":"browser→bridge","rx_ms":1788636442300}'), + [5, 9052, 39, 0.0, 1.42578, -3.19222, 1.21378], + A_OK, + [13, 9053, 39, 8.272, 1.42579, -3.19221, 1.21379], + A_REJECT, + A_TIMEOUT, + A_ERR_WITH_STATUS, + json.loads('{"type":"config","fictrac_port":60000,"gain":1.8,"offset":0,"frames":200,"dir":"browser→bridge","rx_ms":1788636442700}'), + {"type": "session", "event": "logging_stopped", "ms": T0 + 40000}, +] +v2 = bridge.convert_v1_to_v2(V1_FILE) +check("same line count", len(v2), len(V1_FILE)) +check("v2 schema replaces the v1 schema in place", v2[1], {"type": "frame_schema", "level": "behavior_v2", "cols": bridge.BEHAVIOR_V1_COLS, "arena_cols": ["t_off", "dt", "hex", "status", "rx_off"], "t0": T0}) +check("t0 = logging_started.ms", v2[1]["t0"], T0) +check("arena lines became 'a' arrays", sum(1 for o in v2 if bridge.is_arena_array(o)), 4) +check("frame arrays unchanged", v2[5], V1_FILE[5]) +check("runner line verbatim", v2[4], V1_FILE[4]) +check("detect_format(v1) = behavior_v1", bridge.detect_format(V1_FILE), "behavior_v1") +check("detect_format(v2) = behavior_v2", bridge.detect_format(v2), "behavior_v2") +back = bridge.convert_v2_to_v1(v2) +check("round trip: line count", len(back), len(V1_FILE)) +check("round trip: every line canonical-identical", [canon(a) == canon(b) for a, b in zip(V1_FILE, back)], [True] * len(V1_FILE)) +check("round trip: v1 schema restored exactly", back[1], V1_FILE[1]) +check("v2 is smaller", len(bridge.dumps_jsonl(v2)) < len(bridge.dumps_jsonl(V1_FILE)), True) +check_raises("v1→v2 of a v2 file raises", lambda: bridge.convert_v1_to_v2(v2)) +check_raises("v2→v1 of a v1 file raises", lambda: bridge.convert_v2_to_v1(V1_FILE)) +check_raises("strict v1→v2 with an unknown arena key raises", lambda: bridge.convert_v1_to_v2(V1_FILE[:6] + [dict(A_OK, extra=1)])) +check("lenient v1→v2 keeps the unknown-key line verbatim", bridge.convert_v1_to_v2(V1_FILE[:6] + [dict(A_OK, extra=1)], strict=False)[-1], dict(A_OK, extra=1)) +check_raises("v2 schema with unknown key raises", lambda: bridge.convert_v2_to_v1([dict(v2[1], foo=1)] + v2[2:])) +check_raises("v2 schema with other arena_cols raises", lambda: bridge.convert_v2_to_v1([dict(v2[1], arena_cols=["x"])] + v2[2:])) +check_raises("v1 schema with unknown key raises", lambda: bridge.convert_v1_to_v2([V1_FILE[0], dict(V1_FILE[1], extra=1)])) + +print("=== behavior_v2: legacy file without a frame_schema (pre-#140 / full level) ===") +LEGACY = [ + {"type": "session", "event": "logging_started", "file": "arena-log-y.jsonl", "ms": T0}, + json.loads('{"type":"log_control","enabled":true,"dir":"browser→bridge","rx_ms":1788636439305}'), + {"type": "fictrac_frame", "seq": 1999753, "index": 54, "t": T0 + 19, "fictrac": [1999753.0] + [0.0] * 24}, + A_OK, + {"type": "session", "event": "logging_stopped", "ms": T0 + 500}, +] +lv2 = bridge.convert_v1_to_v2(LEGACY) +check("detect_format(legacy full) = full", bridge.detect_format(LEGACY), "full") +check("detect_format(pre-#140) = legacy", bridge.detect_format([LEGACY[0], {"type": "fictrac_frame", "seq": 1, "index": 2, "t": 3}]), "legacy") +check("schema inserted after the session line", (lv2[1]["type"], lv2[1]["level"]), ("frame_schema", "behavior_v2")) +check("inserted schema has cols null (no positional frames)", lv2[1]["cols"], None) +check("one extra line in v2", len(lv2), len(LEGACY) + 1) +check("arena line compacted in legacy file", bridge.is_arena_array(lv2[4]), True) +lback = bridge.convert_v2_to_v1(lv2) +check("legacy round trip: line count restored", len(lback), len(LEGACY)) +check("legacy round trip: identical", [canon(a) == canon(b) for a, b in zip(LEGACY, lback)], [True] * len(LEGACY)) + +print("=== behavior_v2: --convert file round trip (.jsonl and .jsonl.gz) ===") +with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "in.jsonl") + with open(src, "w", encoding="utf-8") as fh: + for o in V1_FILE: + fh.write(json.dumps(o) + "\n") # non-compact separators, like the July logs + gz = os.path.join(d, "out.jsonl.gz") + back_path = os.path.join(d, "back.jsonl") + check("main(--convert v1→v2.gz) exits 0", bridge.main(["--convert", src, gz]), 0) + check("gz output detected as behavior_v2", bridge.detect_format(bridge.read_jsonl(gz)), "behavior_v2") + check("main(--convert v2.gz→v1) exits 0", bridge.main(["--convert", gz, back_path]), 0) + check("file round trip identical", [canon(o) for o in bridge.read_jsonl(back_path)], [canon(o) for o in V1_FILE]) + check("--to v2 on a v2 file fails cleanly (exit 1)", bridge.main(["--convert", gz, os.path.join(d, "x.jsonl"), "--to", "v2"]), 1) + check("gzip output is reproducible (mtime 0)", bridge.write_jsonl(gz, v2) == bridge.write_jsonl(os.path.join(d, "again.jsonl.gz"), v2) and open(gz, "rb").read() == open(os.path.join(d, "again.jsonl.gz"), "rb").read(), True) + +print("=== LogWriter: levels + what lands in the file ===") +def write_session(level, inbound, frames=1): + with tempfile.TemporaryDirectory() as d: + lw = bridge.LogWriter(None, level, d) + check(f"[{level}] fresh writer level", lw.level, level) + lw.start_new_log() + for raw in inbound: + lw.write_inbound(raw) + beh = {"ms": 5, "fc": 9052, "idx": 39, "ft": 0.0, "x": 1.42578, "y": -3.19222, "hd": 1.21378} + for _ in range(frames): + lw.write_frame(beh, [9052.0] + [0.0] * 24) + name = lw.current_name + lw.close() + return bridge.read_jsonl(os.path.join(d, name)) + +check("default level is behavior_v2", bridge.LogWriter(None).level, "behavior_v2") +check("legacy log_frames=True → full", bridge.LogWriter(None, True).level, "full") +check("legacy log_frames=False → default", bridge.LogWriter(None, False).level, "behavior_v2") +check("log_frames alias", bridge.LogWriter(None, "full").log_frames, True) +check_raises("unknown level at construction raises", lambda: bridge.LogWriter(None, "bogus"), ValueError) +lw = bridge.LogWriter(None) +check("set_level accepts behavior_v1", lw.set_level("behavior_v1"), True) +check("set_level rejects unknown and keeps level", (lw.set_level("behavior_v9"), lw.level), (False, "behavior_v1")) +check("set_level accepts full / behavior_v2", (lw.set_level("full"), lw.set_level("behavior_v2"), lw.level), (True, True, "behavior_v2")) + +a_raw = json.dumps({k: v for k, v in A_OK.items() if k not in ("dir", "rx_ms")}) # as the browser sends it +lines = write_session("behavior_v2", [a_raw, '{"type":"log","event":"runner","phase":"sequence-start","total":1}', "not json"]) +sch = lines[1] +check("[v2] schema line level/arena_cols", (sch["level"], sch["arena_cols"]), ("behavior_v2", bridge.BEHAVIOR_V2_ARENA_COLS)) +check("[v2] schema t0 == session line ms", sch["t0"], lines[0]["ms"]) +check("[v2] schema cols = behavior cols", sch["cols"], bridge.BEHAVIOR_V1_COLS) +check("[v2] arena_command written as 'a' array", bridge.is_arena_array(lines[2]), True) +exp = bridge.expand_arena_command(lines[2], sch["t0"]) +check("[v2] expanded echo matches the browser payload (+ dir, rx_ms stamps)", {k: exp[k] for k in A_OK if k not in ("rx_ms",)}, {k: A_OK[k] for k in A_OK if k not in ("rx_ms",)}) +check("[v2] rx_ms stamped as int", isinstance(exp["rx_ms"], int), True) +check("[v2] runner line verbatim object", lines[3]["phase"], "sequence-start") +check("[v2] unparsed inbound kept", lines[4]["event"], "unparsed") +check("[v2] frame row unchanged 7-array", lines[5], [5, 9052, 39, 0.0, 1.42578, -3.19222, 1.21378]) +check("[v2] file converts back to v1 cleanly", bridge.detect_format(bridge.convert_v2_to_v1(lines)), "behavior_v1") +bad = dict(A_OK); bad["head"] = "03 8d 00 01 02 03 04 05 …"; bad["len"] = 12 +lines = write_session("behavior_v2", [json.dumps({k: v for k, v in bad.items() if k not in ("dir", "rx_ms")})]) +check("[v2] non-fitting echo kept verbatim (lossless fallback)", (isinstance(lines[2], dict), lines[2]["head"]), (True, bad["head"])) + +lines = write_session("behavior_v1", [a_raw]) +check("[v1] schema line unchanged shape", lines[1], {"type": "frame_schema", "level": "behavior_v1", "cols": bridge.BEHAVIOR_V1_COLS}) +check("[v1] arena_command stays an object", isinstance(lines[2], dict) and lines[2]["event"] == "arena_command", True) +check("[v1] frame row 7-array", len(lines[3]), 7) +lines = write_session("full", [a_raw]) +check("[full] no schema line", any(o.get("type") == "frame_schema" for o in lines if isinstance(o, dict)), False) +check("[full] fictrac_frame object with 25 cols", (lines[2]["type"], len(lines[2]["fictrac"])), ("fictrac_frame", 25)) + +print("=== dispatcher: hello_ack + log_control_ack ===") +class FakeWs: + def __init__(self): + self.sent = [] + async def send(self, s): + self.sent.append(json.loads(s)) + +class FakeInputs: + port = 60000 + async def rebind(self, port): + self.port = port + +async def drive(): + with tempfile.TemporaryDirectory() as d: + log = bridge.LogWriter(None, bridge.DEFAULT_LOG_LEVEL, d) + pipeline = bridge.Pipeline(None, log, 200, 1.8, 0.0) + dispatch = bridge.make_dispatcher(pipeline, log, FakeInputs()) + ws = FakeWs() + await dispatch(json.dumps({"type": "hello", "client": "webDisplayTools", "v": 1}), ws) + hello = ws.sent[-1] + await dispatch(json.dumps({"type": "log_control", "enabled": True, "level": "behavior_v2"}), ws) + ack_v2 = ws.sent[-1] + f_v2 = log.current_name + await dispatch(json.dumps({"type": "log_control", "enabled": False}), ws) + ack_off = ws.sent[-1] + await dispatch(json.dumps({"type": "log_control", "enabled": True, "level": "behavior_v9"}), ws) + ack_bogus = ws.sent[-1] + await dispatch(json.dumps({"type": "log_control", "enabled": False}), ws) + await dispatch(json.dumps({"type": "log_control", "enabled": True, "level": "behavior_v1"}), ws) + ack_v1 = ws.sent[-1] + await dispatch(json.dumps({"type": "log_control", "enabled": False}), ws) + await dispatch(json.dumps({"type": "log_control", "enabled": True}), None) # no websocket: must not crash + await dispatch(json.dumps({"type": "log_control", "enabled": False}), None) + first = bridge.read_jsonl(os.path.join(d, f_v2)) + return hello, ack_v2, ack_off, ack_bogus, ack_v1, first + +hello, ack_v2, ack_off, ack_bogus, ack_v1, first = asyncio.run(drive()) +check("hello_ack type", hello["type"], "hello_ack") +check("hello_ack advertises levels (v2 first)", hello["levels"], ["behavior_v2", "behavior_v1", "full"]) +check("hello_ack carries the bridge version + current level", (hello["bridge"], hello["level"], hello["logging"]), (bridge.BRIDGE_VERSION, "behavior_v2", False)) +check("log_control_ack v2: enabled + level + file", (ack_v2["type"], ack_v2["enabled"], ack_v2["level"], ack_v2["requested"], ack_v2["file"].startswith("arena-log-")), ("log_control_ack", True, "behavior_v2", "behavior_v2", True)) +check("log_control_ack off: enabled false, level kept", (ack_off["enabled"], ack_off["level"]), (False, "behavior_v2")) +check("unknown level: ack reports the level ACTUALLY in force", (ack_bogus["level"], ack_bogus["requested"], ack_bogus["enabled"]), ("behavior_v2", "behavior_v9", True)) +check("behavior_v1 still selectable", ack_v1["level"], "behavior_v1") +check("the v2 file: hello + log_control logged, schema is v2", (first[0]["type"], first[1]["level"], first[2]["type"]), ("session", "behavior_v2", "log_control")) + print("\n=== Summary ===") print(f"{total - failures} / {total} checks passed") sys.exit(1 if failures else 0)