chore(release): v0.24.0 — waveform-shape metrics on events

7-task TDD plan: shape DSP module (crest factor + points-near-peak), shape_*
columns + auto-migrate, insert_events persistence, ingest population in the
save paths, backfill script, /db/events exposure + v0.24.0 bump. Phase B
(terra-view scoring/UI/review) gets its own plan once this feed is live.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf

CHANGELOG.md

@@ -8,6 +8,37 @@ All notable changes to seismo-relay are documented here.
 
 ---
 
+## v0.24.0 — 2026-08-22
+
+**Waveform-shape metrics on events.** The `events` table and the `/db/events`
+feed now carry per-event crest factor and points-near-peak, computed from the
+decoded waveform samples at ingest — groundwork for Terra-View's
+false-trigger detection (Phase B).
+
+### Added
+
+- **`events` columns** `shape_crest_factor`, `shape_near_peak_count`,
+  `shape_sample_count`, `shape_axis`. Added via the existing incremental
+  `_migrate` ADD COLUMN pass — **auto-migrates on `SeismoDb()` construction,
+  no manual migration**. `query_events` / `get_event` (and thus `/db/events`)
+  return them automatically (`SELECT *`).
+- **Populated at ingest** — crest factor + near-peak-count are computed from
+  the decoded samples in every save path (`shape_from_h5`/
+  `shape_from_samples`), and `insert_events` persists them on INSERT and
+  UPSERT.
+- **Backfill** `scripts/backfill_event_shape.py` — fills the columns for
+  existing events from their on-disk `.h5` waveform samples (idempotent,
+  UPDATE-only).
+
+### Upgrade Notes
+
+Run the backfill once after deploying, against the events DB + waveform store:
+`python3 scripts/backfill_event_shape.py --db-path <seismo_relay.db> --store-root <waveforms/>`
+Events with no decodable samples (or no waveform file) stay NULL and render
+"—" downstream.
+
+---
+
 ## v0.23.0 — 2026-08-04
 
 **Per-channel ZC frequency in the events store.** The `events` table and the

docs/superpowers/plans/2026-08-21-waveform-ft-detection-phase-a.md

@@ -0,0 +1,628 @@
+# Waveform-shape FT detection — Phase A (seismo-relay) Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Compute per-event waveform-shape metrics (crest factor + points-near-peak) in SFM and store them as `events` columns, populated at ingest and by a backfill script, so Terra-View can read them from `/db/events`.
+
+**Architecture:** A pure DSP module (`sfm/shape_metrics.py`) turns decoded `.h5` samples into shape metrics. New nullable `events.shape_*` columns are added via the existing `_migrate` ADD COLUMN loop. The three `WaveformStore.save*` paths compute shape from the just-written `.h5` and hand it to `insert_events`; a backfill script does the same over existing events. Mirrors exactly how per-channel ZC frequency was added.
+
+**Tech Stack:** Python 3.10, numpy, h5py, sqlite3 (raw), pytest. seismo-relay venv: `/home/serversdown/seismo-relay/.venv/bin/python3`.
+
+## Global Constraints
+
+- Metrics are read from the `.h5` `samples/{Tran,Vert,Long}` float32 arrays (physical in/s). The measured channel is the max-|peak| geophone channel.
+- All new columns are nullable; histogram records and events without usable samples store NULL (Terra-View falls back to cheap signals). Legacy rows stay valid.
+- Crest factor = `max(|x|) / rms(x)`; near-peak count = number of samples with `|x| ≥ 0.5·peak`. Threshold `0.5` is a module constant so calibration can tune it.
+- No manual migration: columns are added in `SeismoDb._migrate`, run at `SeismoDb()` construction.
+- `/db/events` needs no change — it returns all columns via `SELECT *` (verify with a test).
+- Run tests with `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest`.
+
+---
+
+### Task 1: Shape DSP module
+
+**Files:**
+- Create: `sfm/shape_metrics.py`
+- Test: `tests/test_shape_metrics.py`
+
+**Interfaces:**
+- Produces:
+  - `channel_shape(x) -> dict | None` — `{"crest_factor": float, "near_peak_count": int, "sample_count": int}` or None for unusable input (size < 2, flat, all-zero).
+  - `shape_from_samples(chans: dict[str, ArrayLike]) -> dict | None` — picks the max-peak geophone channel; returns `{"crest_factor","near_peak_count","sample_count","axis"}` or None.
+  - `shape_from_h5(path) -> dict | None` — reads `samples/{Tran,Vert,Long}` and delegates to `shape_from_samples`; None on any read error.
+  - Constant `NEAR_PEAK_FRACTION = 0.5`.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_shape_metrics.py
+import numpy as np
+from sfm.shape_metrics import channel_shape, shape_from_samples
+
+def test_needle_spike_high_crest_few_near_peak():
+    x = np.zeros(1024); x[500] = 1.0          # one isolated spike
+    s = channel_shape(x)
+    assert s["sample_count"] == 1024
+    assert s["crest_factor"] > 15             # peak towers over rms
+    assert s["near_peak_count"] <= 3          # almost nothing near the peak
+
+def test_ringing_low_crest_many_near_peak():
+    t = np.arange(1024)
+    x = np.sin(2*np.pi*t/32) * np.exp(-t/4000)  # decaying oscillation
+    s = channel_shape(x)
+    assert s["crest_factor"] < 6
+    assert s["near_peak_count"] > 30          # many samples near the peak
+
+def test_channel_shape_none_for_unusable():
+    assert channel_shape(np.zeros(1024)) is None   # flat / all-zero
+    assert channel_shape(np.array([1.0])) is None  # too short
+
+def test_shape_from_samples_picks_max_peak_axis():
+    chans = {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": np.zeros(1024)}
+    chans["Long"][10] = 0.5
+    chans["Vert"] = np.sin(np.arange(1024)/5) * 0.01
+    s = shape_from_samples(chans)
+    assert s["axis"] == "Long"                # Long has the biggest peak
+    assert s["near_peak_count"] <= 3
+
+def test_shape_from_samples_none_when_no_geo():
+    assert shape_from_samples({"MicL": np.ones(1024)}) is None
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics.py -q`
+Expected: FAIL — `ModuleNotFoundError: sfm.shape_metrics`.
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+# sfm/shape_metrics.py
+"""Waveform-shape metrics for false-trigger detection.
+
+A false trigger is an isolated impulse (quiet → spike → quiet); a real event
+rings for many cycles. Two numbers separate them: crest factor (how far the
+peak stands above the typical sample) and how many samples sit near the peak.
+"""
+from __future__ import annotations
+import numpy as np
+
+_GEO_CHANNELS = ("Tran", "Vert", "Long")
+NEAR_PEAK_FRACTION = 0.5   # a sample "near the peak" is >= this * peak amplitude
+
+
+def channel_shape(x) -> dict | None:
+    x = np.asarray(x, dtype=float)
+    if x.size < 2:
+        return None
+    peak = float(np.max(np.abs(x)))
+    if peak <= 0:
+        return None
+    rms = float(np.sqrt(np.mean(x ** 2)))
+    if rms <= 0:
+        return None
+    near = int(np.sum(np.abs(x) >= NEAR_PEAK_FRACTION * peak))
+    return {"crest_factor": peak / rms, "near_peak_count": near,
+            "sample_count": int(x.size)}
+
+
+def shape_from_samples(chans: dict) -> dict | None:
+    best_axis, best_peak, best_x = None, -1.0, None
+    for ax in _GEO_CHANNELS:
+        x = chans.get(ax)
+        if x is None:
+            continue
+        x = np.asarray(x, dtype=float)
+        if x.size < 2:
+            continue
+        p = float(np.max(np.abs(x)))
+        if p > best_peak:
+            best_axis, best_peak, best_x = ax, p, x
+    if best_axis is None:
+        return None
+    s = channel_shape(best_x)
+    if s is None:
+        return None
+    s["axis"] = best_axis
+    return s
+
+
+def shape_from_h5(path) -> dict | None:
+    import h5py
+    try:
+        with h5py.File(path, "r") as f:
+            chans = {ax: f[f"samples/{ax}"][:] for ax in _GEO_CHANNELS
+                     if f"samples/{ax}" in f}
+    except Exception:
+        return None
+    return shape_from_samples(chans)
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics.py -q`
+Expected: PASS (5 tests).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/shape_metrics.py tests/test_shape_metrics.py
+git commit -m "feat(shape): crest-factor + points-near-peak waveform metrics"
+```
+
+---
+
+### Task 2: shape_from_h5 round-trips a real .h5
+
+**Files:**
+- Test: `tests/test_shape_metrics_h5.py`
+
+**Interfaces:**
+- Consumes: `sfm.shape_metrics.shape_from_h5`; `h5py`.
+
+- [ ] **Step 1: Write the failing test** (writes a tiny .h5 the same shape SFM writes, then reads it back)
+
+```python
+# tests/test_shape_metrics_h5.py
+import numpy as np, h5py
+from sfm.shape_metrics import shape_from_h5
+
+def _write_h5(path, chans):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k, v in chans.items():
+            g.create_dataset(k, data=np.asarray(v, dtype="float32"))
+
+def test_shape_from_h5_reads_dominant_axis(tmp_path):
+    p = tmp_path / "ev.h5"
+    long = np.zeros(1024, dtype="float32"); long[100] = 0.48
+    _write_h5(p, {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": long,
+                  "MicL": np.ones(1024)})
+    s = shape_from_h5(str(p))
+    assert s["axis"] == "Long" and s["near_peak_count"] <= 3
+
+def test_shape_from_h5_none_on_missing_or_degenerate(tmp_path):
+    assert shape_from_h5(str(tmp_path / "nope.h5")) is None
+    p = tmp_path / "degen.h5"
+    _write_h5(p, {"Tran": np.zeros(1), "Vert": np.zeros(1), "Long": np.zeros(1)})
+    assert shape_from_h5(str(p)) is None
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics_h5.py -q`
+Expected: FAIL (assertion or, if Task 1 incomplete, import error).
+
+- [ ] **Step 3: Implementation** — none needed; `shape_from_h5` already exists from Task 1. If a test fails, fix `shape_from_h5` (not the test).
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics_h5.py -q`
+Expected: PASS (2 tests).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add tests/test_shape_metrics_h5.py
+git commit -m "test(shape): shape_from_h5 round-trips a real .h5"
+```
+
+---
+
+### Task 3: Add shape columns to the events schema + migration
+
+**Files:**
+- Modify: `sfm/database.py` — `_SCHEMA` CREATE TABLE `events` (after `mic_zc_above_range`); `_migrate` ADD COLUMN loop (the tuple around line 205-218).
+- Test: `tests/test_shape_columns.py`
+
+**Interfaces:**
+- Produces: `events` columns `shape_crest_factor REAL`, `shape_near_peak_count INTEGER`, `shape_sample_count INTEGER`, `shape_axis TEXT`.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_shape_columns.py
+import sqlite3
+from sfm.database import SeismoDb
+
+_SHAPE_COLS = {"shape_crest_factor", "shape_near_peak_count",
+               "shape_sample_count", "shape_axis"}
+
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+
+def test_fresh_db_has_shape_columns(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    assert _SHAPE_COLS <= _cols(db)
+
+def test_existing_db_migrates_shape_columns(tmp_path):
+    p = tmp_path / "s.db"
+    db = SeismoDb(p)
+    with sqlite3.connect(p) as c:      # simulate an older DB missing the columns
+        for col in _SHAPE_COLS:
+            c.execute(f"ALTER TABLE events DROP COLUMN {col}")
+    assert not (_SHAPE_COLS <= _cols(SeismoDb(p)))  # sanity: dropped
+    SeismoDb(p)                        # re-open triggers _migrate
+    assert _SHAPE_COLS <= _cols(SeismoDb(p))
+```
+
+> Note: sqlite `DROP COLUMN` needs sqlite ≥ 3.35 (bundled py3.10 has it). If the runner's sqlite lacks it, replace the "simulate older DB" block with building a table without the columns; keep the assertion that re-open adds them.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_columns.py -q`
+Expected: FAIL — columns absent.
+
+- [ ] **Step 3: Implementation**
+
+In `_SCHEMA`, after the `mic_zc_above_range      INTEGER,` line in the `events` CREATE TABLE, add:
+
+```
+    shape_crest_factor      REAL,                   -- peak / rms of the triggering channel
+    shape_near_peak_count   INTEGER,                -- samples >= 0.5 * peak (FT: few; real: many)
+    shape_sample_count      INTEGER,                -- total samples (to normalize near_peak_count)
+    shape_axis              TEXT,                   -- geophone channel measured ("Tran"/"Vert"/"Long")
+```
+
+In `_migrate`, extend the ADD COLUMN tuple (the `for col, ddl in (...)` list) with:
+
+```python
+            ("shape_crest_factor",    "REAL"),
+            ("shape_near_peak_count", "INTEGER"),
+            ("shape_sample_count",    "INTEGER"),
+            ("shape_axis",            "TEXT"),
+```
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_columns.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/database.py tests/test_shape_columns.py
+git commit -m "feat(db): shape_* columns on events (+ auto-migrate)"
+```
+
+---
+
+### Task 4: insert_events persists shape from the waveform record
+
+**Files:**
+- Modify: `sfm/database.py` — `insert_events` INSERT (column list + placeholders + values) and the UPSERT `UPDATE` block.
+- Test: `tests/test_insert_events_shape.py`
+
+**Interfaces:**
+- Consumes: a `waveform_records` rec dict that may carry `shape_crest_factor`, `shape_near_peak_count`, `shape_sample_count`, `shape_axis`.
+- Produces: those four values stored on the row; refreshed on UPSERT.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_insert_events_shape.py
+from sfm.database import SeismoDb
+from tests.helpers_events import make_event  # existing helper used by other insert tests
+
+def test_insert_stores_shape_from_record(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = make_event(serial="BE1", key="0111abcd")
+    rec = {ev._waveform_key.hex(): {
+        "filename": "F.CE0W", "filesize": 10,
+        "shape_crest_factor": 34.0, "shape_near_peak_count": 3,
+        "shape_sample_count": 1024, "shape_axis": "Long"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec)
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 34.0
+    assert row["shape_near_peak_count"] == 3
+    assert row["shape_axis"] == "Long"
+```
+
+> If `tests/helpers_events.make_event` doesn't exist, build the `Event` inline the way `tests/test_zc_freq_columns.py` does (copy its event-construction helper). Keep the assertion identical.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_insert_events_shape.py -q`
+Expected: FAIL — `KeyError`/`sqlite3.OperationalError` (columns not in INSERT) or values are NULL.
+
+- [ ] **Step 3: Implementation**
+
+In `insert_events` INSERT: add `shape_crest_factor, shape_near_peak_count, shape_sample_count, shape_axis` to the column list, add four `?` placeholders, and add these to the VALUES tuple (after the `mic_zc_above_range` value):
+
+```python
+                            rec.get("shape_crest_factor"),
+                            rec.get("shape_near_peak_count"),
+                            rec.get("shape_sample_count"),
+                            rec.get("shape_axis"),
+```
+
+In the UPSERT `UPDATE ... SET`: add
+
+```sql
+                               shape_crest_factor    = COALESCE(?, shape_crest_factor),
+                               shape_near_peak_count = COALESCE(?, shape_near_peak_count),
+                               shape_sample_count    = COALESCE(?, shape_sample_count),
+                               shape_axis            = COALESCE(?, shape_axis),
+```
+
+and the matching params (before `serial, ts`):
+
+```python
+                            rec.get("shape_crest_factor") if rec else None,
+                            rec.get("shape_near_peak_count") if rec else None,
+                            rec.get("shape_sample_count") if rec else None,
+                            rec.get("shape_axis") if rec else None,
+```
+
+(`COALESCE` on UPSERT so a re-import that lacks samples doesn't wipe a previously-computed shape.)
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_insert_events_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/database.py tests/test_insert_events_shape.py
+git commit -m "feat(db): insert_events persists shape_* from waveform record"
+```
+
+---
+
+### Task 5: Populate shape at ingest (the three save paths)
+
+**Files:**
+- Modify: `sfm/waveform_store.py` — in `save`, `save_imported_bw`, `save_imported_idf`, after the `.h5` is written, add its shape to the returned `rec` dict.
+- Test: `tests/test_save_shape.py`
+
+**Interfaces:**
+- Consumes: `sfm.shape_metrics.shape_from_h5`.
+- Produces: `save*` return dicts carry `shape_crest_factor / shape_near_peak_count / shape_sample_count / shape_axis` (present only when the `.h5` had usable samples).
+
+- [ ] **Step 1: Write the failing test** (drives the BW-import path, which the existing suite already exercises)
+
+```python
+# tests/test_save_shape.py
+from sfm.waveform_store import WaveformStore
+from tests.helpers_bw import sample_bw_bytes, sample_serial  # reuse existing import-test fixtures
+
+def test_save_imported_bw_attaches_shape(tmp_path):
+    store = WaveformStore(tmp_path / "waveforms")
+    ev, rec = store.save_imported_bw(sample_bw_bytes(), serial=sample_serial())
+    # A real BW waveform → shape present with a geo axis.
+    assert rec.get("shape_axis") in ("Tran", "Vert", "Long")
+    assert rec["shape_crest_factor"] > 0
+    assert rec["shape_sample_count"] > 200
+```
+
+> Reuse whatever fixture `tests/test_save_imported_bw*.py` already uses for BW bytes; match its import. If the existing BW fixture produces a degenerate/short waveform, use the fixture from the test that asserts a full h5.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_save_shape.py -q`
+Expected: FAIL — `rec` has no `shape_*` keys.
+
+- [ ] **Step 3: Implementation**
+
+Add a helper near the top of `WaveformStore` methods (module-level import `from sfm.shape_metrics import shape_from_h5`). In each of `save`, `save_imported_bw`, `save_imported_idf`, immediately before building/returning the `rec` dict — and only when the `.h5` was written (i.e. `hdf5_filename`/`hdf5_path` is set) — compute and merge:
+
+```python
+        shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        # ... in the returned rec dict literal, add:
+        #   **(shape and {
+        #       "shape_crest_factor":    shape["crest_factor"],
+        #       "shape_near_peak_count": shape["near_peak_count"],
+        #       "shape_sample_count":    shape["sample_count"],
+        #       "shape_axis":            shape["axis"],
+        #   } or {}),
+```
+
+Concretely, after each method computes `hdf5_filename`, add before its `return {...}`:
+
+```python
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
+```
+
+and spread `**_shape_rec` into the returned dict. (`save_imported_bw`/`save_imported_idf` use their own hdf5 path variable names — use whichever local holds the written `.h5` path in each method.)
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_save_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/waveform_store.py tests/test_save_shape.py
+git commit -m "feat(ingest): compute shape from the written .h5 in all save paths"
+```
+
+---
+
+### Task 6: Backfill script for existing events
+
+**Files:**
+- Create: `scripts/backfill_event_shape.py` (mirror `scripts/backfill_event_zc_freq.py`, but read the `.h5` for samples instead of the sidecar).
+- Test: `tests/test_backfill_event_shape.py`
+
+**Interfaces:**
+- Produces: `backfill_shape(db, store, *, dry_run=False) -> dict` with counts `{"updated","skipped_no_h5","skipped_no_samples"}`; `main(argv)` CLI mirroring the zc-freq script's args (`--db-path`, `--store-root`, `--dry-run`).
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_backfill_event_shape.py
+import numpy as np, h5py
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from scripts.backfill_event_shape import backfill_shape
+from tests.helpers_events import make_event   # or inline as in test_zc_freq_columns
+
+def _h5(path, long):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k in ("Tran", "Vert"): g.create_dataset(k, data=np.zeros(1024, "float32"))
+        g.create_dataset("Long", data=np.asarray(long, "float32"))
+
+def test_backfill_updates_shape_and_is_idempotent(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    store = WaveformStore(tmp_path / "waveforms")
+    ev = make_event(serial="BE1", key="0111abcd")
+    db.insert_events([ev], serial="BE1",
+                     waveform_records={ev._waveform_key.hex():
+                                       {"filename": "F.CE0W", "filesize": 10}})
+    # place the .h5 where store.paths_for expects it
+    long = np.zeros(1024); long[100] = 0.48
+    _h5(store.hdf5_path_for("BE1", "F.CE0W"), long)
+
+    c1 = backfill_shape(db, store)
+    assert c1["updated"] == 1
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_axis"] == "Long" and row["shape_near_peak_count"] <= 3
+    c2 = backfill_shape(db, store)                  # idempotent: re-run overwrites same values
+    assert db.query_events(serial="BE1")[0]["shape_crest_factor"] == row["shape_crest_factor"]
+```
+
+> Match `make_event` / column names to whatever `tests/test_zc_freq_columns.py` uses. `store.hdf5_path_for(serial, filename)` is the existing helper that returns the `.h5` path.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_backfill_event_shape.py -q`
+Expected: FAIL — `ModuleNotFoundError: scripts.backfill_event_shape`.
+
+- [ ] **Step 3: Implementation** (mirror the zc-freq script structure)
+
+```python
+#!/usr/bin/env python3
+"""Backfill events.shape_* from each event's .h5 waveform samples. Idempotent."""
+from __future__ import annotations
+import argparse, logging, sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from sfm.shape_metrics import shape_from_h5
+
+log = logging.getLogger("backfill_event_shape")
+
+def backfill_shape(db: SeismoDb, store: WaveformStore, *, dry_run: bool = False) -> dict:
+    counts = {"updated": 0, "skipped_no_h5": 0, "skipped_no_samples": 0}
+    for row in db.query_events(limit=1_000_000):
+        serial, filename = row.get("serial"), row.get("blastware_filename")
+        if not serial or not filename:
+            counts["skipped_no_h5"] += 1; continue
+        h5_path = store.hdf5_path_for(serial, filename)
+        if not h5_path.exists():
+            counts["skipped_no_h5"] += 1; continue
+        shape = shape_from_h5(h5_path)
+        if shape is None:
+            counts["skipped_no_samples"] += 1; continue
+        if not dry_run:
+            with db._connect() as conn:
+                conn.execute(
+                    "UPDATE events SET shape_crest_factor=?, shape_near_peak_count=?, "
+                    "shape_sample_count=?, shape_axis=? WHERE id=?",
+                    (shape["crest_factor"], shape["near_peak_count"],
+                     shape["sample_count"], shape["axis"], row["id"]))
+        counts["updated"] += 1
+    log.info("backfill_shape: %s", counts)
+    return counts
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db-path", required=True)
+    ap.add_argument("--store-root", required=True)
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    logging.basicConfig(level=logging.INFO)
+    counts = backfill_shape(SeismoDb(a.db_path), WaveformStore(a.store_root), dry_run=a.dry_run)
+    print(counts)
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+```
+
+> `store.hdf5_path_for` and `db._connect` are existing internals used the same way by other scripts. If `hdf5_path_for` isn't public, use `store.paths_for(...)` sibling `.h5` path exactly as `save()` derives `hdf5_path`.
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_backfill_event_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Run the full suite + commit**
+
+```bash
+/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest -q
+git add scripts/backfill_event_shape.py tests/test_backfill_event_shape.py
+git commit -m "feat(scripts): backfill events.shape_* from .h5 samples"
+```
+
+---
+
+### Task 7: Confirm /db/events carries shape + version bump
+
+**Files:**
+- Test: `tests/test_db_events_exposes_shape.py`
+- Modify: `pyproject.toml` version; `sfm/server.py` version string; `CHANGELOG.md`.
+
+**Interfaces:**
+- Consumes: the running `/db/events` route (already returns `SELECT *`).
+
+- [ ] **Step 1: Write the failing test** (guards that the feed dict includes the new keys)
+
+```python
+# tests/test_db_events_exposes_shape.py
+from sfm.database import SeismoDb
+def test_query_events_row_includes_shape_keys(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    # query_events returns dict(row); a fresh insert has the keys (values may be None)
+    from tests.helpers_events import make_event
+    db.insert_events([make_event(serial="BE1", key="0111abcd")], serial="BE1")
+    row = db.query_events(serial="BE1")[0]
+    for k in ("shape_crest_factor","shape_near_peak_count","shape_sample_count","shape_axis"):
+        assert k in row
+```
+
+- [ ] **Step 2: Run to verify it fails / passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_db_events_exposes_shape.py -q`
+Expected: PASS immediately if Task 3 landed (columns present in `SELECT *`). If it fails, the columns weren't added — fix Task 3. (This task is the guard, not new behavior.)
+
+- [ ] **Step 3: Version bump**
+
+Bump `pyproject.toml` `version` 0.23.0 → 0.24.0; set `sfm/server.py` `version="0.24.0"`; add a `## v0.24.0` CHANGELOG entry ("waveform-shape metrics on events: crest factor + points-near-peak, ingest + backfill").
+
+- [ ] **Step 4: Full suite**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest -q`
+Expected: PASS (no regressions).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add tests/test_db_events_exposes_shape.py pyproject.toml sfm/server.py CHANGELOG.md
+git commit -m "chore(release): v0.24.0 — waveform-shape metrics on events"
+```
+
+---
+
+## Self-Review
+
+**Spec coverage:** SFM shape columns (Tasks 3–4) ✓; DSP crest + near-peak (Task 1) ✓; ingest population (Task 5) ✓; backfill (Task 6) ✓; `/db/events` exposure (Task 7) ✓; NULL for histogram/no-sample events (Tasks 1/5/6 return None → NULL) ✓; 94%-coverage / series-4 fallback handled by NULL-then-Terra-View-fallback (Phase B) ✓. Terra-View scoring, 3-state review, twin flagging, export Notes column → **Phase B plan** (separate, depends on this feed). Calibration → Phase C.
+
+**Placeholder scan:** No TBD/TODO; every code step has real code. The two "reuse existing fixture" notes point at concrete existing tests (`test_zc_freq_columns.py`, `test_save_imported_bw*.py`) rather than leaving blanks.
+
+**Type consistency:** `shape_from_h5`/`shape_from_samples`/`channel_shape` return the same dict keys (`crest_factor`, `near_peak_count`, `sample_count`, `axis`) throughout; the DB columns (`shape_crest_factor`, `shape_near_peak_count`, `shape_sample_count`, `shape_axis`) and rec keys match across Tasks 4–6.
+
+## Deferred to Phase B (terra-view, separate plan)
+
+Scoring service combining shape + cheap signals; suspicion column + reason chips; `reviewed_real` mirror + 3-state review; twin-aware flag propagation; Notes-column export + Maximums "(excludes N flagged)". Written once this feed is live so column names/values are real.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "seismo-relay"
-version = "0.23.0"
+version = "0.24.0"
 description = "Python client and REST server for MiniMate Plus seismographs"
 requires-python = ">=3.10"
 dependencies = [

scripts/backfill_event_shape.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""Backfill events.shape_* from each event's .h5 waveform samples. Idempotent."""
+from __future__ import annotations
+import argparse, logging, sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from sfm.shape_metrics import shape_from_h5
+
+log = logging.getLogger("backfill_event_shape")
+
+def backfill_shape(db: SeismoDb, store: WaveformStore, *, dry_run: bool = False) -> dict:
+    counts = {"updated": 0, "skipped_no_h5": 0, "skipped_no_samples": 0}
+    for row in db.query_events(limit=1_000_000):
+        serial, filename = row.get("serial"), row.get("blastware_filename")
+        if not serial or not filename:
+            counts["skipped_no_h5"] += 1; continue
+        h5_path = store.hdf5_path_for(serial, filename)
+        if not h5_path.exists():
+            counts["skipped_no_h5"] += 1; continue
+        shape = shape_from_h5(h5_path)
+        if shape is None:
+            counts["skipped_no_samples"] += 1; continue
+        if not dry_run:
+            with db._connect() as conn:
+                conn.execute(
+                    "UPDATE events SET shape_crest_factor=?, shape_near_peak_count=?, "
+                    "shape_sample_count=?, shape_axis=? WHERE id=?",
+                    (shape["crest_factor"], shape["near_peak_count"],
+                     shape["sample_count"], shape["axis"], row["id"]))
+        counts["updated"] += 1
+    log.info("backfill_shape: %s", counts)
+    return counts
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db-path", required=True)
+    ap.add_argument("--store-root", required=True)
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    logging.basicConfig(level=logging.INFO)
+    counts = backfill_shape(SeismoDb(a.db_path), WaveformStore(a.store_root), dry_run=a.dry_run)
+    print(counts)
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())

sfm/database.py

@@ -94,6 +94,10 @@ CREATE TABLE IF NOT EXISTS events (
     vert_zc_above_range     INTEGER,
     long_zc_above_range     INTEGER,
     mic_zc_above_range      INTEGER,
+    shape_crest_factor      REAL,                   -- peak / rms of the triggering channel
+    shape_near_peak_count   INTEGER,                -- samples >= 0.5 * peak (FT: few; real: many)
+    shape_sample_count      INTEGER,                -- total samples (to normalize near_peak_count)
+    shape_axis              TEXT,                   -- geophone channel measured ("Tran"/"Vert"/"Long")
     created_at              TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
     UNIQUE(serial, timestamp)
 );
@@ -216,6 +220,10 @@ class SeismoDb:
             ("vert_zc_above_range", "INTEGER"),
             ("long_zc_above_range", "INTEGER"),
             ("mic_zc_above_range",  "INTEGER"),
+            ("shape_crest_factor",    "REAL"),
+            ("shape_near_peak_count", "INTEGER"),
+            ("shape_sample_count",    "INTEGER"),
+            ("shape_axis",            "TEXT"),
         ):
             if col not in existing_cols:
                 log.info("_migrate: events ADD COLUMN %s %s", col, ddl)
@@ -418,9 +426,11 @@ class SeismoDb:
                              device_family,
                              tran_zc_freq, vert_zc_freq, long_zc_freq, mic_zc_freq,
                              tran_zc_above_range, vert_zc_above_range,
-                             long_zc_above_range, mic_zc_above_range)
+                             long_zc_above_range, mic_zc_above_range,
+                             shape_crest_factor, shape_near_peak_count,
+                             shape_sample_count, shape_axis)
                         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
-                                ?, ?, ?, ?, ?, ?, ?, ?)
+                                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                         """,
                         (
                             self._new_id(), serial, key, session_id, ts,
@@ -448,6 +458,10 @@ class SeismoDb:
                             (1 if (pv and pv.vert_zc_above_range) else 0),
                             (1 if (pv and pv.long_zc_above_range) else 0),
                             (1 if (pv and pv.mic_zc_above_range)  else 0),
+                            rec.get("shape_crest_factor"),
+                            rec.get("shape_near_peak_count"),
+                            rec.get("shape_sample_count"),
+                            rec.get("shape_axis"),
                         ),
                     )
                     inserted += 1
@@ -497,7 +511,11 @@ class SeismoDb:
                                tran_zc_above_range = ?,
                                vert_zc_above_range = ?,
                                long_zc_above_range = ?,
-                               mic_zc_above_range  = ?
+                               mic_zc_above_range  = ?,
+                               shape_crest_factor    = COALESCE(?, shape_crest_factor),
+                               shape_near_peak_count = COALESCE(?, shape_near_peak_count),
+                               shape_sample_count    = COALESCE(?, shape_sample_count),
+                               shape_axis            = COALESCE(?, shape_axis)
                          WHERE serial = ? AND timestamp = ?
                         """,
                         (
@@ -525,6 +543,10 @@ class SeismoDb:
                             (1 if (pv and pv.vert_zc_above_range) else 0),
                             (1 if (pv and pv.long_zc_above_range) else 0),
                             (1 if (pv and pv.mic_zc_above_range)  else 0),
+                            rec.get("shape_crest_factor")    if rec else None,
+                            rec.get("shape_near_peak_count") if rec else None,
+                            rec.get("shape_sample_count")    if rec else None,
+                            rec.get("shape_axis")            if rec else None,
                             serial,
                             ts,
                         ),

sfm/server.py

@@ -90,7 +90,7 @@ app = FastAPI(
         "Implements the minimateplus RS-232 protocol library.\n"
         "Proxied by terra-view at /api/sfm/*."
     ),
-    version="0.23.0",
+    version="0.24.0",
 )
 
 # Allow requests from the waveform viewer opened as a local file (file://)

sfm/shape_metrics.py

@@ -0,0 +1,58 @@
+"""Waveform-shape metrics for false-trigger detection.
+
+A false trigger is an isolated impulse (quiet → spike → quiet); a real event
+rings for many cycles. Two numbers separate them: crest factor (how far the
+peak stands above the typical sample) and how many samples sit near the peak.
+"""
+from __future__ import annotations
+import numpy as np
+
+_GEO_CHANNELS = ("Tran", "Vert", "Long")
+NEAR_PEAK_FRACTION = 0.5   # a sample "near the peak" is >= this * peak amplitude
+
+
+def channel_shape(x) -> dict | None:
+    x = np.asarray(x, dtype=float)
+    if x.size < 2:
+        return None
+    peak = float(np.max(np.abs(x)))
+    if peak <= 0:
+        return None
+    rms = float(np.sqrt(np.mean(x ** 2)))
+    if rms <= 0:
+        return None
+    near = int(np.sum(np.abs(x) >= NEAR_PEAK_FRACTION * peak))
+    return {"crest_factor": peak / rms, "near_peak_count": near,
+            "sample_count": int(x.size)}
+
+
+def shape_from_samples(chans: dict) -> dict | None:
+    best_axis, best_peak, best_x = None, -1.0, None
+    for ax in _GEO_CHANNELS:
+        x = chans.get(ax)
+        if x is None:
+            continue
+        x = np.asarray(x, dtype=float)
+        if x.size < 2:
+            continue
+        p = float(np.max(np.abs(x)))
+        if p > best_peak:
+            best_axis, best_peak, best_x = ax, p, x
+    if best_axis is None:
+        return None
+    s = channel_shape(best_x)
+    if s is None:
+        return None
+    s["axis"] = best_axis
+    return s
+
+
+def shape_from_h5(path) -> dict | None:
+    import h5py
+    try:
+        with h5py.File(path, "r") as f:
+            chans = {ax: f[f"samples/{ax}"][:] for ax in _GEO_CHANNELS
+                     if f"samples/{ax}" in f}
+    except Exception:
+        return None
+    return shape_from_samples(chans)

sfm/waveform_store.py

@@ -41,6 +41,7 @@ from minimateplus.blastware_file import blastware_filename, write_blastware_file
 from minimateplus.framing import S3Frame
 from minimateplus.models import Event
 from sfm import event_hdf5
+from sfm.shape_metrics import shape_from_h5
 
 log = logging.getLogger("sfm.waveform_store")
 
@@ -262,6 +263,13 @@ class WaveformStore:
             serial, filename, filesize, len(a5_frames),
             hdf5_filename or "(skipped)", sidecar_path.name,
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return {
             "filename":            filename,
             "filesize":            filesize,
@@ -269,6 +277,7 @@ class WaveformStore:
             "a5_pickle_filename":  a5_path.name,
             "hdf5_filename":       hdf5_filename,
             "sidecar_filename":    sidecar_path.name,
+            **_shape_rec,
         }
 
     def save_imported_bw(
@@ -445,6 +454,13 @@ class WaveformStore:
             "h5=%s (no .a5.pkl — A5 source unavailable for BW-imported files)",
             serial, filename, filesize, hdf5_filename or "(skipped)",
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return ev, {
             "filename":           filename,
             "filesize":           filesize,
@@ -453,6 +469,7 @@ class WaveformStore:
             "hdf5_filename":      hdf5_filename,
             "sidecar_filename":   sidecar_path.name,
             "serial":             serial,
+            **_shape_rec,
         }
 
     def save_imported_idf(
@@ -727,6 +744,13 @@ class WaveformStore:
             hdf5_filename or "(skipped)",
             len(idf_intervals) if idf_intervals else 0,
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return ev, {
             "filename":           filename,
             "filesize":           filesize,
@@ -735,6 +759,7 @@ class WaveformStore:
             "hdf5_filename":      hdf5_filename,
             "sidecar_filename":   sidecar_path.name,
             "serial":             serial,
+            **_shape_rec,
         }
 
     def load_a5(self, serial: str, filename: str) -> Optional[list[S3Frame]]:

tests/test_backfill_event_shape.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+import numpy as np, h5py
+
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from scripts.backfill_event_shape import backfill_shape
+from minimateplus.models import Event
+
+
+def _h5(path, long):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k in ("Tran", "Vert"):
+            g.create_dataset(k, data=np.zeros(1024, "float32"))
+        g.create_dataset("Long", data=np.asarray(long, "float32"))
+
+
+def test_backfill_updates_shape_and_is_idempotent(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    store = WaveformStore(tmp_path / "waveforms")
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex("0111abcd")
+    db.insert_events([ev], serial="BE1",
+                     waveform_records={ev._waveform_key.hex():
+                                       {"filename": "F.CE0W", "filesize": 10}})
+    # place the .h5 where store.paths_for expects it
+    long = np.zeros(1024); long[100] = 0.48
+    _h5(store.hdf5_path_for("BE1", "F.CE0W"), long)
+
+    c1 = backfill_shape(db, store)
+    assert c1["updated"] == 1
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_axis"] == "Long" and row["shape_near_peak_count"] <= 3
+    c2 = backfill_shape(db, store)                  # idempotent: re-run overwrites same values
+    assert db.query_events(serial="BE1")[0]["shape_crest_factor"] == row["shape_crest_factor"]

tests/test_db_events_exposes_shape.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+import os, sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+
+SHAPE_COLS = (
+    "shape_crest_factor",
+    "shape_near_peak_count",
+    "shape_sample_count",
+    "shape_axis",
+)
+
+
+def test_query_events_row_includes_shape_keys(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex("0111abcd")
+
+    db.insert_events([ev], serial="BE1")
+    row = db.query_events(serial="BE1")[0]
+
+    for k in SHAPE_COLS:
+        assert k in row
+        assert row[k] is None

tests/test_insert_events_shape.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+from pathlib import Path
+
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp, PeakValues
+
+
+def _event(waveform_key="0111abcd"):
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex(waveform_key)
+    ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0,
+                             month=6, day=25, hour=8, minute=50, second=0)
+    ev.record_type = "Waveform"
+    ev.peak_values = PeakValues(tran=0.075, vert=0.220, long=0.045,
+                                 peak_vector_sum=0.231, micl=0.01)
+    return ev
+
+
+def test_insert_stores_shape_from_record(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec = {ev._waveform_key.hex(): {
+        "filename": "F.CE0W", "filesize": 10,
+        "shape_crest_factor": 34.0, "shape_near_peak_count": 3,
+        "shape_sample_count": 1024, "shape_axis": "Long"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec)
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 34.0
+    assert row["shape_near_peak_count"] == 3
+    assert row["shape_axis"] == "Long"
+
+
+def test_upsert_refreshes_shape_from_record(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec1 = {ev._waveform_key.hex(): {
+        "shape_crest_factor": 10.0, "shape_near_peak_count": 1,
+        "shape_sample_count": 512, "shape_axis": "Tran"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec1)
+
+    ev2 = _event()
+    rec2 = {ev2._waveform_key.hex(): {
+        "shape_crest_factor": 22.5, "shape_near_peak_count": 7,
+        "shape_sample_count": 2048, "shape_axis": "Vert"}}
+    db.insert_events([ev2], serial="BE1", waveform_records=rec2)
+
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 22.5
+    assert row["shape_near_peak_count"] == 7
+    assert row["shape_sample_count"] == 2048
+    assert row["shape_axis"] == "Vert"
+
+
+def test_upsert_preserves_shape_when_record_missing(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec1 = {ev._waveform_key.hex(): {
+        "shape_crest_factor": 10.0, "shape_near_peak_count": 1,
+        "shape_sample_count": 512, "shape_axis": "Tran"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec1)
+
+    ev2 = _event()  # re-import with no waveform_records (e.g. sample missing)
+    db.insert_events([ev2], serial="BE1")
+
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 10.0
+    assert row["shape_near_peak_count"] == 1
+    assert row["shape_sample_count"] == 512
+    assert row["shape_axis"] == "Tran"

tests/test_save_shape.py

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+from sfm.waveform_store import WaveformStore
+
+_FIX = Path(__file__).parent / "fixtures/histogram-extension-re/events-5-21-26/K558LL8B.7I0W"
+
+
+def test_save_imported_bw_attaches_shape(tmp_path):
+    store = WaveformStore(tmp_path / "waveforms")
+    ev, rec = store.save_imported_bw(_FIX.read_bytes(), source_path=_FIX, serial_hint="BE9558")
+    assert rec.get("shape_axis") in ("Tran", "Vert", "Long")
+    assert rec["shape_crest_factor"] > 0
+    assert rec["shape_sample_count"] > 200

tests/test_shape_columns.py

@@ -0,0 +1,27 @@
+import sqlite3
+from sfm.database import SeismoDb
+
+_SHAPE_COLS = {"shape_crest_factor", "shape_near_peak_count",
+               "shape_sample_count", "shape_axis"}
+
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+
+def test_fresh_db_has_shape_columns(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    assert _SHAPE_COLS <= _cols(db)
+
+def test_existing_db_migrates_shape_columns(tmp_path):
+    p = tmp_path / "s.db"
+    db = SeismoDb(p)
+    with sqlite3.connect(p) as c:      # simulate an older DB missing the columns
+        for col in _SHAPE_COLS:
+            c.execute(f"ALTER TABLE events DROP COLUMN {col}")
+    # sanity: dropped. Read via the already-constructed `db` (a raw PRAGMA
+    # read against db.db_path) rather than a fresh SeismoDb(p) — the latter
+    # would immediately re-trigger _migrate's self-healing ADD COLUMN loop
+    # and mask the gap we're trying to confirm.
+    assert not (_SHAPE_COLS <= _cols(db))
+    SeismoDb(p)                        # re-open triggers _migrate
+    assert _SHAPE_COLS <= _cols(SeismoDb(p))

tests/test_shape_metrics.py

@@ -0,0 +1,31 @@
+import numpy as np
+from sfm.shape_metrics import channel_shape, shape_from_samples
+
+def test_needle_spike_high_crest_few_near_peak():
+    x = np.zeros(1024); x[500] = 1.0          # one isolated spike
+    s = channel_shape(x)
+    assert s["sample_count"] == 1024
+    assert s["crest_factor"] > 15             # peak towers over rms
+    assert s["near_peak_count"] <= 3          # almost nothing near the peak
+
+def test_ringing_low_crest_many_near_peak():
+    t = np.arange(1024)
+    x = np.sin(2*np.pi*t/32) * np.exp(-t/4000)  # decaying oscillation
+    s = channel_shape(x)
+    assert s["crest_factor"] < 6
+    assert s["near_peak_count"] > 30          # many samples near the peak
+
+def test_channel_shape_none_for_unusable():
+    assert channel_shape(np.zeros(1024)) is None   # flat / all-zero
+    assert channel_shape(np.array([1.0])) is None  # too short
+
+def test_shape_from_samples_picks_max_peak_axis():
+    chans = {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": np.zeros(1024)}
+    chans["Long"][10] = 0.5
+    chans["Vert"] = np.sin(np.arange(1024)/5) * 0.01
+    s = shape_from_samples(chans)
+    assert s["axis"] == "Long"                # Long has the biggest peak
+    assert s["near_peak_count"] <= 3
+
+def test_shape_from_samples_none_when_no_geo():
+    assert shape_from_samples({"MicL": np.ones(1024)}) is None

tests/test_shape_metrics_h5.py

@@ -0,0 +1,22 @@
+import numpy as np, h5py
+from sfm.shape_metrics import shape_from_h5
+
+def _write_h5(path, chans):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k, v in chans.items():
+            g.create_dataset(k, data=np.asarray(v, dtype="float32"))
+
+def test_shape_from_h5_reads_dominant_axis(tmp_path):
+    p = tmp_path / "ev.h5"
+    long = np.zeros(1024, dtype="float32"); long[100] = 0.48
+    _write_h5(p, {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": long,
+                  "MicL": np.ones(1024)})
+    s = shape_from_h5(str(p))
+    assert s["axis"] == "Long" and s["near_peak_count"] <= 3
+
+def test_shape_from_h5_none_on_missing_or_degenerate(tmp_path):
+    assert shape_from_h5(str(tmp_path / "nope.h5")) is None
+    p = tmp_path / "degen.h5"
+    _write_h5(p, {"Tran": np.zeros(1), "Vert": np.zeros(1), "Long": np.zeros(1)})
+    assert shape_from_h5(str(p)) is None