"""Capture loading: the header contract, the flag columns, and hostile files. capture.py is the only reader of a capture, so every guarantee the format claims -- contiguous index, parseable counts, recoverable time base -- has to be checked here or nowhere. """ import numpy as np import pytest import capture import rm3100 # The writer lives in synthetic.py because test_compare.py needs it too: # one definition of what a valid capture looks like, not two that drift. from synthetic import CYCLE_COUNT, DT, write_capture # -------------------------------------------------------------------------- # The happy path # -------------------------------------------------------------------------- def test_load_round_trips_a_clean_capture(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=500)) assert len(cap.sample_index) == 500 assert cap.cycle_count == CYCLE_COUNT assert not cap.missed.any() assert not cap.ambiguous.any() def test_counts_are_converted_with_the_header_constant(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv")) assert cap.lsb_nt == pytest.approx( rm3100.tesla_per_count(CYCLE_COUNT) * rm3100.NT_PER_TESLA) # x_raw was written around 1000 counts. assert cap.x.mean() == pytest.approx(1000 * cap.lsb_nt, rel=0.01) def test_total_is_the_norm_of_the_three_axes(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv")) assert cap.total == pytest.approx( np.sqrt(cap.x**2 + cap.y**2 + cap.z**2)) def test_true_period_is_recovered_from_the_host_clock(tmp_path): """The whole point of the fit: the nominal rate is wrong and the file says so.""" cap = capture.load(write_capture(tmp_path / "c.csv", dt=DT, nominal_hz=282.0)) # Microsecond-resolution timestamps are the only error left in the slope. assert cap.dt_true == pytest.approx(DT, rel=1e-7) assert cap.true_rate_hz == pytest.approx(250.0, rel=1e-7) def test_the_epoch_offset_does_not_degrade_the_fit(tmp_path): """polyfit does not centre the ordinate, so the fit has to do it itself. Fitting raw epoch seconds against a millisecond slope costs six significant digits to cancellation; a capture starting at t=0 and the same capture starting in 2023 must recover the same period. """ at_zero = capture.load(write_capture(tmp_path / "zero.csv", rows=300, start_time=0.0)) at_epoch = capture.load(write_capture(tmp_path / "epoch.csv", rows=300, start_time=1_700_000_000.0)) assert at_epoch.dt_true == pytest.approx(at_zero.dt_true, rel=1e-7) def test_rate_error_is_signed_like_the_rate_not_the_period(tmp_path): """Chip slower than nominal must read negative, matching the reported Hz.""" slow = capture.load(write_capture(tmp_path / "slow.csv", dt=1 / 250.0, nominal_hz=282.0)) fast = capture.load(write_capture(tmp_path / "fast.csv", dt=1 / 300.0, nominal_hz=282.0)) assert slow.rate_error < 0 assert fast.rate_error > 0 assert slow.rate_error == pytest.approx(250.0 / 282.0 - 1, rel=1e-6) def test_elapsed_bases_differ_by_exactly_the_rate_error(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv")) assert cap.elapsed[0] == 0.0 ratio = cap.elapsed[-1] / cap.elapsed_nominal[-1] assert ratio == pytest.approx(cap.dt_true / cap.dt_nominal) def test_duration_spans_the_capture(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=500, dt=DT)) assert cap.duration == pytest.approx(499 * DT) def test_axes_lists_three_axes_plus_the_derived_total(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv")) assert [k for k, _ in cap.axes()] == ["x", "y", "z", "total"] def test_note_reaches_the_summary(tmp_path): path = write_capture(tmp_path / "c.csv", header={"note": "3V0 LDO"}) assert "3V0 LDO" in capture.load(path).summary() # -------------------------------------------------------------------------- # MISSED and AMBIGUOUS # -------------------------------------------------------------------------- def test_missed_rows_are_flagged_and_interpolated(tmp_path): path = write_capture(tmp_path / "c.csv", rows=200, flags={50: capture.WARN_MISSED}) cap = capture.load(path) assert cap.missed.sum() == 1 assert cap.missed[50] # The placeholder's zeros must not survive into the data. assert cap.x[50] == pytest.approx((cap.x[49] + cap.x[51]) / 2, rel=0.01) def test_consecutive_missed_rows_interpolate_across_the_whole_gap(tmp_path): path = write_capture(tmp_path / "c.csv", rows=200, flags={i: capture.WARN_MISSED for i in (60, 61, 62)}) cap = capture.load(path) assert cap.missed.sum() == 3 assert cap.x[59] < cap.x[62] or cap.x[59] > cap.x[62] # finite, not zero assert np.all(np.abs(cap.x[60:63]) > 0) def test_ambiguous_is_independent_of_missed(tmp_path): """A gap can round to zero losses and still be a coin toss.""" path = write_capture(tmp_path / "c.csv", rows=200, flags={ 70: capture.WARN_AMBIGUOUS, 80: f"{capture.WARN_MISSED} {capture.WARN_AMBIGUOUS}", }) cap = capture.load(path) assert cap.ambiguous.sum() == 2 assert cap.missed.sum() == 1 assert cap.ambiguous[70] and not cap.missed[70] # real row, flagged assert cap.ambiguous[80] and cap.missed[80] # placeholder, both def test_summary_reports_both_flag_kinds(tmp_path): path = write_capture(tmp_path / "c.csv", rows=200, flags={ 70: capture.WARN_AMBIGUOUS, 80: capture.WARN_MISSED}) text = capture.load(path).summary() assert "lost measurement" in text assert "AMBIGUOUS" in text def test_flag_order_within_the_column_does_not_matter(tmp_path): path = write_capture(tmp_path / "c.csv", rows=200, flags={ 90: f"{capture.WARN_AMBIGUOUS} {capture.WARN_MISSED}"}) cap = capture.load(path) assert cap.missed[90] and cap.ambiguous[90] # -------------------------------------------------------------------------- # Drift detection # -------------------------------------------------------------------------- def test_a_uniform_capture_is_not_drift_limited(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=500)) assert not cap.drift_limited assert cap.residual_sd < cap.dt_true def test_a_rate_that_changes_mid_run_is_drift_limited(tmp_path): """One slope cannot describe two rates; that has to be visible.""" path = tmp_path / "drift.csv" write_capture(path, rows=400) lines = path.read_text().splitlines() head = [l for l in lines if l.startswith("#")] body = [l for l in lines if not l.startswith("#")] out = head + [body[0]] t = 1_700_000_000.0 for i, row in enumerate(body[1:]): fields = row.split(",") # Second half runs 30% slower -- far more than any real oscillator, but # this is testing the detector, not the sensor. t += DT if i < 200 else DT * 1.3 out.append(",".join([fields[0], f"{t:.6f}"] + fields[2:])) path.write_text("\n".join(out) + "\n") cap = capture.load(str(path)) assert cap.drift_limited assert "single-rate model does not fit" in cap.summary() def test_read_jitter_is_reported_apart_from_the_fit_residual(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=500)) assert cap.read_jitter >= 0.0 assert cap.residual_sd >= 0.0 # -------------------------------------------------------------------------- # restrict() # -------------------------------------------------------------------------- def test_restrict_refits_on_the_window(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=1000)) window = cap.restrict(0.5, 1.5) assert len(window.sample_index) < len(cap.sample_index) assert window.elapsed[0] >= 0.0 assert window.dt_true == pytest.approx(cap.dt_true, rel=1e-6) def test_restrict_accepts_an_open_ended_window(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=1000)) assert len(cap.restrict(start=0.5).sample_index) < 1000 assert len(cap.restrict(end=1.0).sample_index) < 1000 def test_restrict_refuses_a_window_too_small_to_analyse(tmp_path): cap = capture.load(write_capture(tmp_path / "c.csv", rows=500)) with pytest.raises(capture.CaptureError, match="too few samples|too few"): cap.restrict(0.0, 0.01) def test_restrict_carries_the_flags_through(tmp_path): path = write_capture(tmp_path / "c.csv", rows=1000, flags={500: capture.WARN_MISSED}) cap = capture.load(path) window = cap.restrict(400 * DT, 600 * DT) assert window.missed.sum() == 1 # -------------------------------------------------------------------------- # Rejected files -- each guarantee the format makes, tested by breaking it # -------------------------------------------------------------------------- def test_a_file_without_the_capture_header_is_rejected(tmp_path): path = tmp_path / "old.csv" path.write_text("sample_index,system_time_unix,x_raw,y_raw,z_raw,warning\n" + "".join(f"{i},{i * 0.004},1,2,3,\n" for i in range(100))) with pytest.raises(capture.CaptureError, match="no capture header"): capture.load(str(path)) @pytest.mark.parametrize("key", ["cycle_count", "tesla_per_count", "nominal_rate_hz"]) def test_a_header_missing_a_required_key_is_rejected(tmp_path, key): path = write_capture(tmp_path / "c.csv", drop_header_key=key) with pytest.raises(capture.CaptureError, match=f"header missing.*{key}"): capture.load(path) def test_too_few_samples_to_analyse_is_rejected(tmp_path): path = write_capture(tmp_path / "short.csv", rows=10) with pytest.raises(capture.CaptureError, match="too few"): capture.load(path) def test_a_capture_without_the_warning_column_is_rejected(tmp_path): path = tmp_path / "nowarn.csv" body = "".join(f"{i},{1_700_000_000 + i * 0.004:.6f},1,2,3\n" for i in range(100)) path.write_text(f"# rm3100_capture: 1\n# nominal_rate_hz: 282\n" f"# cycle_count: 100\n" f"# tesla_per_count: {rm3100.tesla_per_count(100)!r}\n" "sample_index,system_time_unix,x_raw,y_raw,z_raw\n" + body) with pytest.raises(capture.CaptureError, match="no 'warning' column"): capture.load(str(path)) def test_a_non_contiguous_index_is_rejected(tmp_path): """A hole means the grid is broken, so the whole time base is unusable.""" path = tmp_path / "holed.csv" write_capture(path, rows=200) lines = path.read_text().splitlines() del lines[100] # drop one data row path.write_text("\n".join(lines) + "\n") with pytest.raises(capture.CaptureError, match="not contiguous"): capture.load(str(path)) def test_a_decreasing_index_is_rejected(tmp_path): path = write_capture(tmp_path / "back.csv", rows=100, index_step=-1, index_from=500) with pytest.raises(capture.CaptureError, match="not contiguous"): capture.load(path) def test_an_all_missed_capture_is_rejected(tmp_path): path = write_capture(tmp_path / "empty.csv", rows=100, flags={i: capture.WARN_MISSED for i in range(100)}) with pytest.raises(capture.CaptureError, match="every row is a lost"): capture.load(path) def test_a_missing_file_raises_oserror_not_captureerror(tmp_path): with pytest.raises(OSError): capture.load(str(tmp_path / "nope.csv")) # -------------------------------------------------------------------------- # Malformed and adversarial input # -------------------------------------------------------------------------- def test_a_header_line_without_a_colon_is_ignored(tmp_path): path = write_capture(tmp_path / "c.csv", extra_lines=["# just a comment with no key"]) cap = capture.load(path) assert "just a comment with no key" not in str(cap.meta) def test_an_index_starting_away_from_zero_still_loads(tmp_path): """A capture cut from the middle of a run is still a valid grid.""" cap = capture.load(write_capture(tmp_path / "c.csv", rows=200, index_from=1_000_000)) assert cap.sample_index[0] == 1_000_000 assert cap.elapsed[0] == pytest.approx(1_000_000 * cap.dt_true) def test_a_huge_index_does_not_lose_precision(tmp_path): """int64 indices must not silently become floats in the fit.""" cap = capture.load(write_capture(tmp_path / "c.csv", rows=200, index_from=2**40)) assert cap.sample_index.dtype == np.int64 assert int(cap.sample_index[-1] - cap.sample_index[0]) == 199 def test_a_non_integer_raw_count_is_rejected(tmp_path): path = tmp_path / "bad.csv" write_capture(path, rows=100) text = path.read_text().replace(",1000,", ",not-a-number,", 1) path.write_text(text) with pytest.raises(ValueError): capture.load(str(path)) def test_a_note_containing_delimiters_survives_the_header(tmp_path): """The header is 'key: value' split on the first colon only.""" path = write_capture(tmp_path / "c.csv", header={"note": "supply: 3.0 V, run #2"}) assert capture.load(path).meta["note"] == "supply: 3.0 V, run #2" def test_an_unknown_warning_flag_is_ignored(tmp_path): """The column generalises to future flags, so today's reader must not choke.""" path = write_capture(tmp_path / "c.csv", rows=200, flags={40: "SOMETHING_NEW"}) cap = capture.load(path) assert not cap.missed.any() assert not cap.ambiguous.any() def test_a_flag_substring_does_not_count_as_the_flag(tmp_path): """Flags are whitespace-separated tokens, not substrings.""" path = write_capture(tmp_path / "c.csv", rows=200, flags={40: "NOTMISSED"}) assert not capture.load(path).missed.any()