File size: 2,979 Bytes
1e1dffa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | """Small offline checks; never import the weather model or run inference."""
import ast
from pathlib import Path
import unittest
import numpy as np
from deepmind_daily_benchmark import aggregate, curve_similarity, local, pressure_mask, route_metrics
class DeepMindBenchmarkTest(unittest.TestCase):
def test_identical_routes_have_zero_error_and_matching_shape(self):
truth = np.stack([np.arange(1, 21)*3, np.arange(1, 21)*4], -1)
metrics = route_metrics({k: truth.copy() for k in ('1.1', '1.2', 'deepmind')}, truth)
for result in metrics.values():
self.assertAlmostEqual(result['mean_track_error_km'], 0)
self.assertAlmostEqual(result['direction_error_deg'], 0)
self.assertAlmostEqual(result['shape_similarity'], 1)
def test_direction_mask_is_common_to_all_three_models(self):
path = np.stack([np.arange(1, 21)*10, np.zeros(20)], -1)
stationary = np.zeros((20, 2))
results = route_metrics({'1.1': path, '1.2': path, 'deepmind': stationary}, path)
self.assertTrue(all(r['direction_valid_steps'] == 0 and r['direction_error_deg'] is None for r in results.values()))
def test_curve_shape_is_not_pressure_level_error(self):
truth = np.arange(20, dtype=float)+950
self.assertAlmostEqual(curve_similarity(truth+20, truth), 1)
self.assertIsNone(curve_similarity(np.ones(20), truth))
self.assertIsNone(curve_similarity(truth[:5], truth[:5]))
def test_missing_and_invalid_pressure_never_become_zero_errors(self):
truth = np.full(20, 950.)
one, two, candidate = truth.copy(), truth.copy(), truth.copy()
one[0], two[1], candidate[2] = np.nan, 0, 700
mask = pressure_mask(truth, {'1.1': one, '1.2': two, 'deepmind': candidate})
self.assertEqual(mask.sum(), 17)
self.assertFalse(mask[:3].any())
with self.assertRaises(ValueError):
pressure_mask(truth, {'model': truth[:19]})
def test_longitude_wrap_keeps_geographic_alignment(self):
result = local(np.array([[0., 1.]]), 0., 359.)
self.assertAlmostEqual(result[0, 0], 222.4)
def test_empty_coverage_is_null_not_a_zero_score(self):
results = aggregate({})
self.assertIsNone(results['route']['deepmind']['mean_track_error_km']['value'])
self.assertIsNone(results['pressure']['JMA']['models']['deepmind']['mae_hpa']['value'])
def test_worker_never_reads_scoring_labels(self):
source = Path(__file__).with_name('deepmind_daily_benchmark.py').read_text()
worker = next(n for n in ast.parse(source).body if isinstance(n, ast.FunctionDef) and n.name == 'worker')
segment = ast.get_source_segment(source, worker)
self.assertNotIn('truth_local', segment)
self.assertNotIn('intensity-v12-v11', segment)
self.assertNotIn('np.load', segment)
self.assertIn('targets*np.nan', segment)
if __name__ == '__main__':
unittest.main()
|