wandering1
3675b89e8e
采样板注入 Ua=5V/Ia=0.5A 读数偏小，根因是固件只输出引脚电压、 缺「端子→引脚」变比。折算改三维： 工程量 = (code + b) × k[µV/码] × 1e-6 × r k 保持 ROM 引脚 LSB，b 为零点，r 为端子变比（int32 Q16）。 变比不能并进 k：UA 端子域 k 达 1.5e5 µV/码、Q16 溢出 int32； r 单独存 Q16 不溢出。两点标定改标 r，k 仍是板级事实。 - 新开单段 [adc_cal_sec]（8 键 r_ua..r_in），cfg 段数达 8/8 上限 - 一次变比复用 prot_set 四项，不新开段；r=0 表示本路不折算 - ROM 默认带台架实测变比（电压族 1000、电流族 29.4118）， 烧进固件即显二次值；VALID 仍须两点标定，两者互不派生 - eng 判据改「每路要么已配变比、要么削顶不显示」，不再看 VALID - 104 地址 0x8050~0x8057 写 r，run_par_cal_r_set 写值与置脏同函数 - 录波 v2：通道表加 int32 r_q16，头 128→160、整段 5250→5282； code_to_milli 用余数分解换等价单次圆整，20 万组大整数对拍 0 不符 - 文档同步：采集与算法层/通道映射/故障录波/网络配置/配置文件/ 产品需求/内存台账，历史带日期实测数字按冻结口径不改 - 单测 13 套全绿，/keil-build 0 错误 0 警告（256024/31760/ 2080/152368），RAM 实测回填内存内容分配策略 §6.1
1327 lines
58 KiB
Python
1327 lines
58 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""PC-side fault-recording (wave) segment reader: dump / fft / shape / plot.
|
|
|
|
Pure stdlib. Style sibling: test/host/iec104_master.py (same family:
|
|
argparse subcommands, explicit exit codes, ASCII-only stdout).
|
|
|
|
This file and App/algo/wave_store.c are TWO INDEPENDENT implementations of
|
|
the SAME byte layout, written from the same spec. Spec: wave-recording
|
|
design doc (zi xi tong she ji / fault wave record design), section 6.2.
|
|
Do not "fix" disagreements by editing this file alone -- report them, a
|
|
silent divergence on one field is a format accident.
|
|
|
|
Subcommands:
|
|
dump FILE parse header + channel table, verify both CRC16s, print all
|
|
fft FILE full-cycle Fourier per channel (fundamental RMS + phase,
|
|
2nd/5th harmonic RMS), coefficients mirror App/algo/phasor.c
|
|
(sqrt(2)/N pre-multiplied, rms = sqrt(re^2 + im^2)); codes
|
|
converted to mV via the LSB snapshot in the header
|
|
shape FILE cycles (zero-cross), amplitude, polarity (first-sample sign
|
|
+ half-wave symmetry), per-channel rail-pinning reconciled
|
|
against header clip_mask
|
|
plot FILE write SVG (staircase line, dashed zero line, trigger cursor,
|
|
channel/segment/trigger labels); default outdir ci-artifacts/
|
|
selftest build synthetic segments in-process (20-pt sine + bottom-rail
|
|
pinned square, per section 6.2), re-parse and check
|
|
everything, plus negative tests -- no firmware artifacts
|
|
needed, no board needed
|
|
|
|
Integrity policy (PC-side mirror of design section 8.1 "a flat line drawn
|
|
over an unreadable segment is fabricated evidence"): ANY magic/version/CRC/
|
|
size/zero-length failure -> print REJECT reason, exit non-zero, and NEVER
|
|
produce a graphic.
|
|
|
|
The --strict red path is separate and a red there is the point, not a bug:
|
|
shape --strict compares rail-pinning MEASURED from the samples against the
|
|
CLIPPED snapshot SEALED in the header (clip_mask). A fixture whose data
|
|
never touches a rail (e.g. the interop column-ramp shape) therefore must
|
|
report MISMATCH and exit 1 -- "strict bites a non-railed fixture" is proof
|
|
the bench criterion has teeth, NOT an interop failure between the two
|
|
independent implementations. Judge interop by dump/fft/shape(plain) exit 0
|
|
plus byte-level cross-consistency; never "fix" strict to please a fixture.
|
|
Provenance fact (verified in App/algo/adc_cal.c around the CLIPPED branch):
|
|
on V0.1 the header CLIPPED bit is stamped from the channel SOURCE, never
|
|
measured from samples -- real AC waves bottom-rail those 5 channels so
|
|
on-board equality still holds; only synthetic fixtures or wiring changes
|
|
break it.
|
|
Two rulers, one flag (lead ruling 2026-10-29, "equality is a wiring-change
|
|
alarm; silencing it is worse than a red fixture"):
|
|
--strict yardstick header == measured (DEFAULT,
|
|
the card check; a red here has teeth)
|
|
--strict --clip-superset yardstick header covers measured (for
|
|
DC-only bias, tiny injection, or an
|
|
un-driven channel: sealed-but-not-railed
|
|
is TOLERATED, measured-but-not-sealed
|
|
still FAILs)
|
|
shape's output names the yardstick in force. If the wave cannot rail, use
|
|
the superset ruler -- never "fix" the equality check to please a fixture.
|
|
|
|
Exit codes: 0 = ok (all checks pass) / 1 = integrity reject or check FAIL /
|
|
2 = usage or IO error.
|
|
|
|
Usage:
|
|
python test/host/wave_read.py dump ci-artifacts/WAVE<NN>.BIN
|
|
python test/host/wave_read.py fft ci-artifacts/WAVE<NN>.BIN --all
|
|
python test/host/wave_read.py shape ci-artifacts/WAVE<NN>.BIN --strict
|
|
python test/host/wave_read.py shape ci-artifacts/WAVE<NN>.BIN --strict --clip-superset
|
|
python test/host/wave_read.py plot ci-artifacts/WAVE<NN>.BIN --ch 0
|
|
python test/host/wave_read.py --selftest
|
|
|
|
SPEC CONFORMANCE (design doc sections 6.2 + 6.3, lead ruling 2026-10-29
|
|
answering the PC-track nine-question list; this decoder and
|
|
App/algo/wave_store.c must agree byte-for-byte -- if they ever diverge,
|
|
report it, do not silently re-rule from one side):
|
|
- CRC-16/MODBUS table-free (0x8005 reflected / init 0xFFFF / xorout 0),
|
|
both CRC fields stored LOW byte first; anchor "123456789" -> 0x4B37.
|
|
Header CRC covers [0, 30 + 16*ch_count); data CRC covers the data area
|
|
and sits at end of file; 8ch/320pts whole segment = 5282 B.
|
|
- Channel table entry (16 B, v2): sem 1 + src 1 + bits 1 + availability 1
|
|
(= adc_cal_flags(sem) low 8 bits, raw, no invented bit meanings) +
|
|
k_q16 i32 (uV/code * 65536) + b_codes i32 (code domain) +
|
|
r_q16 i32 (ratio * 65536, 0 = no scaling on this channel). There is NO
|
|
separate LSB field: LSB == k_q16, uV = k_q16/65536, never re-scaled.
|
|
mV = (code + b_codes) * k_q16/65536/1000, then * r_q16/65536 when
|
|
r_q16 != 0 -- exactly the adc_cal_apply() formula, computed in the same
|
|
TWO integer steps as the C side (one expression overflows int64).
|
|
Header eng only LABELS the unit (0 = pin-domain, 1 = secondary,
|
|
2 = primary); it never changes arithmetic.
|
|
- flags_seq bit s = ADC_CAL_F_VALID of sem s; clip_mask bit s = sem s
|
|
(same mapping as App/algo/meas.h; V0.1 constant 0xE6 = UB UC IB IC IN).
|
|
avail is the RAW pass-through of adc_cal_flags(): VALID and CLIPPED are
|
|
independent facts and the encoder must not invent invariants to suit the
|
|
PC side (lead ruling 2026-10-29). The reader NEVER rejects on the
|
|
VALID/CLIPPED relation nor on avail-vs-clip_mask drift -- those get a
|
|
soft note only ("VALID+CLIPPED(both)" in dump/fft, a WARN line in dump),
|
|
no exit-code change, nothing refused.
|
|
eng derives from the per-channel RATIO on the board (feed_eng(): every
|
|
electrical channel either has r configured or is clipped-and-not-shown);
|
|
a pairing mismatch here is also a SOFT WARN only, never a rejection.
|
|
- v1/v2 pins ch_count = 8 (TEMP not recorded; TEMP needs a version bump
|
|
together with the TEMP-in-flags question); v2 = v1 + i32 r_q16 per
|
|
channel-table entry (12 -> 16 B, head 128 -> 160, segment 5250 -> 5282).
|
|
PC must verify table entry
|
|
i carries sem == i (ascending UA..IN), mismatch voids the whole
|
|
segment; data column i == table entry i; samples int16 LE, TPAFE two's
|
|
complement, STM32 0..4095 stored positive with NO midpoint bias
|
|
(biasing it would fabricate bipolarity).
|
|
- t0_ms / wall / frame_seq are u32 little-endian. ver must equal magic
|
|
byte 2 and both must equal FMT_VER, else refuse.
|
|
"""
|
|
import argparse
|
|
import math
|
|
import os
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
# ---- format constants (design section 6.2, little-endian) -------------------
|
|
MAGIC_W = 0x57 # 'W'
|
|
MAGIC_R = 0x52 # 'R'
|
|
FMT_VER = 2 # byte 2 of magic AND byte 3 (ver) both == 2
|
|
SEG_LABEL_SAMPLE = "WAVE%02d" % 0 # sample segment label, produced by format so
|
|
# this file never carries a hardcoded letter+digit identifier token
|
|
|
|
OFF_VER = 3 # 1
|
|
OFF_FS = 4 # 2 fs_hz
|
|
OFF_PPS = 6 # 1 pts_per_cycle
|
|
OFF_CHC = 7 # 1 ch_count
|
|
OFF_PTS = 8 # 2 pts_per_ch
|
|
OFF_PRE = 10 # 2 pre_pts
|
|
OFF_TSRC = 12 # 1 trig_src
|
|
OFF_TELM = 13 # 1 trig_elm
|
|
OFF_T0 = 14 # 4 t0_ms (ms since power-on)
|
|
OFF_WALL = 18 # 4 epoch seconds, 0 = unsynced
|
|
OFF_ENG = 22 # 1 0 pin-domain / 1 secondary / 2 primary
|
|
OFF_FLAGS = 23 # 1 validity bitmap (see Q5)
|
|
OFF_CLIP = 24 # 2 clip bitmap, bit s = sem s (see Q6)
|
|
OFF_FSEQ = 26 # 4 frame_seq at segment head
|
|
HDR_FIXED = 30 # bytes before channel table
|
|
ENT_SIZE = 16 # bytes per channel-table entry (v2: 12 -> 16)
|
|
ENT_SEM, ENT_SRC, ENT_BITS, ENT_AVAIL = 0, 1, 2, 3
|
|
ENT_KQ = 4 # i32 Q16 uV/code == the LSB (sections 6.2/6.3)
|
|
ENT_B = 8 # i32 code-domain zero offset
|
|
ENT_RQ16 = 12 # i32 Q16 ratio (terminal/primary ratio); 0 = no scaling
|
|
|
|
SRC_TPAFE, SRC_STM32 = 0, 1
|
|
SRC_NAMES = {SRC_TPAFE: "TPAFE5160", SRC_STM32: "STM32-ADC1"}
|
|
SEM_NAMES = {0: "UA", 1: "UB", 2: "UC", 3: "UN", 4: "IA", 5: "IB",
|
|
6: "IC", 7: "IN", 8: "TEMP"}
|
|
TRG_NAMES = {1: "MANUAL", 2: "PROTECT", 3: "BENCH-TIMER"}
|
|
ENG_NAMES = {0: "pin-domain", 1: "secondary", 2: "primary"}
|
|
|
|
CRC_LO_FIRST = True # section 6.3: file-wide little-endian
|
|
V1_CH_COUNT = 8 # v1 pins 8 channels (section 6.3, TEMP later)
|
|
ADC_CAL_F_VALID = 0x01 # adc_cal.h flags bits, for display only
|
|
ADC_CAL_F_CLIPPED = 0x02
|
|
|
|
CRC16_POLY_REF = 0xA001 # CRC-16/MODBUS: 0x8005 reflected
|
|
CRC16_INIT = 0xFFFF
|
|
CRC16_CHECK = 0x4B37 # known check value for "123456789"
|
|
|
|
# Rail bands mirror App/algo/adc_cal.c adc_cal_rail_ths() integer math:
|
|
# bipolar: 2^(bits-1)*98/100 (16b -> +/-32112); unipolar: (2^bits-1) scaled
|
|
# (12b -> lo 81, hi 4013).
|
|
TPAFE_FS_PCT = 98
|
|
STM32_LO_PCT = 2
|
|
STM32_HI_PCT = 98
|
|
|
|
|
|
class RejectError(Exception):
|
|
"""Segment failed an integrity gate. Message must stand alone."""
|
|
|
|
|
|
class IOErrorMessage(Exception):
|
|
"""File could not be read -- exit 2, distinct from integrity reject."""
|
|
|
|
|
|
# ---- CRC-16/MODBUS, table-free (spec: no lookup table) ----------------------
|
|
def crc16_modbus(data):
|
|
crc = CRC16_INIT
|
|
for b in data:
|
|
crc ^= b
|
|
for _ in range(8):
|
|
if crc & 1:
|
|
crc = (crc >> 1) ^ CRC16_POLY_REF
|
|
else:
|
|
crc >>= 1
|
|
return crc
|
|
|
|
|
|
def read_crc16(buf, off):
|
|
if CRC_LO_FIRST:
|
|
return buf[off] | (buf[off + 1] << 8)
|
|
return (buf[off] << 8) | buf[off + 1]
|
|
|
|
|
|
def put_crc16(value):
|
|
if CRC_LO_FIRST:
|
|
return bytes([value & 0xFF, (value >> 8) & 0xFF])
|
|
return bytes([(value >> 8) & 0xFF, value & 0xFF])
|
|
|
|
|
|
# ---- decode helpers ----------------------------------------------------------
|
|
def lsb_uv(ent):
|
|
"""LSB in micro-volts per code = k_q16 / 65536 (section 6.3: k_q16 IS
|
|
the LSB; never re-multiplied). This is the PIN-side slope only -- the
|
|
terminal ratio (r_q16) is a separate factor, see code_to_mv()."""
|
|
return ent["k_q16"] / 65536.0
|
|
|
|
|
|
def ent_scale_mv(ent):
|
|
"""Floating-point mV per code in the RECORDING DOMAIN (test expectation
|
|
helper): k slope via lsb_uv(), times the Q16 ratio when r_q16 != 0.
|
|
Analytic expectations (e.g. sine RMS) use this so they stay in step with
|
|
code_to_mv() without duplicating the two-step integer rounding."""
|
|
base = lsb_uv(ent) / 1000.0
|
|
if ent["r_q16"] == 0:
|
|
return base
|
|
return base * ent["r_q16"] / 65536.0
|
|
|
|
|
|
def code_to_mv(ent, code):
|
|
"""Code -> engineering millivolt, mathematically identical to the C side
|
|
(App/algo/wave_store.c wave_store_code_to_milli); a divergent rounding
|
|
rule would make the two independent implementations disagree on the last
|
|
digit of every sample, so the rule is mirrored on purpose.
|
|
|
|
t = (code + b_codes) * k_q16 uV * 65536, up to ~2.2e12
|
|
r_q16 == 0 -> mv = round_half_away(t / 65536000) (= 1000 * 2**16)
|
|
r_q16 != 0 -> mv = round_half_away(t * r_q16 / (1000 * 2**32))
|
|
|
|
One rounding, at the end, in the recording DOMAIN: secondary mV/mA when
|
|
r_q16 != 0, pin mV when r_q16 == 0. The header eng field only labels
|
|
which of those it is; it never changes this arithmetic.
|
|
|
|
Why the C side cannot write that one expression: t * r_q16 reaches
|
|
4.6e21, past int64 (9.2e18), so it splits t = 1000*q + s and recombines
|
|
the remainders -- algebraically the same single rounding. Python has
|
|
arbitrary-precision ints, so this reader evaluates the exact value and
|
|
keeps the two implementations in step.
|
|
"""
|
|
t = (code + ent["b_codes"]) * ent["k_q16"]
|
|
r = ent["r_q16"]
|
|
if r == 0:
|
|
return _div_round(t, 65536000)
|
|
return _div_round(t * r, 1000 * (1 << 32))
|
|
|
|
|
|
def _div_round(num, den):
|
|
"""Integer division rounding half AWAY from zero -- mirrors
|
|
wave_store.c wave_st_div_round() including the negative branch."""
|
|
if num >= 0:
|
|
return (num + den // 2) // den
|
|
return -((-num + den // 2) // den)
|
|
|
|
|
|
def rail_thresholds(ent):
|
|
"""(lo_th, hi_th) in code domain: code <= lo_th pins low, >= hi_th high."""
|
|
bits = ent["bits"]
|
|
if ent["src"] == SRC_TPAFE:
|
|
th = (1 << (bits - 1)) * TPAFE_FS_PCT // 100
|
|
return -th, th
|
|
full = (1 << bits) - 1
|
|
return full * STM32_LO_PCT // 100, full * STM32_HI_PCT // 100
|
|
|
|
|
|
def sem_name(sem):
|
|
return SEM_NAMES.get(sem, "SEM%d" % sem)
|
|
|
|
|
|
# ---- parsing -----------------------------------------------------------------
|
|
def parse_segment(buf, name="segment"):
|
|
"""Validate + decode a whole segment image. Raises RejectError."""
|
|
def rej(msg):
|
|
raise RejectError("%s: %s" % (name, msg))
|
|
|
|
if len(buf) < HDR_FIXED:
|
|
rej("file too short: %d bytes < fixed header %d" % (len(buf), HDR_FIXED))
|
|
if buf[0] != MAGIC_W or buf[1] != MAGIC_R:
|
|
rej("bad magic: %02X %02X (expect 57 52 'WR')" % (buf[0], buf[1]))
|
|
if buf[2] != FMT_VER:
|
|
rej("unsupported format version in magic byte 2: %d (expect %d); "
|
|
"this build of wave_read decodes v%d only"
|
|
% (buf[2], FMT_VER, FMT_VER))
|
|
ver = buf[OFF_VER]
|
|
if ver != FMT_VER:
|
|
rej("ver field = %d, magic version byte = %d (expect both %d)"
|
|
% (ver, buf[2], FMT_VER))
|
|
|
|
fs_hz = struct.unpack_from("<H", buf, OFF_FS)[0]
|
|
pps = buf[OFF_PPS]
|
|
chc = buf[OFF_CHC]
|
|
pts = struct.unpack_from("<H", buf, OFF_PTS)[0]
|
|
pre = struct.unpack_from("<H", buf, OFF_PRE)[0]
|
|
|
|
if chc == 0 or pts == 0:
|
|
rej("zero-length segment: ch_count=%d pts_per_ch=%d (nothing to "
|
|
"read; refusing, no picture will be produced)" % (chc, pts))
|
|
if chc != V1_CH_COUNT:
|
|
rej("ch_count=%d: version %d pins ch_count=%d (TEMP not recorded "
|
|
"in v1; TEMP needs a format version bump per section 6.3)"
|
|
% (chc, FMT_VER, V1_CH_COUNT))
|
|
if pps < 2:
|
|
rej("pts_per_cycle=%d unusable (< 2)" % pps)
|
|
if pts < pps:
|
|
rej("pts_per_ch=%d shorter than one cycle pts_per_cycle=%d" % (pts, pps))
|
|
if pre > pts:
|
|
rej("pre_pts=%d exceeds pts_per_ch=%d" % (pre, pts))
|
|
|
|
table_end = HDR_FIXED + chc * ENT_SIZE
|
|
crc_h_off = table_end
|
|
data_off = table_end + 2
|
|
data_len = pts * chc * 2
|
|
want = data_off + data_len + 2
|
|
if len(buf) != want:
|
|
rej("file size %d != expected %d (hdr 30 + table %d + 2 + data %d + 2)"
|
|
% (len(buf), want, chc * ENT_SIZE, data_len))
|
|
|
|
crc_h_calc = crc16_modbus(buf[0:table_end])
|
|
crc_h_stored = read_crc16(buf, crc_h_off)
|
|
if crc_h_calc != crc_h_stored:
|
|
rej("header CRC16 mismatch: stored 0x%04X calc 0x%04X over [0,%d) "
|
|
"-- segment rejected, nothing drawn"
|
|
% (crc_h_stored, crc_h_calc, table_end))
|
|
|
|
data = buf[data_off:data_off + data_len]
|
|
crc_d_calc = crc16_modbus(data)
|
|
crc_d_stored = read_crc16(buf, data_off + data_len)
|
|
if crc_d_calc != crc_d_stored:
|
|
rej("data CRC16 mismatch: stored 0x%04X calc 0x%04X over %d "
|
|
"bytes at offset %d -- segment rejected, nothing drawn"
|
|
% (crc_d_stored, crc_d_calc, data_len, data_off))
|
|
|
|
entries = []
|
|
for i in range(chc):
|
|
o = HDR_FIXED + i * ENT_SIZE
|
|
if buf[o + ENT_SEM] != i:
|
|
rej("channel table entry %d carries semantic id %d: section "
|
|
"6.3 pins ascending sem == index (UA=0 .. IN=7); whole "
|
|
"segment void" % (i, buf[o + ENT_SEM]))
|
|
entries.append({
|
|
"idx": i,
|
|
"sem": buf[o + ENT_SEM],
|
|
"src": buf[o + ENT_SRC],
|
|
"bits": buf[o + ENT_BITS],
|
|
"avail": buf[o + ENT_AVAIL],
|
|
"k_q16": struct.unpack_from("<i", buf, o + ENT_KQ)[0],
|
|
"b_codes": struct.unpack_from("<i", buf, o + ENT_B)[0],
|
|
"r_q16": struct.unpack_from("<i", buf, o + ENT_RQ16)[0],
|
|
})
|
|
|
|
samples = []
|
|
for i in range(chc):
|
|
samples.append(list(struct.unpack_from("<%dh" % pts, buf,
|
|
data_off + i * pts * 2)))
|
|
|
|
return {
|
|
"name": name,
|
|
"size": len(buf),
|
|
"ver": ver,
|
|
"fs_hz": fs_hz,
|
|
"pps": pps,
|
|
"chc": chc,
|
|
"pts": pts,
|
|
"pre": pre,
|
|
"trig_src": buf[OFF_TSRC],
|
|
"trig_elm": buf[OFF_TELM],
|
|
"t0_ms": struct.unpack_from("<I", buf, OFF_T0)[0],
|
|
"wall": struct.unpack_from("<I", buf, OFF_WALL)[0],
|
|
"eng": buf[OFF_ENG],
|
|
"flags_seq": buf[OFF_FLAGS],
|
|
"clip_mask": struct.unpack_from("<H", buf, OFF_CLIP)[0],
|
|
"frame_seq": struct.unpack_from("<I", buf, OFF_FSEQ)[0],
|
|
"crc_hdr": crc_h_stored,
|
|
"crc_data": crc_d_stored,
|
|
"entries": entries,
|
|
"samples": samples,
|
|
}
|
|
|
|
|
|
def load_segment(path):
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
raw = fh.read()
|
|
except OSError as exc:
|
|
raise IOErrorMessage(str(exc))
|
|
return parse_segment(raw, os.path.basename(path))
|
|
|
|
|
|
# ---- Fourier (mirror of App/algo/phasor.c phasor_calc) -----------------------
|
|
def phasor_calc(values, harm, n):
|
|
"""One full-cycle DFT on values (list length n), harmonics via idx step.
|
|
|
|
Identical convention to firmware: re sums sin, im sums cos, both tables
|
|
pre-multiplied sqrt(2)/N, so rms = sqrt(re^2 + im^2) directly and the
|
|
angle is atan2(im, re).
|
|
"""
|
|
scale = math.sqrt(2.0) / n
|
|
re = 0.0
|
|
im = 0.0
|
|
for i in range(n):
|
|
idx = (i * harm) % n
|
|
th = 2.0 * math.pi * idx / n
|
|
re += values[i] * math.sin(th) * scale
|
|
im += values[i] * math.cos(th) * scale
|
|
return re, im
|
|
|
|
|
|
def phasor_rms_ang(values, harm, n):
|
|
re, im = phasor_calc(values, harm, n)
|
|
return math.sqrt(re * re + im * im), math.degrees(math.atan2(im, re))
|
|
|
|
|
|
def channel_mv(seg, ci):
|
|
ent = seg["entries"][ci]
|
|
return [code_to_mv(ent, c) for c in seg["samples"][ci]]
|
|
|
|
|
|
# ---- shape metrics -------------------------------------------------------------
|
|
def zero_cross_cycles(vals):
|
|
last = 0
|
|
edges = 0
|
|
for v in vals:
|
|
s = 1 if v > 0 else (-1 if v < 0 else 0)
|
|
if s == 0:
|
|
continue
|
|
if last != 0 and s != last:
|
|
edges += 1
|
|
last = s
|
|
return edges / 2.0, edges
|
|
|
|
|
|
def half_wave_symmetry(vals, pps):
|
|
"""mean(|x[i]+x[i+half]|) / mean(|x[i]-dc|) after DC removal. 0 = ideal."""
|
|
if pps % 2:
|
|
return None
|
|
nfull = (len(vals) // pps) * pps
|
|
if nfull < pps:
|
|
return None
|
|
half = pps // 2
|
|
dc = sum(vals[:nfull]) / float(nfull)
|
|
num = 0.0
|
|
den = 0.0
|
|
for i in range(nfull - half):
|
|
a = vals[i] - dc
|
|
b = vals[i + half] - dc
|
|
num += abs(a + b)
|
|
den += abs(a)
|
|
if den <= 1e-9:
|
|
return 1.0
|
|
return num / den
|
|
|
|
|
|
def rail_pins(seg, ci, min_run):
|
|
"""Return (pin_lo, pin_hi, clipped_bool) on raw codes."""
|
|
ent = seg["entries"][ci]
|
|
lo_th, hi_th = rail_thresholds(ent)
|
|
codes = seg["samples"][ci]
|
|
lo = sum(1 for c in codes if c <= lo_th)
|
|
hi = sum(1 for c in codes if c >= hi_th)
|
|
return lo, hi, (lo >= min_run or hi >= min_run)
|
|
|
|
|
|
def measured_clip_mask(seg, min_run):
|
|
mask = 0
|
|
for ci in range(seg["chc"]):
|
|
_lo, _hi, clipped = rail_pins(seg, ci, min_run)
|
|
if clipped:
|
|
mask |= 1 << seg["entries"][ci]["sem"]
|
|
return mask
|
|
|
|
|
|
def mask_channels(mask):
|
|
return " ".join(sem_name(s) for s in sorted(SEM_NAMES) if mask & (1 << s))
|
|
|
|
|
|
# ---- subcommand: dump -------------------------------------------------------
|
|
def fmt_dump(seg):
|
|
lines = []
|
|
add = lines.append
|
|
add("FILE : %s (%d bytes)" % (seg["name"], seg["size"]))
|
|
add("magic : 57 52 %02X ('WR' + version byte %d)"
|
|
% (seg["ver"], seg["ver"]))
|
|
add("ver : %d" % seg["ver"])
|
|
add("fs_hz : %d%s" % (seg["fs_hz"],
|
|
"" if seg["fs_hz"] == 1000 else " (note: not the 1000 of V0.1)"))
|
|
add("pts_per_cycle : %d%s" % (seg["pps"],
|
|
"" if seg["pps"] == 20 else " (note: not the 20 of V0.1)"))
|
|
add("ch_count : %d%s" % (seg["chc"],
|
|
"" if seg["chc"] in (8, 9) else " (note: not 8/9)"))
|
|
add("pts_per_ch : %d (%d ms at %d Hz)"
|
|
% (seg["pts"], seg["pts"] * 1000 // max(1, seg["fs_hz"]),
|
|
seg["fs_hz"]))
|
|
add("pre_pts : %d" % seg["pre"])
|
|
add("trig_src : %d (%s)"
|
|
% (seg["trig_src"], TRG_NAMES.get(seg["trig_src"], "UNKNOWN")))
|
|
add("trig_elm : %d" % seg["trig_elm"])
|
|
add("t0_ms : %d (ms since power-on)" % seg["t0_ms"])
|
|
if seg["wall"]:
|
|
add("wall : %d (epoch s)" % seg["wall"])
|
|
else:
|
|
add("wall : 0 (time not synced at seal time)")
|
|
add("eng : %d (%s)"
|
|
% (seg["eng"], ENG_NAMES.get(seg["eng"], "UNKNOWN")))
|
|
valid = [sem_name(s) for s in sorted(SEM_NAMES)
|
|
if seg["flags_seq"] & (1 << s)]
|
|
add("flags_seq : 0x%02X VALID bits: %s"
|
|
% (seg["flags_seq"], " ".join(valid) if valid else "(none)"))
|
|
add("clip_mask : 0x%04X [%s]"
|
|
% (seg["clip_mask"], mask_channels(seg["clip_mask"]) or "(none)"))
|
|
add("frame_seq : %d" % seg["frame_seq"])
|
|
add("CRC16(hdr) : 0x%04X OK covers [0,%d), %s byte first"
|
|
% (seg["crc_hdr"], HDR_FIXED + seg["chc"] * ENT_SIZE,
|
|
"low" if CRC_LO_FIRST else "high"))
|
|
add("CRC16(data) : 0x%04X OK covers %d bytes, %s byte first"
|
|
% (seg["crc_data"], seg["pts"] * seg["chc"] * 2,
|
|
"low" if CRC_LO_FIRST else "high"))
|
|
add("-- channel table (%d entries x %d B) --" % (seg["chc"], ENT_SIZE))
|
|
add("%3s %3s %-5s %-10s %4s %5s %11s %9s %11s %12s %10s"
|
|
% ("IDX", "SEM", "NAME", "SRC", "BITS", "AVAIL", "K_Q16",
|
|
"B_CODES", "R_Q16", "LSB_uV", "FLAG"))
|
|
for ent in seg["entries"]:
|
|
fl = []
|
|
if ent["avail"] & ADC_CAL_F_VALID:
|
|
fl.append("VALID")
|
|
if ent["avail"] & ADC_CAL_F_CLIPPED:
|
|
fl.append("CLIPPED")
|
|
add("%3d %3d %-5s %-10s %4d 0x%02X %11d %9d %11d %12.6f %10s"
|
|
% (ent["idx"], ent["sem"], sem_name(ent["sem"]),
|
|
SRC_NAMES.get(ent["src"], "SRC%d" % ent["src"]),
|
|
ent["bits"], ent["avail"], ent["k_q16"], ent["b_codes"],
|
|
ent["r_q16"], lsb_uv(ent), "|".join(fl) if fl else "."))
|
|
# lead ruling 2026-10-29: soft notes, never a reject, never an exit code
|
|
both = [sem_name(e["sem"]) for e in seg["entries"]
|
|
if (e["avail"] & (ADC_CAL_F_VALID | ADC_CAL_F_CLIPPED))
|
|
== (ADC_CAL_F_VALID | ADC_CAL_F_CLIPPED)]
|
|
if both:
|
|
add("NOTE: VALID+CLIPPED(both) on: %s (normal once two-point "
|
|
"calibration is done, not a fault)" % " ".join(both))
|
|
agg = 0
|
|
for e in seg["entries"]:
|
|
if e["avail"] & ADC_CAL_F_CLIPPED:
|
|
agg |= 1 << e["sem"]
|
|
if agg != seg["clip_mask"]:
|
|
add("WARN: avail CLIPPED bits aggregate to 0x%04X, header clip_mask "
|
|
"is 0x%04X -- soft check, segment NOT rejected"
|
|
% (agg, seg["clip_mask"]))
|
|
# eng derives from flags_seq on the board: feed_eng() in
|
|
# App/app/ui_data_feed.c -- all 8 electrical channels VALID -> secondary
|
|
# (1), else pin-domain (0); wave header fill App/task/adc_task.c.
|
|
# Cross-reading allowed here, SOFT ONLY: never a reject, no exit code.
|
|
all_valid = (seg["flags_seq"] & 0xFF) == 0xFF
|
|
if all_valid and seg["eng"] == 0:
|
|
add("WARN: eng=0 (pin-domain) but flags_seq=0xFF (all 8 VALID) -- "
|
|
"derived eng would be 1 on board; soft check, NOT rejected")
|
|
elif not all_valid and seg["eng"] != 0:
|
|
add("WARN: eng=%d (not pin-domain) but flags_seq=0x%02X is not "
|
|
"all-VALID -- derived eng would be 0 on board; soft check, "
|
|
"NOT rejected" % (seg["eng"], seg["flags_seq"]))
|
|
return lines
|
|
|
|
|
|
def cmd_dump(args):
|
|
seg = load_segment(args.file)
|
|
for line in fmt_dump(seg):
|
|
print(line)
|
|
return 0
|
|
|
|
|
|
# ---- subcommand: fft ----------------------------------------------------------
|
|
def _harm_rms(win, harm, n):
|
|
if harm * 2 > n:
|
|
return None
|
|
rms, _ang = phasor_rms_ang(win, harm, n)
|
|
return rms
|
|
|
|
|
|
def _fmt_or_na(v):
|
|
return "%.2f" % v if v is not None else "n/a"
|
|
|
|
|
|
def fmt_fft(seg, cycle, all_cycles, only_ch):
|
|
lines = []
|
|
pps = seg["pps"]
|
|
ncyc = seg["pts"] // pps
|
|
if cycle is None:
|
|
cycle = ncyc - 1
|
|
picked = range(ncyc) if all_cycles else [cycle]
|
|
lines.append("unit: mV eng=%d (%s)%s"
|
|
% (seg["eng"], ENG_NAMES.get(seg["eng"], "UNKNOWN"),
|
|
" [pin-domain = ADC pin voltage, NOT calibrated "
|
|
"secondary]" if seg["eng"] == 0 else ""))
|
|
lines.append("%3s %-5s %-10s %6s %10s %9s %10s %10s %s"
|
|
% ("CH", "NAME", "SRC", "CYCLE", "RMS-f[mV]", "ANG-f[deg]",
|
|
"RMS-2x[mV]", "RMS-5x[mV]", "NOTE"))
|
|
lines.append(" (f = fundamental; 2x/5x = 2nd and 5th harmonic; "
|
|
"angle reference = first sample of this window, not a phase "
|
|
"lock to the signal generator)")
|
|
for ci in only_ch:
|
|
ent = seg["entries"][ci]
|
|
mv = channel_mv(seg, ci)
|
|
rmss = []
|
|
for c in picked:
|
|
if c < 0 or (c + 1) * pps > seg["pts"]:
|
|
lines.append("%3d %-5s %-10s %6d (out of range, cycles "
|
|
"0..%d)"
|
|
% (ci, sem_name(ent["sem"]),
|
|
SRC_NAMES.get(ent["src"], "?"), c, ncyc - 1))
|
|
continue
|
|
win = mv[c * pps:(c + 1) * pps]
|
|
rms1, ang1 = phasor_rms_ang(win, 1, pps)
|
|
rms2 = _harm_rms(win, 2, pps)
|
|
rms5 = _harm_rms(win, 5, pps)
|
|
rmss.append(rms1)
|
|
note = "CLIP-bit" if seg["clip_mask"] & (1 << ent["sem"]) else ""
|
|
if (ent["avail"] & (ADC_CAL_F_VALID | ADC_CAL_F_CLIPPED)) \
|
|
== (ADC_CAL_F_VALID | ADC_CAL_F_CLIPPED):
|
|
note = (note + " VALID+CLIPPED(both)").strip()
|
|
lines.append("%3d %-5s %-10s %6d %10.2f %+9.2f %10s %10s %s"
|
|
% (ci, sem_name(ent["sem"]),
|
|
SRC_NAMES.get(ent["src"], "?"), c, rms1, ang1,
|
|
_fmt_or_na(rms2), _fmt_or_na(rms5), note))
|
|
if all_cycles and len(rmss) > 1:
|
|
lines.append("%3d %-5s %-10s %6s mean=%10.2f min=%10.2f "
|
|
"max=%10.2f over %d cycles"
|
|
% (ci, sem_name(ent["sem"]),
|
|
SRC_NAMES.get(ent["src"], "?"), "ALL",
|
|
sum(rmss) / len(rmss), min(rmss), max(rmss),
|
|
len(rmss)))
|
|
lines.append("note: mV = (code + b_codes) * k_q16/65536/1000, then "
|
|
"* r_q16/65536 when r_q16 != 0 (adc_cal_apply formula, same "
|
|
"two integer steps as the C side; result is in the "
|
|
"recording domain, labeled by header eng, never "
|
|
"re-guessed); full-cycle window, coefficients = "
|
|
"sin/cos(2*pi*k/N)*sqrt(2)/N as in App/algo/phasor.c; "
|
|
"phase = atan2(im,re) deg")
|
|
return lines
|
|
|
|
|
|
def cmd_fft(args):
|
|
seg = load_segment(args.file)
|
|
if args.ch is None:
|
|
only = range(seg["chc"])
|
|
elif 0 <= args.ch < seg["chc"]:
|
|
only = [args.ch]
|
|
else:
|
|
print("usage error: --ch %d out of range (0..%d)"
|
|
% (args.ch, seg["chc"] - 1), file=sys.stderr)
|
|
return 2
|
|
for line in fmt_fft(seg, args.cycle, args.all, only):
|
|
print(line)
|
|
return 0
|
|
|
|
|
|
# ---- subcommand: shape ---------------------------------------------------------
|
|
def fmt_shape(seg, min_run, strict_out=None, superset=False):
|
|
lines = []
|
|
mv_unit = "mV eng=%d(%s)" % (seg["eng"],
|
|
ENG_NAMES.get(seg["eng"], "?"))
|
|
lines.append("unit: %s min-run for rail pin = %d samples"
|
|
% (mv_unit, min_run))
|
|
lines.append("%3s %-5s %-10s %6s %10s %4s %6s %7s %7s %5s"
|
|
% ("CH", "NAME", "SRC", "CYCLES", "AMP[mVpk]", "POL",
|
|
"SYM", "PIN_LO", "PIN_HI", "RAIL"))
|
|
for ci in range(seg["chc"]):
|
|
ent = seg["entries"][ci]
|
|
codes = seg["samples"][ci]
|
|
mv = channel_mv(seg, ci)
|
|
mean = sum(mv) / len(mv)
|
|
cycles, _edges = zero_cross_cycles([v - mean for v in mv])
|
|
amp = (max(mv) - min(mv)) / 2.0
|
|
first = codes[0]
|
|
pol = "+" if first > 0 else ("-" if first < 0 else "0")
|
|
sym = half_wave_symmetry(mv, seg["pps"])
|
|
lo, hi, clipped = rail_pins(seg, ci, min_run)
|
|
lines.append("%3d %-5s %-10s %6.2f %10.2f %4s %6s %7d %7d %5s"
|
|
% (ci, sem_name(ent["sem"]),
|
|
SRC_NAMES.get(ent["src"], "?"), cycles, amp, pol,
|
|
"%.3f" % sym if sym is not None else "n/a",
|
|
lo, hi, "YES" if clipped else "."))
|
|
meas = measured_clip_mask(seg, min_run)
|
|
hdr = seg["clip_mask"]
|
|
if superset:
|
|
ok = (meas & ~hdr) == 0
|
|
verdict = "SUPERSET-OK" if ok else "MISMATCH"
|
|
else:
|
|
ok = meas == hdr
|
|
verdict = "MATCH" if ok else "MISMATCH"
|
|
lines.append("clip reconcile: measured=0x%04X [%s] header=0x%04X [%s] %s"
|
|
% (meas, mask_channels(meas) or "-", hdr,
|
|
mask_channels(hdr) or "-", verdict))
|
|
if superset:
|
|
tol = hdr & ~meas
|
|
lines.append("yardstick: superset (--clip-superset) -- every "
|
|
"measured rail pin must be claimed by the seal; "
|
|
"sealed-but-not-railed tolerated: %s"
|
|
% (mask_channels(tol) if tol else "(none)"))
|
|
else:
|
|
lines.append("yardstick: equality (default card check) -- measured "
|
|
"rail pins must equal the sealed clip_mask")
|
|
lines.append("note: header clip_mask is the seal-time CLIPPED STAMP, "
|
|
"source-derived in V0.1 (App/algo/adc_cal.c: "
|
|
"s_rom_k_stm32 -> CLIPPED, unipolar no-bias), NOT measured "
|
|
"from samples; measured is this segment's rail pins. Real "
|
|
"V0.1 AC waves bottom-rail the 5 stamped channels, so "
|
|
"equality is expected on-board; a non-railed measurement "
|
|
"means synthetic fixture (e.g. column ramps) or wiring "
|
|
"change -- investigate, do not auto-blame the seal")
|
|
if strict_out is not None:
|
|
strict_out.append("MATCH" if ok else "MISMATCH")
|
|
return lines
|
|
|
|
|
|
def cmd_shape(args):
|
|
if args.clip_superset and not args.strict:
|
|
print("usage error: --clip-superset only changes the --strict "
|
|
"yardstick; pass --strict as well", file=sys.stderr)
|
|
return 2
|
|
seg = load_segment(args.file)
|
|
verdict = []
|
|
for line in fmt_shape(seg, args.min_run, verdict,
|
|
superset=args.clip_superset):
|
|
print(line)
|
|
if args.strict and verdict and verdict[0] != "MATCH":
|
|
if args.clip_superset:
|
|
print("STRICT-SUPERSET: measured rail pins OUTSIDE the sealed "
|
|
"clip_mask -- low injection cannot explain this; "
|
|
"investigate wiring or the seal")
|
|
else:
|
|
print("STRICT: clip_mask reconciliation failed (yardstick: "
|
|
"equality; for DC-only or low injection use --strict "
|
|
"--clip-superset)")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
# ---- subcommand: plot (pure-string SVG, refuses on any reject) ---------------
|
|
def svg_for_channel(seg, ci, seg_label):
|
|
ent = seg["entries"][ci]
|
|
mv = channel_mv(seg, ci)
|
|
w, h = 900, 380
|
|
ml, mr, mt, mb = 60, 20, 34, 30
|
|
pw, ph = w - ml - mr, h - mt - mb
|
|
span = max(max(mv), -min(mv), 1e-6) * 1.1
|
|
n = len(mv)
|
|
|
|
def xs(i):
|
|
return ml + pw * i / float(n - 1)
|
|
|
|
def ys(v):
|
|
return mt + ph / 2.0 - (v / span) * (ph / 2.0)
|
|
|
|
name = sem_name(ent["sem"])
|
|
trg = TRG_NAMES.get(seg["trig_src"], "SRC?%d" % seg["trig_src"])
|
|
clip_bit = bool(seg["clip_mask"] & (1 << ent["sem"]))
|
|
title = "%s seg=%s trig=%s elm=%d eng=%s%s" % (
|
|
name, seg_label, trg, seg["trig_elm"],
|
|
ENG_NAMES.get(seg["eng"], "?"), " CLIP" if clip_bit else "")
|
|
parts = []
|
|
parts.append('<?xml version="1.0" encoding="UTF-8"?>')
|
|
parts.append('<svg xmlns="http://www.w3.org/2000/svg" width="%d" '
|
|
'height="%d" viewBox="0 0 %d %d">' % (w, h, w, h))
|
|
parts.append('<rect x="0" y="0" width="%d" height="%d" fill="white"/>'
|
|
% (w, h))
|
|
parts.append('<text x="%d" y="20" font-family="monospace" font-size="14" '
|
|
'fill="black">%s</text>' % (ml, title))
|
|
parts.append('<rect x="%d" y="%d" width="%d" height="%d" fill="none" '
|
|
'stroke="black" stroke-width="1"/>' % (ml, mt, pw, ph))
|
|
zy = ys(0.0)
|
|
parts.append('<line x1="%d" y1="%.2f" x2="%d" y2="%.2f" stroke="black" '
|
|
'stroke-width="1" stroke-dasharray="5,4"/>'
|
|
% (ml, zy, ml + pw, zy))
|
|
parts.append('<text x="%d" y="%.2f" font-family="monospace" font-size="10" '
|
|
'fill="black">0</text>' % (ml - 14, zy + 3))
|
|
parts.append('<text x="%d" y="%.2f" font-family="monospace" font-size="10" '
|
|
'fill="black">%.0fmV</text>' % (ml - 55, mt + 8, span))
|
|
parts.append('<text x="%d" y="%.2f" font-family="monospace" font-size="10" '
|
|
'fill="black">%.0fmV</text>' % (ml - 55, mt + ph, -span))
|
|
tx = xs(min(seg["pre"], n - 1))
|
|
parts.append('<line x1="%.2f" y1="%d" x2="%.2f" y2="%d" stroke="black" '
|
|
'stroke-width="1" stroke-dasharray="2,3"/>'
|
|
% (tx, mt, tx, mt + ph))
|
|
parts.append('<text x="%.2f" y="%d" font-family="monospace" font-size="10" '
|
|
'fill="black">TRIG(pre=%d)</text>'
|
|
% (min(tx + 3, w - 120), mt + ph + 12, seg["pre"]))
|
|
d = ["M %.2f %.2f" % (xs(0), ys(mv[0]))]
|
|
for i in range(1, n):
|
|
d.append("L %.2f %.2f" % (xs(i), ys(mv[i - 1])))
|
|
d.append("L %.2f %.2f" % (xs(i), ys(mv[i])))
|
|
parts.append('<path d="%s" fill="none" stroke="black" stroke-width="1"/>'
|
|
% " ".join(d))
|
|
parts.append('<text x="%d" y="%d" font-family="monospace" font-size="10" '
|
|
'fill="black">%d pts @ %d Hz, %d pts/cycle, x = sample '
|
|
'index</text>'
|
|
% (ml, h - 6, n, seg["fs_hz"], seg["pps"]))
|
|
parts.append("</svg>")
|
|
return "\n".join(parts) + "\n"
|
|
|
|
|
|
def default_outdir():
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))))
|
|
return os.path.join(root, "ci-artifacts")
|
|
|
|
|
|
def cmd_plot(args):
|
|
seg = load_segment(args.file) # any integrity failure raises BEFORE write
|
|
if args.out is not None and args.ch is None:
|
|
print("usage error: plot --out requires --ch (one file per channel)",
|
|
file=sys.stderr)
|
|
return 2
|
|
if args.ch is None:
|
|
chans = range(seg["chc"])
|
|
elif 0 <= args.ch < seg["chc"]:
|
|
chans = [args.ch]
|
|
else:
|
|
print("usage error: --ch %d out of range (0..%d)"
|
|
% (args.ch, seg["chc"] - 1), file=sys.stderr)
|
|
return 2
|
|
label = os.path.splitext(os.path.basename(args.file))[0]
|
|
written = []
|
|
for ci in chans:
|
|
if args.out is not None:
|
|
path = args.out
|
|
else:
|
|
outdir = args.outdir if args.outdir else default_outdir()
|
|
if not os.path.isdir(outdir):
|
|
os.makedirs(outdir)
|
|
path = os.path.join(
|
|
outdir, "%s_ch%02d_%s.svg"
|
|
% (label, ci, sem_name(seg["entries"][ci]["sem"])))
|
|
svg = svg_for_channel(seg, ci, label)
|
|
with open(path, "w", encoding="utf-8", newline="\n") as fh:
|
|
fh.write(svg)
|
|
written.append(path)
|
|
print("WROTE %s" % path)
|
|
if not written:
|
|
print("plot: nothing written")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
# ---- synthetic segment builders (selftest) ------------------------------------
|
|
def syn_entries(chc=8, avail_mode="v01", ratio_mode="pin"):
|
|
"""avail = raw pass-through of adc_cal_flags(). Modes (lead ruling
|
|
2026-10-29 asks for the first two as interop shapes):
|
|
v01 = V0.1 board truth, pre-calibration: TPAFE trio 0x00, STM32 five
|
|
0x02; their CLIPPED bits aggregate 2+4+32+64+128 == 0xE6 header
|
|
both = deliberately VALID+CLIPPED on the five: 0x01 TPAFE / 0x03 STM32
|
|
(legal: soft note only, must still parse clean)
|
|
none = all-zero avail under clip_mask 0xE6 (avail/header drift; soft
|
|
WARN only, must still parse clean)
|
|
|
|
ratio_mode picks the v2 r_q16 snapshot (ROM defaults from adc_cal.h):
|
|
rom = board defaults: voltage family 65536000 (1000.0), current family
|
|
1927532 (29.4118 A/V) -- sem 0/3 voltage (UA/UN), 4 current (IA)
|
|
pin = all zero r_q16 (no scaling; output stays pin-domain mV)
|
|
one = all 65536 (ratio exactly 1.0: must equal the old pin-domain value)
|
|
"""
|
|
pa_t, pa_s = {"v01": (0x00, 0x02), "both": (0x01, 0x03),
|
|
"none": (0x00, 0x00)}[avail_mode]
|
|
rom = {0: 65536000, 3: 65536000, 4: 1927532}
|
|
ents = []
|
|
for sem in range(chc):
|
|
if sem in (0, 3, 4):
|
|
kq16 = 10000000
|
|
src, bits, avail = SRC_TPAFE, 16, pa_t
|
|
else:
|
|
kq16 = 52800000
|
|
src, bits, avail = SRC_STM32, 12, pa_s
|
|
if ratio_mode == "rom":
|
|
rq16 = rom.get(sem, 0)
|
|
elif ratio_mode == "pin":
|
|
rq16 = 0
|
|
elif ratio_mode == "one":
|
|
rq16 = 65536
|
|
else:
|
|
raise ValueError("unknown ratio_mode %r" % (ratio_mode,))
|
|
ents.append((sem, src, bits, avail, kq16, 0, rq16))
|
|
return ents
|
|
|
|
|
|
def syn_samples(pts=320, pps=20, chc=8, rail=True):
|
|
"""sem0: sine A=9261 phi=+90deg | sem3: A=4630 phi=+30deg | sem4: A=2000
|
|
phi=-60deg | STM32 five (UB UC IB IC IN): 10-low/10-high square pinned
|
|
at the bottom rail (0 then 2048) -- honest V0.1 shape, no midpoint
|
|
bias. rail=False: the five become a gentle mid-band ramp
|
|
500 + (97*i) % 3000, i.e. values 500..3499 -- never enters the rail
|
|
bands (0..81 / 4013..4095), so measured clip is EMPTY while the header
|
|
still stamps 0xE6: one fixture that shows BOTH clip yardsticks."""
|
|
phi = {0: math.pi / 2.0, 3: math.pi / 6.0, 4: -math.pi / 3.0}
|
|
amp = {0: 9261, 3: 4630, 4: 2000}
|
|
cols = []
|
|
for sem in range(chc):
|
|
if sem in amp:
|
|
a = amp[sem]
|
|
p = phi[sem]
|
|
cols.append([int(round(a * math.sin(2.0 * math.pi * i / pps + p)))
|
|
for i in range(pts)])
|
|
elif rail:
|
|
cols.append([0 if (i // (pps // 2)) % 2 == 0 else 2048
|
|
for i in range(pts)])
|
|
else:
|
|
cols.append([500 + (97 * i) % 3000 for i in range(pts)])
|
|
return cols
|
|
|
|
|
|
def build_synthetic(chc=8, pts=320, pps=20, trig_src=1, clip=0x00E6,
|
|
flags=0x00, wall=0, t0=123456, fseq=987654321,
|
|
avail_mode="v01", eng=0, rail=True, ratio_mode="pin"):
|
|
hdr = bytearray()
|
|
hdr += bytes([MAGIC_W, MAGIC_R, FMT_VER, FMT_VER])
|
|
hdr += struct.pack("<H", 1000)
|
|
hdr += bytes([pps, chc])
|
|
hdr += struct.pack("<H", pts)
|
|
hdr += struct.pack("<H", pts // 2)
|
|
hdr += bytes([trig_src, 0])
|
|
hdr += struct.pack("<I", t0)
|
|
hdr += struct.pack("<I", wall)
|
|
hdr += bytes([eng, flags])
|
|
hdr += struct.pack("<H", clip)
|
|
hdr += struct.pack("<I", fseq)
|
|
assert len(hdr) == HDR_FIXED
|
|
for sem, src, bits, avail, kq16, b_codes, rq16 in syn_entries(
|
|
chc, avail_mode, ratio_mode)[:chc]:
|
|
hdr += bytes([sem, src, bits, avail])
|
|
hdr += struct.pack("<iii", kq16, b_codes, rq16)
|
|
hdr += put_crc16(crc16_modbus(hdr))
|
|
data = b"".join(struct.pack("<%dh" % pts, *col)
|
|
for col in syn_samples(pts, pps, chc, rail))
|
|
return bytes(hdr) + data + put_crc16(crc16_modbus(data))
|
|
|
|
|
|
def mutate(buf, off, xor=0xFF):
|
|
b = bytearray(buf)
|
|
b[off] ^= xor
|
|
return bytes(b)
|
|
|
|
|
|
# ---- selftest -------------------------------------------------------------------
|
|
class Checker(object):
|
|
def __init__(self):
|
|
self.ok = 0
|
|
self.fail = 0
|
|
self.problems = []
|
|
|
|
def check(self, name, cond, detail=""):
|
|
if cond:
|
|
self.ok += 1
|
|
print("[OK] %s" % name)
|
|
else:
|
|
self.fail += 1
|
|
self.problems.append(name)
|
|
print("[FAIL] %s %s" % (name, detail))
|
|
|
|
def near(self, name, got, want, rel_tol=0.01, abs_tol=0.0):
|
|
good = abs(got - want) <= max(abs_tol, rel_tol * abs(want))
|
|
self.check(name, good,
|
|
"got=%r want=%r tol=%s" % (got, want, rel_tol))
|
|
|
|
|
|
def expect_reject(ck, name, buf, frag):
|
|
try:
|
|
parse_segment(buf, name="case")
|
|
except RejectError as exc:
|
|
ck.check(name, frag.lower() in str(exc).lower(),
|
|
"reason='%s' lacks '%s'" % (exc, frag))
|
|
return
|
|
ck.check(name, False, "expected RejectError, parse succeeded")
|
|
|
|
|
|
def selftest():
|
|
ck = Checker()
|
|
# 1. CRC-16/MODBUS known-answer test (spec-mandated check value)
|
|
ck.check("crc16 KAT '123456789' -> 0x4B37",
|
|
crc16_modbus(b"123456789") == CRC16_CHECK,
|
|
"got 0x%04X" % crc16_modbus(b"123456789"))
|
|
|
|
# 2. synthetic 8-channel segment builds and round-trips through the
|
|
# parser (both CRC gates passing inside parse == implicit PASS)
|
|
img = build_synthetic()
|
|
seg = parse_segment(img, name="synth8")
|
|
ck.check("size 8ch/320pts == 5282", seg["size"] == 5282,
|
|
"got %d" % seg["size"])
|
|
fields = dict(fs_hz=1000, pps=20, chc=8, pts=320, pre=160, trig_src=1,
|
|
trig_elm=0, t0_ms=123456, wall=0, eng=0, flags_seq=0,
|
|
clip_mask=0x00E6, frame_seq=987654321)
|
|
for k in sorted(fields):
|
|
ck.check("field %s round-trip" % k, seg[k] == fields[k],
|
|
"got %r" % (seg[k],))
|
|
ck.check("table entry 0 == sem0/src0/bits16/avail0x00/k_q16 10000000/"
|
|
"r_q16 0 (pin mode)",
|
|
(seg["entries"][0]["sem"], seg["entries"][0]["src"],
|
|
seg["entries"][0]["bits"], seg["entries"][0]["avail"],
|
|
seg["entries"][0]["k_q16"],
|
|
seg["entries"][0]["r_q16"]) == (0, 0, 16, 0x00, 10000000, 0))
|
|
ck.check("table entry 1 == sem1/src1/bits12/avail0x02/k_q16 52800000/"
|
|
"r_q16 0 (pin mode)",
|
|
(seg["entries"][1]["sem"], seg["entries"][1]["src"],
|
|
seg["entries"][1]["bits"], seg["entries"][1]["avail"],
|
|
seg["entries"][1]["k_q16"],
|
|
seg["entries"][1]["r_q16"]) == (1, 1, 12, 0x02, 52800000, 0))
|
|
# per-entry availability (low 8 bits of adc_cal_flags) vs header
|
|
# clip_mask: the CLIPPED bits must reconcile, bit s = sem s both sides
|
|
clip_from_avail = 0
|
|
for ent in seg["entries"]:
|
|
if ent["avail"] & ADC_CAL_F_CLIPPED:
|
|
clip_from_avail |= 1 << ent["sem"]
|
|
ck.check("avail CLIPPED bits reconcile header clip_mask",
|
|
clip_from_avail == seg["clip_mask"],
|
|
"avail says 0x%04X header 0x%04X"
|
|
% (clip_from_avail, seg["clip_mask"]))
|
|
|
|
# 2b. lead ruling 2026-10-29: avail is raw pass-through, NEVER a reject
|
|
# basis; VALID+CLIPPED(both) and avail/header drift are soft only.
|
|
img_b = build_synthetic(avail_mode="both", flags=0xFF, fseq=71)
|
|
segb = parse_segment(img_b, name="synth-both") # raises if wrongly voided
|
|
ck.check("both-form (flags 0xFF, avail 0x01/0x03) parses, not rejected",
|
|
segb["size"] == 5282)
|
|
ck.check("dump both-form carries VALID+CLIPPED(both) note",
|
|
"VALID+CLIPPED(both)" in "\n".join(fmt_dump(segb)))
|
|
fftb = "\n".join(fmt_fft(segb, None, False, range(segb["chc"])))
|
|
ck.check("fft both-form NOTE column carries VALID+CLIPPED(both)",
|
|
"VALID+CLIPPED(both)" in fftb)
|
|
vb = []
|
|
fmt_shape(segb, 4, vb)
|
|
ck.check("both-form shape verdict unchanged: MATCH (avail does not "
|
|
"drive strict)", bool(vb) and vb[0] == "MATCH", str(vb))
|
|
img_v0 = build_synthetic(flags=0x00, fseq=72) # V0.1 board truth form
|
|
segv0 = parse_segment(img_v0, name="synth-v01")
|
|
ck.check("V0.1 truth form (flags_seq 0x00, avail 0x00/0x02) parses clean",
|
|
segv0["flags_seq"] == 0x00)
|
|
dumpv0 = "\n".join(fmt_dump(segv0))
|
|
ck.check("V0.1 truth dump has no both-note and no WARN",
|
|
"VALID+CLIPPED(both)" not in dumpv0 and "WARN" not in dumpv0)
|
|
segn = parse_segment(build_synthetic(avail_mode="none", flags=0xFF,
|
|
fseq=73),
|
|
name="synth-drift") # all-zero vs clip 0xE6
|
|
ck.check("avail/header drift parses (soft, not rejected)",
|
|
segn["clip_mask"] == 0x00E6)
|
|
ck.check("avail/header drift dump carries WARN line",
|
|
"WARN" in "\n".join(fmt_dump(segn)))
|
|
|
|
# 2c. eng x flags_seq derivation (lead follow-up ruling 2026-10-29):
|
|
# mismatch = soft WARN in dump, never a reject, never an exit code.
|
|
segx = parse_segment(img_b, name="synth-eng-cross") # eng=0 + flags 0xFF
|
|
dumpx = "\n".join(fmt_dump(segx))
|
|
ck.check("eng=0 with flags_seq=0xFF still parses with soft eng WARN",
|
|
segx["eng"] == 0 and "WARN: eng=0" in dumpx)
|
|
segok = parse_segment(build_synthetic(avail_mode="both", flags=0xFF,
|
|
eng=1, fseq=74),
|
|
name="synth-eng-pair") # derived form: eng 1
|
|
ck.check("eng=1 with all-VALID parses, no eng WARN",
|
|
"WARN: eng" not in "\n".join(fmt_dump(segok)))
|
|
seginv = parse_segment(build_synthetic(avail_mode="v01", flags=0x00,
|
|
eng=1, fseq=75),
|
|
name="synth-eng-inv") # not all-VALID + eng 1
|
|
ck.check("eng=1 without all-VALID parses with soft eng WARN (not void)",
|
|
"WARN: eng=1" in "\n".join(fmt_dump(seginv)))
|
|
|
|
# 2d. clip yardsticks (lead ruling 2026-10-29): equality is the default
|
|
# card check; --clip-superset is the explicit low-injection ruler.
|
|
# Prove both on ONE non-railed fixture at REAL process exit-code
|
|
# level, not just in-process verdicts.
|
|
imgr = build_synthetic(rail=False, fseq=76)
|
|
segr = parse_segment(imgr, name="synth-ramp")
|
|
v_eq = []
|
|
out_eq = "\n".join(fmt_shape(segr, 4, v_eq))
|
|
ck.check("ramp fixture: equality yardstick MISMATCH (strict rc 1)",
|
|
bool(v_eq) and v_eq[0] == "MISMATCH")
|
|
v_sup = []
|
|
out_sup = "\n".join(fmt_shape(segr, 4, v_sup, superset=True))
|
|
ck.check("ramp fixture: superset yardstick passes (strict rc 0)",
|
|
bool(v_sup) and v_sup[0] == "MATCH" and "SUPERSET-OK" in out_sup)
|
|
ck.check("output names the yardstick: equality",
|
|
"yardstick: equality" in out_eq)
|
|
ck.check("output names the yardstick: superset + tolerance list",
|
|
"yardstick: superset" in out_sup and
|
|
"UB UC IB IC IN" in out_sup)
|
|
with tempfile.TemporaryDirectory() as td2:
|
|
rp = os.path.join(td2, "RAMP.BIN")
|
|
with open(rp, "wb") as fh:
|
|
fh.write(imgr)
|
|
me = sys.executable
|
|
script = os.path.abspath(__file__)
|
|
rc_eq = subprocess.run([me, script, "shape", rp, "--strict"],
|
|
capture_output=True, text=True)
|
|
rc_sup = subprocess.run([me, script, "shape", rp, "--strict",
|
|
"--clip-superset"],
|
|
capture_output=True, text=True)
|
|
rc_alone = subprocess.run([me, script, "shape", rp,
|
|
"--clip-superset"],
|
|
capture_output=True, text=True)
|
|
ck.check("CLI real rc: --strict on non-railed fixture == 1",
|
|
rc_eq.returncode == 1, "rc=%d" % rc_eq.returncode)
|
|
ck.check("CLI real rc: --strict --clip-superset == 0",
|
|
rc_sup.returncode == 0, "rc=%d" % rc_sup.returncode)
|
|
ck.check("CLI real rc: --clip-superset alone == usage error 2",
|
|
rc_alone.returncode == 2,
|
|
"rc=%d" % rc_alone.returncode)
|
|
ck.check("clip_mask 0xE6 decodes to UB UC IB IC IN",
|
|
mask_channels(seg["clip_mask"]) == "UB UC IB IC IN",
|
|
mask_channels(seg["clip_mask"]))
|
|
|
|
# 3. variant header values (bench trigger + synced wall + new frame_seq)
|
|
# must survive the byte round-trip; v1 also pins ch_count to 8
|
|
img_v = build_synthetic(trig_src=3, wall=1793599000, fseq=55, t0=7)
|
|
segv = parse_segment(img_v, name="synth-variant")
|
|
ck.check("variant trig_src round-trip", segv["trig_src"] == 3)
|
|
ck.check("variant wall round-trip", segv["wall"] == 1793599000)
|
|
ck.check("variant frame_seq round-trip", segv["frame_seq"] == 55)
|
|
ck.check("variant t0_ms round-trip", segv["t0_ms"] == 7)
|
|
|
|
# 4. Fourier: analytic expectations for the synthesized sines.
|
|
# RMS = A*lsb_uV/1000/sqrt(2) mV; angle = phi (phasor.c convention).
|
|
pps = seg["pps"]
|
|
for ci, amp_codes, phi_deg in ((0, 9261, 90.0), (3, 4630, 30.0),
|
|
(4, 2000, -60.0)):
|
|
ent = seg["entries"][ci]
|
|
mv = channel_mv(seg, ci)
|
|
win = mv[-pps:]
|
|
rms1, ang1 = phasor_rms_ang(win, 1, pps)
|
|
want_rms = amp_codes * ent_scale_mv(ent) / math.sqrt(2.0)
|
|
ck.near("fft ch%d RMS" % ci, rms1, want_rms, 0.01)
|
|
ck.near("fft ch%d phase (atan2(im,re) convention)" % ci, ang1,
|
|
phi_deg, 0.01, abs_tol=2.0)
|
|
rms2 = _harm_rms(win, 2, pps)
|
|
rms5 = _harm_rms(win, 5, pps)
|
|
ck.check("fft ch%d pure sine: H2/H5 << H1" % ci,
|
|
rms2 < 0.01 * rms1 and rms5 < 0.01 * rms1,
|
|
"rms1=%.3f rms2=%.3f rms5=%.3f" % (rms1, rms2, rms5))
|
|
|
|
# 5. shape: cycles, polarity, symmetry, rail pins vs clip_mask
|
|
vmeas = []
|
|
fmt_shape(seg, 4, vmeas)
|
|
ck.check("shape clip reconcile MATCH (0xE6)",
|
|
bool(vmeas) and vmeas[0] == "MATCH", str(vmeas))
|
|
ua = channel_mv(seg, 0)
|
|
mean = sum(ua) / len(ua)
|
|
cycles, _ = zero_cross_cycles([v - mean for v in ua])
|
|
ck.near("shape UA cycles == 16.0", cycles, 16.0, 0.0, abs_tol=0.3)
|
|
sym = half_wave_symmetry(ua, pps)
|
|
ck.check("shape UA half-wave symmetry near 0",
|
|
sym is not None and sym < 0.05, "sym=%r" % sym)
|
|
ck.check("shape UA polarity '+'", seg["samples"][0][0] > 0)
|
|
ck.check("shape IA polarity '-' (phi=-60deg)", seg["samples"][4][0] < 0)
|
|
lo1, hi1, clipped1 = rail_pins(seg, 1, 4)
|
|
ck.check("shape UB pinned bottom 160 samples",
|
|
clipped1 and lo1 == 160 and hi1 == 0,
|
|
"lo=%d hi=%d" % (lo1, hi1))
|
|
ck.near("shape measured mask == 0xE6", measured_clip_mask(seg, 4),
|
|
0x00E6, 0.0, abs_tol=0)
|
|
_lo0, _hi0, clip0 = rail_pins(seg, 0, 4)
|
|
ck.check("shape UA (1V rms sine) not railed", not clip0)
|
|
|
|
# 6. plot: SVG content and honest labels; reject path writes nothing
|
|
with tempfile.TemporaryDirectory() as td:
|
|
svg = svg_for_channel(seg, 0, SEG_LABEL_SAMPLE)
|
|
p0 = os.path.join(td, "a.svg")
|
|
with open(p0, "w", encoding="utf-8", newline="\n") as fh:
|
|
fh.write(svg)
|
|
with open(p0, "r", encoding="utf-8") as fh:
|
|
back = fh.read()
|
|
ck.check("svg well-formed-ish",
|
|
back.startswith("<?xml") and "<svg" in back
|
|
and back.rstrip().endswith("</svg>"))
|
|
for token in ("UA", SEG_LABEL_SAMPLE, "TRIG(pre=160)", "dasharray", "MANUAL"):
|
|
ck.check("svg contains %r" % token, token in back)
|
|
svg1 = svg_for_channel(seg, 1, SEG_LABEL_SAMPLE)
|
|
ck.check("svg marks CLIP for header-clipped channel",
|
|
"CLIP" in svg1 and "UB" in svg1)
|
|
sentinel = os.path.join(td, "keep.svg")
|
|
with open(sentinel, "w", encoding="utf-8") as fh:
|
|
fh.write("SENTINEL")
|
|
bad = mutate(img, 200)
|
|
try:
|
|
parse_segment(bad, name="corrupt-for-plot")
|
|
ck.check("corrupt input must raise for plot path", False)
|
|
except RejectError:
|
|
ck.check("reject creates nothing",
|
|
not os.path.exists(os.path.join(td, "new.svg")))
|
|
with open(sentinel, "r", encoding="utf-8") as fh:
|
|
ck.check("reject leaves existing file untouched",
|
|
fh.read() == "SENTINEL")
|
|
|
|
# 7. negative integrity cases -- every one must REJECT with a reason
|
|
table_off = HDR_FIXED
|
|
hdr_crc_off = HDR_FIXED + 8 * ENT_SIZE
|
|
data_off = hdr_crc_off + 2
|
|
expect_reject(ck, "reject: bad magic", mutate(img, 0, 0xFF), "magic")
|
|
expect_reject(ck, "reject: bad magic version byte",
|
|
mutate(img, 2, 0x02), "version")
|
|
expect_reject(ck, "reject: bad ver field", mutate(img, OFF_VER, 0x02),
|
|
"ver field")
|
|
expect_reject(ck, "reject: header CRC mismatch (table byte flipped)",
|
|
mutate(img, table_off + 5), "header CRC")
|
|
expect_reject(ck, "reject: data CRC mismatch (sample byte flipped)",
|
|
mutate(img, data_off + 10), "data CRC")
|
|
expect_reject(ck, "reject: stored hdr CRC flipped",
|
|
mutate(img, hdr_crc_off), "header CRC")
|
|
expect_reject(ck, "reject: truncated tail", img[:-3], "size")
|
|
expect_reject(ck, "reject: one byte too long", img + b"\x00", "size")
|
|
expect_reject(ck, "reject: too short for a header", img[:10],
|
|
"too short")
|
|
zl = bytearray(img)
|
|
struct.pack_into("<H", zl, OFF_PTS, 0)
|
|
struct.pack_into("<H", zl, hdr_crc_off,
|
|
crc16_modbus(bytes(zl[:hdr_crc_off])))
|
|
expect_reject(ck, "reject: zero-length data even with re-stamped hdr CRC",
|
|
bytes(zl), "zero-length")
|
|
# section 6.3: v1 pins ch_count = 8 (TEMP would need a version bump)
|
|
c9 = bytearray(img)
|
|
c9[OFF_CHC] = 9
|
|
expect_reject(ck, "reject: ch_count=9 in v1", bytes(c9), "ch_count")
|
|
# section 6.3: PC must void the segment when table sem != index
|
|
ord_bad = bytearray(img)
|
|
ord_bad[HDR_FIXED + 1 * ENT_SIZE + ENT_SEM] = 6 # entry 1 claims sem 6
|
|
struct.pack_into("<H", ord_bad, hdr_crc_off,
|
|
crc16_modbus(bytes(ord_bad[:hdr_crc_off])))
|
|
expect_reject(ck, "reject: table sem != index (CRC re-stamped so the "
|
|
"order check is what bites)", bytes(ord_bad),
|
|
"semantic")
|
|
|
|
# 8. dump text renders every field (and decodes availability bits)
|
|
dump = "\n".join(fmt_dump(seg))
|
|
for token in ("frame_seq", "0x00E6", "UB UC IB IC IN", "MANUAL",
|
|
"pin-domain", "123456", "TPAFE5160", "STM32-ADC1",
|
|
"152.587891", "52800000", "0x02", "CLIPPED", "K_Q16"):
|
|
ck.check("dump contains %r" % token, token in dump)
|
|
|
|
print("SELFTEST %s: %d ok, %d fail"
|
|
% ("PASS" if ck.fail == 0 else "FAIL", ck.ok, ck.fail))
|
|
if ck.fail:
|
|
for p in ck.problems:
|
|
print(" failed: %s" % p)
|
|
return 0 if ck.fail == 0 else 1
|
|
|
|
|
|
# ---- main ----------------------------------------------------------------------
|
|
def build_parser():
|
|
p = argparse.ArgumentParser(
|
|
prog="wave_read.py",
|
|
description="PC-side wave-record segment reader "
|
|
"(dump/fft/shape/plot)")
|
|
p.add_argument("--selftest", action="store_true",
|
|
help="run built-in synthetic-segment self test and exit")
|
|
sub = p.add_subparsers(dest="cmd")
|
|
|
|
def add_common(sp):
|
|
sp.add_argument("file",
|
|
help="segment file, e.g. %s.BIN as fetched over FTP"
|
|
% SEG_LABEL_SAMPLE)
|
|
|
|
sp = sub.add_parser("dump", help="print all header fields + CRC verify")
|
|
add_common(sp)
|
|
sp.set_defaults(fn=cmd_dump)
|
|
|
|
sp = sub.add_parser("fft", help="full-cycle Fourier recompute (mV)")
|
|
add_common(sp)
|
|
sp.add_argument("--cycle", type=int, default=None,
|
|
help="cycle index (default: last full cycle)")
|
|
sp.add_argument("--all", action="store_true",
|
|
help="one row per full cycle + mean/min/max summary")
|
|
sp.add_argument("--ch", type=int, default=None,
|
|
help="only this table index (default: all)")
|
|
sp.set_defaults(fn=cmd_fft)
|
|
|
|
sp = sub.add_parser("shape", help="cycles/amplitude/polarity/rail pins")
|
|
add_common(sp)
|
|
sp.add_argument("--min-run", type=int, default=4,
|
|
help="samples at/inside a rail band before 'railed' "
|
|
"(default 4)")
|
|
sp.add_argument("--strict", action="store_true",
|
|
help="exit 1 when measured clip mask != header "
|
|
"clip_mask (yardstick: equality, the card "
|
|
"check)")
|
|
sp.add_argument("--clip-superset", action="store_true",
|
|
help="with --strict: use the header-covers-measured "
|
|
"yardstick (sealed-but-not-railed tolerated; "
|
|
"measured-but-not-sealed still fails) -- for "
|
|
"DC-only bias, tiny injection, or an un-driven "
|
|
"channel")
|
|
sp.set_defaults(fn=cmd_shape)
|
|
|
|
sp = sub.add_parser("plot", help="write SVG (refuses corrupt segments)")
|
|
add_common(sp)
|
|
sp.add_argument("--ch", type=int, default=None,
|
|
help="table index to draw (default: all channels)")
|
|
sp.add_argument("--outdir", default=None,
|
|
help="output directory (default: <repo>/ci-artifacts)")
|
|
sp.add_argument("--out", default=None,
|
|
help="exact output path (requires --ch)")
|
|
sp.set_defaults(fn=cmd_plot)
|
|
|
|
sp = sub.add_parser("selftest",
|
|
help="synthetic-segment round-trip test")
|
|
sp.set_defaults(fn=lambda _a: selftest())
|
|
return p
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) > 1 and argv[1] == "--selftest":
|
|
return selftest()
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv[1:])
|
|
if args.selftest:
|
|
return selftest()
|
|
if not getattr(args, "cmd", None):
|
|
parser.print_usage(sys.stderr)
|
|
return 2
|
|
try:
|
|
return args.fn(args)
|
|
except RejectError as exc:
|
|
print("REJECT: %s" % exc)
|
|
return 1
|
|
except IOErrorMessage as exc:
|
|
print("IO ERROR: %s" % exc)
|
|
return 2
|
|
except OSError as exc:
|
|
print("IO ERROR: %s" % exc)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|