"""Offline temporal/identity/coverage checks; no model or network imports.""" import hashlib import json import unittest from build_deepmind_period_comparison import FOLDER, MODELS, ROOT, period_key, summarize from import_deepmind_release_results import same from plot_model_announcement import metric_available, period_metrics class DeepMindPeriodComparisonTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.report = json.loads((FOLDER / 'period_comparison.json').read_text()) cls.protocol = json.loads((FOLDER / 'protocol.json').read_text()) cls.original = json.loads((FOLDER / 'benchmark.json').read_text()) def test_strict_utc_boundary_not_2024_onward(self): self.assertEqual(period_key('2023-12-31T18:00:00Z'), 'before_2024') self.assertEqual(period_key('2024-01-01T00:00:00Z'), 'year_2024') self.assertEqual(period_key('2024-12-31T18:00:00Z'), 'year_2024') self.assertEqual(period_key('2025-01-01T00:00:00Z'), 'after_2024') with self.assertRaises(ValueError): period_key('2024-12-31T23:00:00+08:00') def test_counts_and_proportions_are_exact_partition(self): expected = {'before_2024': (1336, 230), 'year_2024': (76, 22), 'after_2024': (61, 18)} all_ids = [] by_id = {c['case_index']: c for c in self.protocol['cases']} for key, (days, storms) in expected.items(): period = self.report['periods'][key] self.assertEqual(period['daily_issues'], days) self.assertEqual(period['storms'], storms) self.assertAlmostEqual(period['daily_issue_percentage'], days / 1473 * 100) self.assertAlmostEqual(period['storm_percentage'], storms / 270 * 100) self.assertEqual(len(period['case_indices']), days) self.assertEqual(len({by_id[i]['storm_id'] for i in period['case_indices']}), storms) for index in period['case_indices']: self.assertEqual(period_key(by_id[index]['issue_time_utc']), key) all_ids.extend(period['case_indices']) self.assertEqual(sorted(all_ids), list(range(1473))) def test_total_preserves_all_original_scores(self): same(self.report['periods']['total']['scores'], self.original['aggregate_equal_complete_storm'], 'original totals') def test_pressure_coverage_never_becomes_track_coverage(self): total = self.report['periods']['total'] post = self.report['periods']['after_2024'] self.assertEqual(total['scores']['pressure']['JMA']['daily_cases'], 134) self.assertEqual(total['scores']['pressure']['JMA']['common_lead_points'], 2637) self.assertEqual(post['scores']['pressure']['JMA']['daily_cases'], 59) self.assertEqual(post['scores']['pressure']['JMA']['common_lead_points'], 1174) self.assertEqual(period_metrics(post)['mean_track_error_km']['coverage']['daily_issues'], 61) self.assertEqual(period_metrics(post)['pressure_JMA_hpa']['coverage']['daily_issues'], 59) missing = self.report['periods']['before_2024']['scores']['pressure']['JMA'] for model in MODELS: self.assertIsNone(missing['models'][model]['mae_hpa']['value']) self.assertIsNone(period_metrics(self.report['periods']['before_2024'])['pressure_JMA_hpa']['values']['deepmind']) def test_empty_coverage_is_null_not_zero(self): empty = summarize([], 1473, 270) self.assertIsNone(empty['scores']['route']['deepmind']['mean_track_error_km']['value']) self.assertIsNone(empty['scores']['pressure']['JMA']['models']['deepmind']['mae_hpa']['value']) def test_equal_storm_not_pooled_daily_average(self): def record(storm, index, score): values = dict(mean_track_error_km=score, track_error_120h_km=score, direction_error_deg=score, shape_similarity=.5, track_error_by_lead_km=[score] * 20) return {'storm_id': storm, 'case_index': index, 'issue_time_utc': '2025-01-01T00:00:00Z', 'route_metrics': {m: values for m in MODELS}, 'pressure_metrics': {}} rows = [record('a', 0, 10), record('a', 1, 10), record('a', 2, 10), record('b', 3, 30)] self.assertEqual(summarize(rows, 4, 2)['scores']['route']['deepmind']['mean_track_error_km']['value'], 20) def test_same_checkpoint_and_immutable_source_hashes(self): self.assertTrue(self.report['single_checkpoint_across_all_periods']) self.assertEqual(self.report['model']['model_checkpoint'], 'WeatherNextCyclones_Mini_<2024') self.assertEqual(self.report['model']['trained_through'], 2023) self.assertEqual(self.report['model']['software_version'], '0.3.0') for name in ('protocol.json', 'verification.json', 'case-manifest.json', 'benchmark.json'): self.assertEqual(hashlib.sha256((FOLDER / name).read_bytes()).hexdigest(), self.report['source_hashes'][name]) for key in ('new_inference_performed', 'model_weights_changed', 'forecast_arrays_modified', 'raw_inputs_reaudited'): self.assertFalse(self.report[key]) self.assertEqual(self.report['missing_results_scored_as_zero'], 0) def test_chart_receipts_match_exact_data_and_bytes(self): for name, key in (('model_1_2_benchmark', 'total'), ('model_1_2_before_2024_benchmark', 'before_2024'), ('model_1_2_after_2024_benchmark', 'after_2024')): base = ROOT / 'evaluation/released_daily' / name receipt = json.loads(base.with_suffix('.json').read_text()) self.assertEqual(receipt['period'], key) self.assertEqual(receipt['deepmind_model'], self.report['model']) self.assertEqual(receipt['source_sha256'], hashlib.sha256((FOLDER / 'period_comparison.json').read_bytes()).hexdigest()) self.assertEqual(receipt['chart_sha256'], hashlib.sha256(base.with_suffix('.png').read_bytes()).hexdigest()) expected = period_metrics(self.report['periods'][key]) self.assertEqual(receipt['values'], {k: v['values'] for k, v in expected.items()}) self.assertFalse(receipt['zero_fill']) self.assertEqual(receipt['available_bars_per_panel'], {k: 3 if metric_available(v) else 0 for k, v in expected.items()}) self.assertEqual(receipt['coverage'], {k: v['coverage'] for k, v in expected.items()}) if key == 'before_2024': self.assertIsNone(receipt['bars_per_panel']) self.assertEqual(receipt['unavailable_metrics'], ['pressure_JMA_hpa']) def test_unavailable_panel_is_not_zero_fill_or_invalid_numeric_score(self): empty = {'values': {m: None for m in MODELS}, 'coverage': {'daily_issues': 0, 'storms': 0}} self.assertFalse(metric_available(empty)) for values, coverage in (({m: 0 for m in MODELS}, empty['coverage']), (empty['values'], {'daily_issues': 1, 'storms': 1}), ({m: float('nan') for m in MODELS}, {'daily_issues': 1, 'storms': 1})): with self.assertRaises(ValueError): metric_available({'values': values, 'coverage': coverage}) if __name__ == '__main__': unittest.main()