| import copy |
| import unittest |
|
|
| import numpy as np |
|
|
| from forecast_historical_video50_mac import requested_cases, exact_history_indices |
| from storm_structure_diagnostics import ( |
| initialize, radius_km, envelope, outer_radius, diagnose, summarize, |
| ambient_pressure, official_radius_mae, THRESHOLDS, QUADRANTS, METRICS, |
| ) |
|
|
|
|
| class StormStructureTests(unittest.TestCase): |
| def row(self): |
| row = dict(SID='2015211N13162', ISO_TIME='2015-08-05 00:00:00', |
| USA_WIND='100', TOKYO_PRES='950', USA_RMW='20', |
| LAT='20', LON='135', USA_LAT='20', USA_LON='135', |
| USA_AGENCY='jtwc_wp', TRACK_TYPE='main') |
| for t, r in ((34, 140), (50, 80), (64, 50)): |
| for i, q in enumerate(QUADRANTS): |
| row[f'USA_R{t}_{q}'] = str(r+i*5) |
| return row |
|
|
| def initial(self, row=None): |
| return initialize(row or self.row(), '2015-08-05T00:00:00Z', 1010) |
|
|
| def member(self, cp=950, initial=None): |
| return diagnose(initial or self.initial(), cp, {'value_hpa':1010, 'status':'available'}) |
|
|
| def test_nautical_miles_are_radius_not_diameter(self): |
| self.assertAlmostEqual(radius_km({'USA_RMW':'12'}, 'USA_RMW', rmw=True), 22.224) |
| self.assertEqual(radius_km({'USA_R34_NE':'0'}, 'USA_R34_NE'), 0) |
| self.assertIsNone(radius_km({'USA_R34_NE':''}, 'USA_R34_NE')) |
| self.assertIsNone(radius_km({'USA_RMW':'0'}, 'USA_RMW', rmw=True)) |
|
|
| def test_exact_issue_only_and_no_future_influence(self): |
| before = self.initial() |
| row = self.row() |
| row.update(future_wind='999', future_r34='1', USA_EYE='700', TOKYO_R30_LONG='900') |
| self.assertEqual(before, self.initial(row)) |
| row['ISO_TIME'] = '2015-08-05 06:00:00' |
| with self.assertRaises(ValueError): |
| self.initial(row) |
|
|
| def test_native_quadrant_isotachs_initialized_without_rms_or_period_conversion(self): |
| initial = self.initial() |
| for q in QUADRANTS: |
| points, warnings = envelope(initial, q) |
| self.assertEqual(warnings, []) |
| for t in THRESHOLDS: |
| self.assertAlmostEqual(outer_radius(points, initial['holland_inspired_tail_b'], t), |
| initial['native_quadrant_radii_km'][f'r{t}_{q}']) |
| result = self.member() |
| self.assertNotEqual(result['r34_NE_estimate_km'], result['r34_SW_estimate_km']) |
| self.assertEqual(initial['definitions']['target_wind_averaging_seconds'], 60) |
| self.assertIsNone(initial['definitions']['validated_wind_averaging_seconds']) |
|
|
| def test_pressure_drives_future_and_does_not_change_rmw_persistence(self): |
| initial = self.initial() |
| before = copy.deepcopy(initial) |
| weak, strong = self.member(980, initial), self.member(900, initial) |
| self.assertGreater(strong['wind_estimate_kt'], weak['wind_estimate_kt']) |
| self.assertGreater(strong['r34_NE_estimate_km'], weak['r34_NE_estimate_km']) |
| self.assertEqual(strong['rmw_persistence_km'], weak['rmw_persistence_km']) |
| self.assertEqual(initial, before) |
|
|
| def test_missing_rmw_does_not_block_wind_or_invent_a_radius(self): |
| row = self.row() |
| row['USA_RMW'] = '' |
| result = self.member(initial=self.initial(row)) |
| self.assertIsNotNone(result['wind_estimate_kt']) |
| self.assertIsNone(result['rmw_persistence_km']) |
| self.assertTrue(all(result[f'r{t}_{q}_estimate_km'] is None for t in THRESHOLDS for q in QUADRANTS)) |
|
|
| def test_below_threshold_is_candidate_zero_not_missing_reference(self): |
| result = self.member(1005) |
| self.assertLess(result['wind_estimate_kt'], 34) |
| self.assertEqual(result['r34_NE_estimate_km'], 0) |
| self.assertIsNone(outer_radius([(30, 100)], 1, 1)) |
|
|
| def test_missing_zero_or_inconsistent_initial_quadrant_is_visible(self): |
| row = self.row() |
| row['USA_R34_NE'] = '0' |
| result = self.member(initial=self.initial(row)) |
| self.assertIsNone(result['r34_NE_estimate_km']) |
| self.assertIsNotNone(result['r34_SE_estimate_km']) |
| self.assertIn('reported_zero', result['warnings'][0]) |
| row['USA_R34_NE'] = '1' |
| points, warnings = envelope(self.initial(row), 'NE') |
| self.assertTrue(any('inconsistent' in w for w in warnings)) |
| self.assertTrue(points) |
|
|
| def test_missing_pressure_invalid_tracks_do_not_create_wind(self): |
| initial = self.initial() |
| for cp, ambient, valid in ((np.nan, {'value_hpa':1010, 'status':'available'}, True), |
| (950, {'value_hpa':None, 'status':'insufficient_ambient_coverage'}, True), |
| (950, {'value_hpa':1010, 'status':'available'}, False)): |
| result = diagnose(initial, cp, ambient, track_valid=valid) |
| self.assertTrue(all(result[k] is None for k in METRICS)) |
|
|
| def test_true_member_mean_noncommutativity_and_partial_members(self): |
| a, b = self.member(980), self.member(900) |
| result = summarize([a,b], expected_members=2)['estimates']['wind_estimate_kt'] |
| self.assertAlmostEqual(result['mean'], (a['wind_estimate_kt']+b['wind_estimate_kt'])/2) |
| self.assertNotAlmostEqual(result['mean'], self.member(940)['wind_estimate_kt'], places=3) |
| broken = dict(a, wind_estimate_kt=None) |
| result = summarize([a]*49+[broken], expected_members=50)['estimates']['wind_estimate_kt'] |
| self.assertIsNone(result['mean']) |
| self.assertEqual(result['valid_members'],49) |
| with self.assertRaises(ValueError): |
| summarize([a,b], expected_members=50) |
|
|
| def test_ambient_uses_physical_grid_cells_and_fails_closed_at_domain(self): |
| lat, lon = np.linspace(60,0,25), np.linspace(100,180,33) |
| p = np.full((25,33),1010.) |
| before = p.copy() |
| result = ambient_pressure(p,lat,lon,[20,135]) |
| self.assertEqual(result['value_hpa'],1010) |
| self.assertGreaterEqual(min(result['quadrant_cells'].values()),2) |
| self.assertIsNone(ambient_pressure(p,lat,lon,[20,181])['value_hpa']) |
| self.assertIsNone(ambient_pressure(p*np.nan,lat,lon,[20,135])['value_hpa']) |
| np.testing.assert_array_equal(p,before) |
|
|
| def test_strict_skill_gate_not_bypassed_by_matching_units_or_targets(self): |
| with self.assertRaisesRegex(ValueError,'calibration'): |
| official_radius_mae(self.member(),self.row()) |
|
|
| def test_generic_forecast_case_validation_and_causal_exact_history(self): |
| cases=requested_cases(None,[['nari','2001248N22135','2001-09-08T00:00:00Z']]) |
| self.assertEqual(len(cases),1) |
| for slug, sid, issue in (('../bad','2001248N22135','2001-09-08T00:00:00Z'), |
| ('nari','wrong','2001-09-08T00:00:00Z'), |
| ('nari','2001248N22135','2001-09-08T03:00:00Z')): |
| with self.assertRaises(ValueError): |
| requested_cases(None,[[slug,sid,issue]]) |
| hour=3600*10**9 |
| end=np.datetime64('2001-09-08T00:00:00','ns').astype('int64') |
| times=end+np.arange(-8,2)*6*hour |
| indices=exact_history_indices(times,'2001-09-08T00:00:00Z') |
| self.assertEqual(int(times[indices[-1]]),end) |
| with self.assertRaises(ValueError): |
| exact_history_indices(np.delete(times,4),'2001-09-08T00:00:00Z') |
|
|
|
|
| if __name__=='__main__': |
| unittest.main() |
|
|
