istrain
559caead36
Public community release. The complete working system — DSP + decoders (scripts/), web dashboard + API (dashboard/), container stack (docker/), config templates (config/) — plus APOCALYPSE-EDITION.md, the full build guide for a human or a coding agent. GPLv3 (PyEOT dependency). Scrubbed of all secrets, internal IPs, hostnames, and location detail. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
73 lines
3.5 KiB
Python
73 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
# hot_explore.py — Head-of-Train (HOT, 452.9375) reverse-engineering explorer. STANDALONE / offline.
|
|
#
|
|
# STATUS (2026-06-26): structure cracked, fields NOT decoded. There is no public HOT bit-map (see
|
|
# ../../EOT-HOT-PROTOCOL.md) — this is pure RE off our own bot_clips/*.s16. What works so far:
|
|
# - reuse the proven EOT FFSK front-end (eot_scan: 1200-baud, mark 1200 / space 1800 Hz)
|
|
# - a strong HOT burst shows a ~440-bit alternating PREAMBLE (spec: 456) then a 64-bit REPEAT PERIOD
|
|
# (spec: a 64-bit block sent 3x). Majority-voting the 3 internal copies -> a CLEAN 64-bit block.
|
|
# - artifact test: different sessions give DIFFERENT clean blocks (real frames, not a stuck carrier).
|
|
# NEXT (open): pin the 24-bit frame-sync word, locate the 17-bit addressed-EOT-ID in the 30-bit payload
|
|
# (validate against a known EOT unit heard concurrently), recover the HOT BCH(33) polynomial.
|
|
#
|
|
# Usage: python3 hot_explore.py ~/istrain/bot_clips/bot_*.s16 (defaults to a strong-clip sample)
|
|
import sys, os, glob, numpy as np
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # eot_scan lives next door
|
|
import eot_scan as E
|
|
|
|
def best_bits(path):
|
|
"""Demod FFSK to raw bits; pick the tone pair giving the longest alternating preamble (= bit-sync)."""
|
|
x, fr = E.load(path); W = max(8, int(round(fr / E.BAUD)))
|
|
best = None
|
|
for flo, fhi in E.TONE_PAIRS:
|
|
bs = E.bitstring(E.discriminator(x, fr, flo, fhi, W), fr)
|
|
run = mx = 0
|
|
for i in range(1, len(bs)):
|
|
if bs[i] != bs[i-1]: run += 1; mx = max(mx, run)
|
|
else: run = 0
|
|
if best is None or mx > best[0]: best = (mx, bs)
|
|
return best
|
|
|
|
def preamble_end(bs):
|
|
run = mx = end = 0
|
|
for i in range(1, len(bs)):
|
|
if bs[i] != bs[i-1]:
|
|
run += 1
|
|
if run > mx: mx = run; end = i + 1
|
|
else: run = 0
|
|
return end
|
|
|
|
def vote(bits):
|
|
"""Find (period, offset) maximizing 3-way agreement among 3 consecutive blocks, then majority-vote."""
|
|
a = np.array([int(c) for c in bits]); best = None
|
|
for p in range(60, 69):
|
|
for off in range(0, min(len(a) - 3*p, 30)) if len(a) >= 3*p else []:
|
|
b1, b2, b3 = a[off:off+p], a[off+p:off+2*p], a[off+2*p:off+3*p]
|
|
agree = np.sum(b1 == b2) + np.sum(b1 == b3) + np.sum(b2 == b3)
|
|
if best is None or agree > best[0]: best = (agree, p, off, (b1, b2, b3))
|
|
if best is None: return None
|
|
agree, p, off, (b1, b2, b3) = best
|
|
cons = ((b1.astype(int) + b2 + b3) >= 2).astype(int)
|
|
errs = [int(np.sum(b != cons)) for b in (b1, b2, b3)]
|
|
return p, off, ''.join(map(str, cons.tolist())), errs, agree, 3 * p
|
|
|
|
def explore(path):
|
|
mx, bs = best_bits(path)
|
|
body = bs[preamble_end(bs):]
|
|
if len(body) < 3 * 60:
|
|
print(f"{os.path.basename(path):30s} preamble={mx:3d}b (body too short: {len(body)}b — likely not a frame)")
|
|
return None
|
|
r = vote(body)
|
|
if not r: return None
|
|
p, off, cons, errs, agree, maxa = r
|
|
print(f"{os.path.basename(path):30s} preamble={mx:3d}b period={p} off={off} "
|
|
f"3way={agree}/{maxa}={agree/maxa:.0%} vote-err={errs} block={cons}")
|
|
return cons
|
|
|
|
if __name__ == '__main__':
|
|
paths = sys.argv[1:] or sorted(glob.glob(os.path.expanduser("~/istrain/bot_clips/bot_20260625_225*.s16")))[:4]
|
|
blocks = [b for b in (explore(p) for p in paths) if b]
|
|
if len(blocks) > 1:
|
|
print("\nartifact test — distinct consensus blocks (same everywhere = artifact; varying = real):",
|
|
len(set(blocks)))
|