from datetime import datetime,timedelta,timezone import json import unittest from netplan_v4.metering import CounterSeries,Reading,MeterEvidence,audit_capture,native_meter_id UTC=timezone.utc # October begins at September 30 22:00 UTC in Europe/Zurich. BEGIN=datetime(2026,9,30,22,0,tzinfo=UTC) END=BEGIN+timedelta(minutes=15) SOURCES=[{'VariableID':59607,'ElternID':11490,'Ident':'Energy_0','FaktorZuKWh':1.,'Messgroesse':'WirkenergieBezug'}, {'VariableID':26620,'ElternID':11490,'Ident':'Energy_1','FaktorZuKWh':1.,'Messgroesse':'WirkenergieBezug'}] def stream(name,pairs,verified=False,**kwargs): return CounterSeries(name,[Reading(BEGIN+timedelta(seconds=t),v) for t,v in pairs], timestamp_verified=verified,identity_verified=verified,chronology_verified=verified,**kwargs) def meter(t1=None,t2=None,verified=False): return MeterEvidence([stream('symcon:59607',t1 or [(0,100.),(900,102.)],verified), stream('symcon:26620',t2 or [(0,0.),(900,0.)],verified)],native_sources=SOURCES) class MeterEvidenceTest(unittest.TestCase): def test_t1_t2_sum_and_average(self): result=meter(t2=[(0,10.),(900,11.)],verified=True).interval(BEGIN,END) self.assertAlmostEqual(result['lowerKwh'],3.) self.assertAlmostEqual(result['upperAverageKw'],12.) self.assertTrue(result['billingEvidence']) def test_known_zero_is_valid_not_missing(self): self.assertEqual(meter(verified=True).strict_contract(END)['measuredPeaks']['2026-10']['kw'],8.) def test_missing_tariff_not_assumed_zero(self): m=MeterEvidence([stream('a',[(0,1.),(900,2.)]),stream('b',[])]) self.assertEqual(m.interval(BEGIN,END)['status'],'missing') def test_current_zero_does_not_prove_past(self): m=MeterEvidence([stream('a',[(0,1.),(900,2.)]),stream('b',[(900,0.)])]) self.assertFalse(m.month(END)['historyComplete']) def test_missing_month_peak_not_zero(self): result=meter(t2=[(900,0.)]).month(END) self.assertIsNone(result['verifiedMonthPeakKw']) self.assertIsNone(result['monthPeakUpperKw']) def test_first_quarter_not_already_paid_peak(self): result=meter(verified=True).month(BEGIN+timedelta(minutes=5)) self.assertEqual(result['completedQuarters'],0) self.assertFalse(result['billingEvidence']) self.assertIsNone(result['verifiedMonthPeakKw']) def test_reset_quarantines_interval(self): result=meter(t1=[(0,100.),(400,0.),(900,2.)]).interval(BEGIN,END) self.assertEqual(result['status'],'missing') def test_unverified_archive_never_becomes_billing_evidence(self): m=meter() self.assertTrue(m.month(END)['historyComplete']) with self.assertRaises(ValueError):m.strict_contract(END) def test_boundaries_not_interpolated_into_fact(self): s=stream('a',[(-2,100.),(2,100.1),(898,101.9),(902,102.)],verified=True) e=MeterEvidence([s]).interval(BEGIN,END) self.assertEqual(e['status'],'bounded_records') self.assertAlmostEqual(e['lowerKwh'],1.8) self.assertAlmostEqual(e['upperKwh'],2.) self.assertFalse(e['billingEvidence']) def test_wide_boundary_gap_is_unknown(self): s=stream('a',[(-200,100.),(200,100.1),(900,101.)]) self.assertIsNone(s.energy(BEGIN,END)) def test_no_extrapolation(self): self.assertIsNone(stream('a',[(0,1.),(900,2.)]).energy(BEGIN,END+timedelta(seconds=1))) def test_duplicate_identical_records_are_idempotent(self): self.assertEqual(len(stream('a',[(0,1.),(0,1.),(900,2.)]).times),2) def test_duplicate_conflicting_records_fail(self): with self.assertRaises(ValueError):stream('a',[(0,1.),(0,1.1)]) def test_invalid_counter_values_fail(self): for value in (float('nan'),float('inf'),True,-1.,'2'): with self.subTest(value=value),self.assertRaises(ValueError):stream('a',[(0,value)]) def test_duplicate_tariffs_fail(self): with self.assertRaises(ValueError):MeterEvidence([stream('a',[]),stream('a',[])]) def test_native_identity_ignores_order_not_factor(self): self.assertEqual(native_meter_id(SOURCES),native_meter_id(list(reversed(SOURCES)))) changed=[dict(s) for s in SOURCES];changed[0]['FaktorZuKWh']=.001 self.assertNotEqual(native_meter_id(SOURCES),native_meter_id(changed)) def test_identity_matches_php_encoder(self): # Captured from PHP 8.3 on the test host, not calculated by this Python code. cases={1.:'29f33f47e5c3a312e9c3c85735f95ea16d237860d76c0e3720c59e567d77ec8c', .001:'5600bff9a5d82aa6230196f90f836c69c966374a1458864e770083656ed65f4c', .000001:'477aca4ce6eb1079b58114911125675137a5f7d0ec8b0b3a2025e3323cb269a4'} for factor,digest in cases.items(): sources=[dict(s) for s in SOURCES];sources[0]['FaktorZuKWh']=factor with self.subTest(factor=factor):self.assertEqual(native_meter_id(sources),'symcon-active-import:'+digest) def test_native_source_set_must_match(self): with self.assertRaises(ValueError):MeterEvidence([stream('symcon:53476',[])],native_sources=SOURCES) def test_no_legacy_evidence_identity(self): m=MeterEvidence([stream('a',[(0,1.),(900,2.)],verified=True)]) self.assertTrue(m.meter_id.startswith('audit-only:')) with self.assertRaises(ValueError):m.strict_contract(END) def test_native_other_meter_or_quantity_rejected(self): for field,value in [('ElternID',1),('Messgroesse','Blindenergie')]: sources=[dict(s) for s in SOURCES];sources[0][field]=value with self.subTest(field=field),self.assertRaises(ValueError):native_meter_id(sources) def test_partial_quarter_must_be_verified(self): at=END+timedelta(minutes=5) m=meter(t1=[(0,100.),(900,102.),(1200,103.)],t2=[(0,0.),(900,0.),(1200,0.)],verified=True) contract=m.strict_contract(at) self.assertEqual(contract['quarterPast']['measuredSeconds'],300) self.assertEqual(contract['quarterPast']['importKwh'],1.) def test_partial_quarter_not_extrapolated(self): with self.assertRaises(ValueError):meter(verified=True).strict_contract(END+timedelta(seconds=1)) def test_zurich_month_not_utc_month(self): self.assertEqual(meter().month(END)['month'],'2026-10') def test_dst_repeated_hour_not_merged(self): from zoneinfo import ZoneInfo zone=ZoneInfo('Europe/Zurich') a=datetime(2026,10,25,2,0,tzinfo=zone,fold=0) b=datetime(2026,10,25,2,0,tzinfo=zone,fold=1) s=CounterSeries('a',[Reading(a,1.),Reading(b,2.)]) self.assertEqual(len(s.times),2) self.assertEqual((s.times[1]-s.times[0]).total_seconds(),3600) def test_naive_timestamps_rejected(self): with self.assertRaises(ValueError):CounterSeries('a',[Reading(datetime(2026,10,1),1.)]) def test_capture_is_only_an_audit(self): channels={} for var,ident,value in [('59607','Energy_0',2.),('26620','Energy_1',0.)]: channels[var]={'ident':ident,'logging':True,'queryComplete':True, 'history':[{'TimeStamp':int(BEGIN.timestamp()),'Value':0.}, {'TimeStamp':int(END.timestamp()),'Value':value}], 'snapshotConsistent':True,'snapshot':{'capturedAt':END.isoformat(),'value':value}} result=audit_capture({'schemaVersion':1,'kind':'v4_meter_capture', 'capturedAt':END.isoformat(),'channels':channels}) self.assertTrue(result['month']['historyComplete']) self.assertFalse(result['billingEvidence']) self.assertFalse(result['liveEnabled']) def test_missing_logging_survives_without_fake_1970(self): channels={str(var):{'ident':ident,'logging':False,'queryComplete':False, 'history':[],'snapshotConsistent':True,'snapshot':{'capturedAt':END.isoformat(),'value':0.}} for var,ident in [(59607,'Energy_0'),(26620,'Energy_1')]} result=audit_capture({'schemaVersion':1,'kind':'v4_meter_capture','capturedAt':END.isoformat(),'channels':channels}) self.assertFalse(result['month']['historyComplete']) self.assertNotIn('1970-',json.dumps(result)) if __name__=='__main__':unittest.main()