Files
seismo-relay/docs/superpowers/plans/2026-08-21-waveform-ft-detection-phase-a.md
serversdownandClaude Opus 4.8 eb92b13aac docs(plan): waveform-shape FT detection — Phase A (seismo-relay)
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
2026-08-21 21:32:59 +00:00

26 KiB
Raw Permalink Blame History

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

# 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
# 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
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)

# 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
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

# 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:

            ("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
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

# 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):

                            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

                               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):

                            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
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)

# 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:

        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 {...}:

        _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
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

# 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)
#!/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
/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)

# 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
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.