"""Numerical and source-integrity checks for the curated field analysis.""" import unittest import numpy as np import pandas as pd import analyze_2026_11_03 as analysis def synthetic(times, currents=None, uptimes=None): n = len(times) return pd.DataFrame({ "timestamp": pd.to_datetime(times), "c_raw": currents if currents is not None else [1000000] * n, "v_raw": [4500000] * n, "p_raw": [800000] * n, "uptime_s": uptimes if uptimes is not None else list(range(n)), "x_steps": list(range(n)), "y_steps": list(range(n)), "er": [900] * n, "source_line": list(range(1, n + 1)), }) class NumericalTests(unittest.TestCase): def test_signed_24_bit_edges(self): np.testing.assert_array_equal(analysis.sign24([0, 8388607, 8388608, 16777215, -1]), [0, 8388607, -8388608, -1, -1]) def test_piecewise_shunt_doubles_current_not_voltage_or_irradiance(self): frame = analysis.calibrate(synthetic(["2026-11-03 12:38:07", "2026-11-03 12:46:01"])) self.assertAlmostEqual(frame.isc_A.iloc[1], 2 * frame.isc_A.iloc[0]) self.assertAlmostEqual(frame.voc_V.iloc[1], frame.voc_V.iloc[0]) self.assertAlmostEqual(frame.irradiance_W_m2.iloc[1], frame.irradiance_W_m2.iloc[0]) def test_saturation_is_not_reconstructed(self): frame = analysis.calibrate(synthetic(["2026-11-03 12:00:00"], [8388607])) self.assertTrue(frame.adc_clipped.iloc[0]) self.assertFalse(frame.electrical_valid.iloc[0]) self.assertTrue(np.isnan(frame.isc_A.iloc[0])) self.assertAlmostEqual(frame.isc_observed_A.iloc[0], 2.048 / 20 / 0.260) def test_near_rail_guard(self): frame = analysis.calibrate(synthetic(["2026-11-03 12:00:00"], [int(0.995 * 8388607)])) self.assertFalse(frame.adc_clipped.iloc[0]) self.assertTrue(frame.near_current_rail.iloc[0]) self.assertFalse(frame.electrical_valid.iloc[0]) def test_setup_and_uncertain_change_excluded(self): frame = analysis.calibrate(synthetic(["2026-11-03 10:50:00", "2026-11-03 12:42:00"])) self.assertFalse(frame.electrical_valid.any()) self.assertTrue(frame.setup.iloc[0]) self.assertTrue(frame.transition_uncertain.iloc[1]) def test_ff_power_and_aperture_formula(self): frame = analysis.calibrate(synthetic(["2026-11-03 13:00:00"])) current = 1000000 / 8388607 * 2.048 / 20 / 0.130 voltage = 4500000 / 8388607 * 2.048 * 2 irradiance = 800000 / 8388607 * 2048 / 0.2 self.assertAlmostEqual(frame.power_FF_W.iloc[0], current * voltage * 0.82) self.assertAlmostEqual(frame.aperture_efficiency_pct.iloc[0], 100 * current * voltage * 0.82 / (0.002497 * irradiance)) def test_rejected_row_never_bridged(self): frame = analysis.calibrate(synthetic(pd.date_range("2026-11-03 13:00", periods=5, freq="5s"))) edges = analysis.interval_mask(frame, [True, True, False, True, True]) np.testing.assert_array_equal(edges, [True, False, False, True]) self.assertAlmostEqual(analysis.integrate(frame, np.ones(5), edges), 10 / 3600) def test_long_gap_and_reboot_excluded(self): frame = analysis.calibrate(synthetic( ["2026-11-03 13:00:00", "2026-11-03 13:00:05", "2026-11-03 13:01:00", "2026-11-03 13:01:05"], uptimes=[10, 15, 70, 1])) np.testing.assert_array_equal(analysis.interval_mask(frame, [True] * 4), [True, False, False]) def test_shunt_boundary_never_integrated(self): frame = analysis.calibrate(synthetic(["2026-11-03 12:45:58", "2026-11-03 12:46:01"])) self.assertFalse(analysis.interval_mask(frame, [True, True]).any()) def test_exact_trapezoidal_energy(self): frame = analysis.calibrate(synthetic(pd.date_range("2026-11-03 13:00", periods=3, freq="5s"))) self.assertAlmostEqual(analysis.integrate(frame, [0, 2, 0], [True, True]), 10 / 3600) class SourceTests(unittest.TestCase): @classmethod def setUpClass(cls): raw, cls.telemetry, cls.events = analysis.parse_log(analysis.SOURCE) cls.frame = analysis.calibrate(raw) def test_raw_record_counts_and_saturation(self): self.assertEqual(len(self.frame), 3429) self.assertEqual(int(self.frame.adc_clipped.sum()), 334) self.assertEqual(int(self.frame.near_current_rail.sum()), 358) self.assertFalse(self.frame.loc[self.frame.timestamp >= analysis.CHANGE_AFTER, "adc_clipped"].any()) def test_curated_rows_preserve_raw_observations(self): curated = pd.read_csv(analysis.SOURCE.with_name("log_2026-11-03_curated.csv")) for column in ("c_raw", "v_raw", "p_raw", "x_steps", "y_steps", "source_line"): np.testing.assert_array_equal(curated[column], self.frame[column]) self.assertTrue(curated.loc[~curated.electrical_valid, "power_FF_W"].isna().all()) self.assertEqual(int(curated.integrate_from_previous.sum()), 2876) def test_primary_metrics_and_late_failure_are_retained(self): frame, metrics, _, _ = analysis.calculate(self.frame, self.telemetry, self.events) post = metrics["post_change"] self.assertEqual(post["accepted_illuminated_samples"], 2085) self.assertAlmostEqual(post["integrated_duration_h"], 2.89) self.assertAlmostEqual(post["energy_weighted_aperture_efficiency_pct"], 28.3440609, places=5) self.assertAlmostEqual(post["empirical_tracking_efficiency_pct"], 84.9719272, places=5) self.assertTrue(frame.loc[frame.timestamp > pd.Timestamp("2026-11-03 15:00"), "electrical_valid"].all()) self.assertLess(metrics["post_change"]["responsivity_on_target_sample_pct"], metrics["fixed_hour_13_14"]["responsivity_on_target_sample_pct"]) if __name__ == "__main__": unittest.main()