Compare commits
3
Commits
dc74c97ade
...
2c5c20cfd7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2c5c20cfd7 | ||
|
|
ab9d84fde6 | ||
|
|
6341432524 |
@@ -0,0 +1,146 @@
|
|||||||
|
r"""Decode the Blastware sensor self-check waveforms from a series-3 event binary.
|
||||||
|
|
||||||
|
Reverse-engineered 2026-09-15 against 7 BE12844 (MiniMate Plus) oracle events.
|
||||||
|
After the main waveform record-chain and the trailing metadata / per-channel
|
||||||
|
calibration records, the binary carries four length-prefixed records tagged
|
||||||
|
0x3c-0x3f: the sensor self-check traces the unit records when it pulses each
|
||||||
|
sensor before monitoring. Blastware draws these as the little waveforms in the
|
||||||
|
"Sensor Check" strip on the right of the Event Report.
|
||||||
|
|
||||||
|
* 0x3c / 0x3d / 0x3e = Tran / Vert / Long geophone ring-downs (a damped
|
||||||
|
oscillation at the geophone's resonance, ~7-8 Hz at 1024 sps).
|
||||||
|
* 0x3f = MicL, a pulse train at the mic self-test frequency
|
||||||
|
(~20 Hz), whose zero-crossing frequency is BW's mic "Channel Test" freq.
|
||||||
|
|
||||||
|
Record framing (per record, all four chained by their length prefix)::
|
||||||
|
|
||||||
|
[len:2 BE][id:1][00 00][Nchan:1][12-byte header][delta stream][40 02][6B]
|
||||||
|
\_________________ payload (len bytes) _______________________________/
|
||||||
|
|
||||||
|
The delta stream is ``payload[20 : len-8]`` (the ``40 02`` terminator sits at
|
||||||
|
``len-8``, followed by 6 trailing bytes). It uses the exact same 10/20/30/00
|
||||||
|
delta-block tags as the main waveform codec
|
||||||
|
(:mod:`minimateplus.waveform_codec`), decoded here from an implicit anchor of 0
|
||||||
|
— so the traces come out in the same 16-count raw units as the main waveform
|
||||||
|
(LSB = 0.005 in/s at Normal range for the geophones).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Dict, List
|
||||||
|
|
||||||
|
from minimateplus.waveform_codec import walk_body
|
||||||
|
|
||||||
|
# Record id → channel. Order mirrors the trailing per-channel calibration
|
||||||
|
# records (Tran / Vert / Long / MicL), confirmed against BW's sensor-check
|
||||||
|
# frequencies on all 7 oracle events.
|
||||||
|
_ID_TO_CHANNEL = {0x3C: "Tran", 0x3D: "Vert", 0x3E: "Long", 0x3F: "MicL"}
|
||||||
|
_CHAIN_IDS = (0x3C, 0x3D, 0x3E, 0x3F)
|
||||||
|
|
||||||
|
_HEADER_LEN = 20 # payload bytes before the delta stream
|
||||||
|
_TRAILER_LEN = 8 # 40 02 terminator + 6 trailing bytes after the stream
|
||||||
|
|
||||||
|
|
||||||
|
def _s4(nib: int) -> int:
|
||||||
|
"""Sign-extend a 4-bit nibble delta."""
|
||||||
|
return nib - 16 if nib >= 8 else nib
|
||||||
|
|
||||||
|
|
||||||
|
def _i8(byte: int) -> int:
|
||||||
|
"""Sign-extend an 8-bit int delta."""
|
||||||
|
return byte - 256 if byte >= 128 else byte
|
||||||
|
|
||||||
|
|
||||||
|
def _decode_delta_stream(buf: bytes) -> List[int]:
|
||||||
|
"""Accumulate a 10/20/30/00 delta-block stream from an anchor of 0,
|
||||||
|
stopping at the 0x40 terminator.
|
||||||
|
|
||||||
|
Mirrors the block semantics in
|
||||||
|
:func:`minimateplus.waveform_codec.decode_waveform_v2` (fully decoded &
|
||||||
|
byte-exact as of 2026-05-11); see that module for the format details.
|
||||||
|
"""
|
||||||
|
out: List[int] = []
|
||||||
|
cur = 0
|
||||||
|
for blk in walk_body(buf, 0):
|
||||||
|
fam = blk.tag_hi & 0xF0
|
||||||
|
if fam == 0x10:
|
||||||
|
# nibble deltas, high nibble first
|
||||||
|
for byte in blk.data:
|
||||||
|
for nib in ((byte >> 4) & 0xF, byte & 0xF):
|
||||||
|
cur += _s4(nib)
|
||||||
|
out.append(cur)
|
||||||
|
elif fam == 0x20:
|
||||||
|
# int8 deltas
|
||||||
|
for byte in blk.data:
|
||||||
|
cur += _i8(byte)
|
||||||
|
out.append(cur)
|
||||||
|
elif fam == 0x30:
|
||||||
|
# 12-bit signed deltas, packed as tag_lo/4 groups of 6 bytes
|
||||||
|
for g in range(blk.tag_lo // 4):
|
||||||
|
grp = blk.data[g * 6:(g + 1) * 6]
|
||||||
|
if len(grp) < 6:
|
||||||
|
break
|
||||||
|
high_word = (grp[0] << 8) | grp[1]
|
||||||
|
for k in range(4):
|
||||||
|
nib = (high_word >> (12 - 4 * k)) & 0xF
|
||||||
|
v = (nib << 8) | grp[2 + k]
|
||||||
|
if v >= 0x800:
|
||||||
|
v -= 0x1000
|
||||||
|
cur += v
|
||||||
|
out.append(cur)
|
||||||
|
elif fam == 0x00:
|
||||||
|
# RLE zero-delta run (wide form carries the high nibble in the tag)
|
||||||
|
run = ((blk.tag_hi & 0x0F) << 8) | blk.tag_lo
|
||||||
|
out.extend([cur] * run)
|
||||||
|
elif fam == 0x40:
|
||||||
|
# segment / record terminator
|
||||||
|
break
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def _find_chain(body: bytes):
|
||||||
|
"""Locate the four length-prefixed sensor-check records.
|
||||||
|
|
||||||
|
Returns a list of ``(offset, id, length)`` or ``None``. The chain is
|
||||||
|
validated by walking the ids 0x3c → 0x3d → 0x3e → 0x3f via their own length
|
||||||
|
prefixes, so a stray 0x3c byte in the waveform data cannot match.
|
||||||
|
"""
|
||||||
|
for p in range(len(body) - 6):
|
||||||
|
if body[p + 2] == 0x3C and body[p + 3] == 0 and body[p + 4] == 0:
|
||||||
|
q = p
|
||||||
|
recs = []
|
||||||
|
ok = True
|
||||||
|
for expect in _CHAIN_IDS:
|
||||||
|
if q + 3 > len(body) or body[q + 2] != expect:
|
||||||
|
ok = False
|
||||||
|
break
|
||||||
|
length = int.from_bytes(body[q:q + 2], "big")
|
||||||
|
recs.append((q, expect, length))
|
||||||
|
q = q + 2 + length
|
||||||
|
if ok and len(recs) == 4:
|
||||||
|
return recs
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def decode_sensor_check(raw: bytes) -> Dict[str, List[int]]:
|
||||||
|
"""Decode the four sensor self-check traces from a series-3 event binary.
|
||||||
|
|
||||||
|
Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}`` in
|
||||||
|
raw decode units (same 16-count LSB as the main waveform), or ``{}`` if the
|
||||||
|
binary carries no sensor-check block (a histogram event, a non-series-3
|
||||||
|
file, or a unit/firmware that doesn't store it).
|
||||||
|
"""
|
||||||
|
strt = raw.find(b"STRT")
|
||||||
|
if strt < 0 or len(raw) < strt + 21 + 26:
|
||||||
|
return {}
|
||||||
|
body = raw[strt + 21: len(raw) - 26]
|
||||||
|
chain = _find_chain(body)
|
||||||
|
if not chain:
|
||||||
|
return {}
|
||||||
|
out: Dict[str, List[int]] = {}
|
||||||
|
for off, rid, length in chain:
|
||||||
|
payload = body[off + 2: off + 2 + length]
|
||||||
|
if len(payload) < _HEADER_LEN + _TRAILER_LEN:
|
||||||
|
continue
|
||||||
|
stream = payload[_HEADER_LEN: length - _TRAILER_LEN]
|
||||||
|
out[_ID_TO_CHANNEL[rid]] = _decode_delta_stream(stream)
|
||||||
|
return out
|
||||||
+82
-6
@@ -121,6 +121,11 @@ class ReportData:
|
|||||||
t0_ms: Optional[float] = None
|
t0_ms: Optional[float] = None
|
||||||
dt_ms: Optional[float] = None
|
dt_ms: Optional[float] = None
|
||||||
|
|
||||||
|
# Sensor self-check traces — {ch: [samples]} in raw decode units, decoded
|
||||||
|
# from the binary's trailing block (see minimateplus.sensor_check). The
|
||||||
|
# little waveforms BW draws in its "Sensor Check" strip. Empty when absent.
|
||||||
|
sensor_check_waveforms: dict = field(default_factory=dict)
|
||||||
|
|
||||||
# Record-type discriminator
|
# Record-type discriminator
|
||||||
record_type: Optional[str] = None
|
record_type: Optional[str] = None
|
||||||
is_histogram: bool = False
|
is_histogram: bool = False
|
||||||
@@ -246,6 +251,8 @@ def gather_report_data(
|
|||||||
"peak_accel_g": ch.get("peak_accel_g"),
|
"peak_accel_g": ch.get("peak_accel_g"),
|
||||||
"peak_disp_in": ch.get("peak_disp_in"),
|
"peak_disp_in": ch.get("peak_disp_in"),
|
||||||
"sensor_check": sc_ch.get("result"),
|
"sensor_check": sc_ch.get("result"),
|
||||||
|
"sc_freq_hz": sc_ch.get("freq_hz"),
|
||||||
|
"sc_ratio": sc_ch.get("ratio"),
|
||||||
"peak_date": peak_date,
|
"peak_date": peak_date,
|
||||||
"peak_time": peak_time,
|
"peak_time": peak_time,
|
||||||
})
|
})
|
||||||
@@ -290,6 +297,19 @@ def gather_report_data(
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
log.warning("gather_report_data: hdf5 read failed: %s", exc)
|
log.warning("gather_report_data: hdf5 read failed: %s", exc)
|
||||||
|
|
||||||
|
# ── Sensor self-check traces — decoded from the retained raw binary ──
|
||||||
|
# The .h5 holds only the main waveform; the sensor-check traces live in the
|
||||||
|
# binary's trailing block, so decode them straight from the kept BW file.
|
||||||
|
# Waveform events only (histograms have no sensor-check strip).
|
||||||
|
if not rd.is_histogram:
|
||||||
|
try:
|
||||||
|
from minimateplus.sensor_check import decode_sensor_check
|
||||||
|
bw_path, _a5 = store.paths_for(serial, filename)
|
||||||
|
if bw_path.exists():
|
||||||
|
rd.sensor_check_waveforms = decode_sensor_check(bw_path.read_bytes())
|
||||||
|
except Exception as exc:
|
||||||
|
log.warning("gather_report_data: sensor-check decode failed: %s", exc)
|
||||||
|
|
||||||
# ── Histogram aggregation ──
|
# ── Histogram aggregation ──
|
||||||
# Codec emits ~N per-block samples (typically 1/sec); BW reports
|
# Codec emits ~N per-block samples (typically 1/sec); BW reports
|
||||||
# one bar per configured interval (1 min / 5 min / etc.). When
|
# one bar per configured interval (1 min / 5 min / etc.). When
|
||||||
@@ -569,14 +589,17 @@ def _draw_header_columns(ax, rows_left, rd: ReportData) -> None:
|
|||||||
("File Name", rd.file_name),
|
("File Name", rd.file_name),
|
||||||
("Post Event Notes", rd.post_event_notes),
|
("Post Event Notes", rd.post_event_notes),
|
||||||
]
|
]
|
||||||
|
# fontsize 7.5 (BW's header is a touch smaller than our body text) + a
|
||||||
|
# tighter right-column value indent so the long serial+firmware line
|
||||||
|
# ("BE##### V ##.##-#.## MiniMate Plus") fits without running off the page.
|
||||||
y = 0.95
|
y = 0.95
|
||||||
dy = 0.095
|
dy = 0.095
|
||||||
for label, value in rows_left:
|
for label, value in rows_left:
|
||||||
_kv(ax, 0.0, y, label, value, label_w=0.18)
|
_kv(ax, 0.0, y, label, value, label_w=0.18, fontsize=7.5)
|
||||||
y -= dy
|
y -= dy
|
||||||
y = 0.95
|
y = 0.95
|
||||||
for label, value in rows_right:
|
for label, value in rows_right:
|
||||||
_kv(ax, 0.55, y, label, value, label_w=0.20)
|
_kv(ax, 0.55, y, label, value, label_w=0.14, fontsize=7.5)
|
||||||
y -= dy
|
y -= dy
|
||||||
|
|
||||||
|
|
||||||
@@ -656,6 +679,10 @@ def _draw_channel_stats_waveform(ax, rd: ReportData) -> None:
|
|||||||
("Peak Acceleration", "peak_accel_g", "g"),
|
("Peak Acceleration", "peak_accel_g", "g"),
|
||||||
("Peak Displacement", "peak_disp_in", "in"),
|
("Peak Displacement", "peak_disp_in", "in"),
|
||||||
("Sensor Check", "sensor_check", ""),
|
("Sensor Check", "sensor_check", ""),
|
||||||
|
# Sensor-check sub-rows (indented under "Sensor Check", like BW): the
|
||||||
|
# geophone ring-down frequency + overswing ratio from the self-check.
|
||||||
|
(" Frequency", "sc_freq_hz", "Hz"),
|
||||||
|
(" Overswing Ratio", "sc_ratio", ""),
|
||||||
]
|
]
|
||||||
# Compacted to the left half so the enlarged compliance chart (BW-sized,
|
# Compacted to the left half so the enlarged compliance chart (BW-sized,
|
||||||
# right against the page margin) has room — see _COMPLIANCE_BOX.
|
# right against the page margin) has room — see _COMPLIANCE_BOX.
|
||||||
@@ -772,6 +799,8 @@ def _draw_stats_table(
|
|||||||
if field == "zc_freq_hz":
|
if field == "zc_freq_hz":
|
||||||
prefix = ">" if ch_rec.get("zc_freq_above_range") else ""
|
prefix = ">" if ch_rec.get("zc_freq_above_range") else ""
|
||||||
return f"{prefix}{val:.0f}"
|
return f"{prefix}{val:.0f}"
|
||||||
|
if field in ("sc_freq_hz", "sc_ratio"):
|
||||||
|
return f"{val:.1f}" # BW shows 1 decimal (7.5 Hz, 3.6)
|
||||||
return f"{val:.3f}"
|
return f"{val:.3f}"
|
||||||
return str(val)
|
return str(val)
|
||||||
|
|
||||||
@@ -816,9 +845,22 @@ def _channel_axis_color(ch: str) -> str:
|
|||||||
def _draw_waveform_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
def _draw_waveform_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
||||||
"""4-channel stacked waveform plot — Instantel printout order
|
"""4-channel stacked waveform plot — Instantel printout order
|
||||||
(MicL on top, Tran on bottom), shared x-axis in SECONDS, trigger
|
(MicL on top, Tran on bottom), shared x-axis in SECONDS, trigger
|
||||||
triangle markers at t=0, '0.0' baseline label on right of each."""
|
triangle markers at t=0, '0.0' baseline label on right of each.
|
||||||
inner = gridspec_cell.subgridspec(4, 1, hspace=0.0)
|
|
||||||
|
When sensor self-check traces are present (rd.sensor_check_waveforms), a
|
||||||
|
narrow "Sensor Check" strip of per-channel mini-plots is drawn to the right,
|
||||||
|
aligned to the lanes — matching Blastware's Event Report.
|
||||||
|
"""
|
||||||
|
from matplotlib.ticker import MaxNLocator
|
||||||
|
|
||||||
order = ["MicL", "Long", "Vert", "Tran"]
|
order = ["MicL", "Long", "Vert", "Tran"]
|
||||||
|
has_sc = bool(rd.sensor_check_waveforms)
|
||||||
|
if has_sc:
|
||||||
|
# main lanes + a narrow sensor-check strip column on the right
|
||||||
|
inner = gridspec_cell.subgridspec(4, 2, width_ratios=[1.0, 0.15],
|
||||||
|
wspace=0.04, hspace=0.0)
|
||||||
|
else:
|
||||||
|
inner = gridspec_cell.subgridspec(4, 1, hspace=0.0)
|
||||||
sr = rd.sample_rate_sps or 1024
|
sr = rd.sample_rate_sps or 1024
|
||||||
# Convert ms-based time axis to seconds for the x-axis
|
# Convert ms-based time axis to seconds for the x-axis
|
||||||
dt_s = (rd.dt_ms or (1000.0 / sr)) / 1000.0
|
dt_s = (rd.dt_ms or (1000.0 / sr)) / 1000.0
|
||||||
@@ -837,9 +879,12 @@ def _draw_waveform_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
|||||||
_geo_amax = _a
|
_geo_amax = _a
|
||||||
geo_shared = max(_geo_amax * 1.10, GEO_FLOOR_INS)
|
geo_shared = max(_geo_amax * 1.10, GEO_FLOOR_INS)
|
||||||
|
|
||||||
|
main_axes = []
|
||||||
|
sc_axes = []
|
||||||
last_idx = len(order) - 1
|
last_idx = len(order) - 1
|
||||||
for i, ch in enumerate(order):
|
for i, ch in enumerate(order):
|
||||||
ax = fig.add_subplot(inner[i])
|
ax = fig.add_subplot(inner[i, 0] if has_sc else inner[i])
|
||||||
|
main_axes.append(ax)
|
||||||
values = rd.channels.get(ch) or []
|
values = rd.channels.get(ch) or []
|
||||||
times = [t0_s + j * dt_s for j in range(len(values))]
|
times = [t0_s + j * dt_s for j in range(len(values))]
|
||||||
|
|
||||||
@@ -874,12 +919,43 @@ def _draw_waveform_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
|||||||
else:
|
else:
|
||||||
ax.tick_params(axis="x", labelsize=7)
|
ax.tick_params(axis="x", labelsize=7)
|
||||||
ax.tick_params(axis="y", labelsize=6)
|
ax.tick_params(axis="y", labelsize=6)
|
||||||
|
# Stacked lanes touch, so the top/bottom y-tick labels of adjacent lanes
|
||||||
|
# would overprint at the shared boundary. Prune the extreme ticks so
|
||||||
|
# each boundary shows clean interior ticks (0.5 / 0.0 / -0.5) only.
|
||||||
|
ax.yaxis.set_major_locator(MaxNLocator(nbins=4, prune="both"))
|
||||||
|
|
||||||
|
# Sensor self-check mini-plot in the right strip (aligned to this lane).
|
||||||
|
if has_sc:
|
||||||
|
scx = fig.add_subplot(inner[i, 1])
|
||||||
|
sc_axes.append(scx)
|
||||||
|
sc_vals = rd.sensor_check_waveforms.get(ch) or []
|
||||||
|
if sc_vals:
|
||||||
|
_col = _channel_axis_color(ch)
|
||||||
|
# Faint zero baseline (BW draws the channel baseline through the
|
||||||
|
# strip) — reference for the one-sided geophone ring-downs.
|
||||||
|
scx.axhline(0.0, color=_col, linewidth=0.3, alpha=0.4)
|
||||||
|
scx.plot(range(len(sc_vals)), sc_vals, color=_col, linewidth=0.5)
|
||||||
|
# Fit the trace to the box (BW-style) rather than a symmetric
|
||||||
|
# scale: the geo self-checks are one-sided dips, so a symmetric
|
||||||
|
# scale would strand them in the bottom half with an empty top.
|
||||||
|
_lo, _hi = min(sc_vals), max(sc_vals)
|
||||||
|
_pad = 0.10 * ((_hi - _lo) or 1.0)
|
||||||
|
scx.set_ylim(_lo - _pad, _hi + _pad)
|
||||||
|
scx.set_xticks([]); scx.set_yticks([])
|
||||||
|
for _s in scx.spines.values():
|
||||||
|
_s.set_linewidth(0.4); _s.set_color("#999")
|
||||||
|
|
||||||
# Trigger triangle marker ▼ above the top channel at t=0
|
# Trigger triangle marker ▼ above the top channel at t=0
|
||||||
top_ax = fig.axes[-4] # MicL is the first added in this gridspec
|
top_ax = main_axes[0] # MicL
|
||||||
top_ax.plot([0], [top_ax.get_ylim()[1]], marker="v", color="black",
|
top_ax.plot([0], [top_ax.get_ylim()[1]], marker="v", color="black",
|
||||||
markersize=8, clip_on=False, zorder=10)
|
markersize=8, clip_on=False, zorder=10)
|
||||||
|
|
||||||
|
# "Sensor Check" caption under the strip (BW convention)
|
||||||
|
if has_sc and sc_axes:
|
||||||
|
pos = sc_axes[-1].get_position()
|
||||||
|
fig.text((pos.x0 + pos.x1) / 2, pos.y0 - 0.012, "Sensor Check",
|
||||||
|
fontsize=7, color="#555", ha="center", va="top")
|
||||||
|
|
||||||
# Compute scale-per-division for the footer (10 divs across the chart)
|
# Compute scale-per-division for the footer (10 divs across the chart)
|
||||||
# and find peak geo amplitude for the geo amp/div setting.
|
# and find peak geo amplitude for the geo amp/div setting.
|
||||||
total_s = times[-1] - times[0] if values else 0
|
total_s = times[-1] - times[0] if values else 0
|
||||||
|
|||||||
@@ -0,0 +1,67 @@
|
|||||||
|
"""Blastware sensor self-check waveform decode (minimateplus.sensor_check).
|
||||||
|
|
||||||
|
Reverse-engineered 2026-09-15 against 7 BE12844 (MiniMate Plus) oracle events.
|
||||||
|
After the main waveform record-chain and the trailing metadata / per-channel
|
||||||
|
calibration records, a series-3 binary carries four length-prefixed records
|
||||||
|
tagged 0x3c-0x3f: the sensor self-check traces the unit records when it pulses
|
||||||
|
each sensor before monitoring (Blastware draws these as the little waveforms in
|
||||||
|
the "Sensor Check" strip on the right of the Event Report).
|
||||||
|
|
||||||
|
* 0x3c / 0x3d / 0x3e = Tran / Vert / Long geophone ring-downs.
|
||||||
|
* 0x3f = MicL, a pulse train at the mic self-test frequency.
|
||||||
|
|
||||||
|
The self-check injects a fixed pulse, so the response is near-identical across
|
||||||
|
events — asserted here as an invariant shape (damped one-sided ring-down for
|
||||||
|
the geophones, a multi-pulse train for the mic).
|
||||||
|
"""
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from minimateplus.sensor_check import decode_sensor_check
|
||||||
|
|
||||||
|
FIXDIR = Path(__file__).parent / "fixtures" / "fft-oracle-2026-09-14"
|
||||||
|
EVENTS = sorted(p.name for p in FIXDIR.iterdir()) # 7 BE12844 event binaries
|
||||||
|
|
||||||
|
|
||||||
|
def _decode(name):
|
||||||
|
return decode_sensor_check((FIXDIR / name).read_bytes())
|
||||||
|
|
||||||
|
|
||||||
|
def test_all_four_channels_present():
|
||||||
|
for name in EVENTS:
|
||||||
|
sc = _decode(name)
|
||||||
|
assert set(sc) == {"Tran", "Vert", "Long", "MicL"}, name
|
||||||
|
|
||||||
|
|
||||||
|
def test_geo_channels_are_damped_ringdowns():
|
||||||
|
# Each geophone self-check is a large one-sided deflection (~-990 raw) that
|
||||||
|
# rings back and damps toward a settled value well above the trough.
|
||||||
|
for name in EVENTS:
|
||||||
|
sc = _decode(name)
|
||||||
|
for ch in ("Tran", "Vert", "Long"):
|
||||||
|
tr = np.asarray(sc[ch], dtype=float)
|
||||||
|
assert 240 <= len(tr) <= 260, f"{name}:{ch} n={len(tr)}"
|
||||||
|
assert abs(tr[:3].mean()) < 50, f"{name}:{ch} starts off-baseline"
|
||||||
|
assert tr.min() < -800, f"{name}:{ch} min {tr.min()}"
|
||||||
|
assert tr.max() < 60, f"{name}:{ch} unexpected positive swing {tr.max()}"
|
||||||
|
# damped: settles between the trough and zero, well above the trough
|
||||||
|
assert tr.min() < tr[-1] < 0, f"{name}:{ch} end {tr[-1]} not between trough and 0"
|
||||||
|
assert abs(tr[-1]) < 0.6 * abs(tr.min()), f"{name}:{ch} not damped, end {tr[-1]}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_mic_channel_is_a_pulse_train():
|
||||||
|
for name in EVENTS:
|
||||||
|
tr = np.asarray(_decode(name)["MicL"], dtype=float)
|
||||||
|
assert 235 <= len(tr) <= 255, f"{name} mic n={len(tr)}"
|
||||||
|
# larger dynamic range than the geo ring-down, and swings both ways
|
||||||
|
assert tr.min() < -1500, f"{name} mic min {tr.min()}"
|
||||||
|
assert tr.max() > 100, f"{name} mic max {tr.max()}"
|
||||||
|
# multiple pulses: several deep local minima
|
||||||
|
deep = (tr[1:-1] < tr[:-2]) & (tr[1:-1] < tr[2:]) & (tr[1:-1] < -800)
|
||||||
|
assert int(deep.sum()) >= 4, f"{name} mic pulses {int(deep.sum())}"
|
||||||
|
|
||||||
|
|
||||||
|
def test_returns_empty_when_no_sensor_check_block():
|
||||||
|
assert decode_sensor_check(b"not a blastware file") == {}
|
||||||
|
assert decode_sensor_check(b"") == {}
|
||||||
Reference in New Issue
Block a user