115 lines
5.8 KiB
Python
115 lines
5.8 KiB
Python
"""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()
|