145 lines
9.4 KiB
Python
145 lines
9.4 KiB
Python
import copy
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import asdict
|
|
from datetime import datetime,timedelta,timezone
|
|
from pathlib import Path
|
|
from netplan_v4 import economic_replay as e, measurement_pipeline as m
|
|
from netplan_v4.domain import Battery,Limits,Step,Price
|
|
from netplan_v4.store import PlannerStore
|
|
from netplan_v4.selection import choose_family
|
|
from test_measurement_pipeline import config,record
|
|
from test_v4 import AID
|
|
|
|
AT=datetime(2026,10,2,22,tzinfo=timezone.utc)
|
|
END=AT+timedelta(days=1)
|
|
TS=int(AT.timestamp())
|
|
|
|
|
|
def fixture():
|
|
battery=Battery('b',20.,50.,10.,90.,5000.,5000.,AT,grid_charging=True)
|
|
steps=[Step(AT+timedelta(minutes=5*i),1000.,0.,Price(.1 if i<144 else .4),Price(.03)) for i in range(288)]
|
|
data={'steps':steps,'batteries':[battery],'limits':Limits(import_w=10000.),'observed_peaks':{'2026-10':5.},'peak_prices':{'2026-10':5.},'quarter_history':{},'at':AT}
|
|
def optimizer(**kw):
|
|
points=[]
|
|
family=kw['family']
|
|
for i,s in enumerate(kw['steps']):
|
|
target=1000. if family in ('3','13') else (1600. if i<144 else 500.)
|
|
points.append({'time':s.start.isoformat(),'validUntil':s.end.isoformat(),'gridTargetW':target,
|
|
'importPriceChfKwh':s.import_price.chf_kwh,'exportPriceChfKwh':s.export_price.chf_kwh})
|
|
return {'executable':True,'points':points}
|
|
def assemble(*args):return copy.deepcopy(data),END,{}
|
|
return data,assemble,optimizer
|
|
|
|
|
|
class EconomicReplayTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.s=PlannerStore(str(Path(self.tmp.name)/'db.sqlite'));self.c=config()
|
|
m.register_dataset(self.s.con,AID,self.c,TS)
|
|
self.s.save_settings(AID,{'forecastSource':'corrected_profile','measurementDataset':self.c['datasetId'],'autoMinimumDays':1},0,AT)
|
|
self.data,self.assemble,self.optimizer=fixture()
|
|
def tearDown(self):self.s.close();self.tmp.cleanup()
|
|
def capture(self):e.capture(self.s,AID,AT,self.assemble,self.optimizer)
|
|
def cohort(self):return json.loads(self.s.con.execute('SELECT value FROM planner_economic_cohorts').fetchone()[0])
|
|
def populate(self):
|
|
now=int(END.timestamp())+180
|
|
rows=[record(t,self.c) for t in range(TS-60,now+1,30)]
|
|
for i in range(0,len(rows),120):m.ingest_batch(self.s.con,AID,{'version':1,'datasetId':self.c['datasetId'],'records':rows[i:i+120]},now)
|
|
def outcomes(self,value=1000.):return {TS+300*i:{'loadW':value,'coverage':1.,'profileUsable':True} for i in range(288)}
|
|
def test_snapshot_frozen_once(self):
|
|
self.capture();first=self.cohort();self.capture();self.assertEqual(first,self.cohort())
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],1)
|
|
def test_no_retroactive_midday_snapshot(self):
|
|
e.capture(self.s,AID,AT+timedelta(hours=12),self.assemble,self.optimizer)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],0)
|
|
def test_no_silent_legacy_replay(self):
|
|
self.s.save_settings(AID,{'forecastSource':'legacy'},1,AT);self.capture()
|
|
self.assertEqual(e.scores(self.s,AID,END),[])
|
|
def test_all_families_same_initial_energy(self):
|
|
self.capture();c=self.cohort();values=[e.simulate(c,f,self.outcomes()) for f in c['plans']]
|
|
self.assertEqual(len({v['initialEnergyKwh'] for v in values}),1)
|
|
for v in values:self.assertGreaterEqual(v['finalEnergyKwh'],2.-1e-6);self.assertLessEqual(v['finalEnergyKwh'],18.00001)
|
|
def test_profitable_time_shift_has_lower_normalized_cost(self):
|
|
self.capture();c=self.cohort();a=e.simulate(c,'3',self.outcomes());b=e.simulate(c,'23',self.outcomes())
|
|
self.assertLess(b['costChf'],a['costChf']);self.assertTrue(b['terminalNormalized'])
|
|
def test_end_energy_not_free_savings(self):
|
|
self.capture();c=self.cohort()
|
|
for p in c['plans']['3']['points']:p['gridTargetW']=0.;p['importPriceChfKwh']=.4
|
|
d=e.simulate(c,'3',self.outcomes());self.assertGreater(d['terminalAdjustmentChf'],0)
|
|
def test_missing_actual_not_zero(self):
|
|
self.capture();out=self.outcomes();out.pop(TS+300)
|
|
with self.assertRaises(ValueError):e.simulate(self.cohort(),'3',out)
|
|
def test_export_and_import_limits_not_ignored(self):
|
|
self.capture();c=self.cohort();c['context']['limits']['import_w']=500.
|
|
result=e.simulate(c,'3',self.outcomes(20000.));self.assertGreater(result['constraintBreaches'],0)
|
|
def test_peak_increment_not_total_billed_twice(self):
|
|
self.capture();c=self.cohort()
|
|
for p in c['plans']['3']['points']:p['gridTargetW']=6000.
|
|
result=e.simulate(c,'3',self.outcomes(6000.));self.assertAlmostEqual(result['peakCostChf'],5.)
|
|
def test_no_actual_results_before_day_end(self):
|
|
self.capture();self.populate();e.advance(self.s,AID,AT+timedelta(hours=12))
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_results').fetchone()[0],0)
|
|
def test_real_pipeline_from_frozen_plans_to_actuals_to_selection(self):
|
|
self.capture();self.populate();now=END+timedelta(minutes=4)
|
|
e.advance(self.s,AID,now);scores=e.scores(self.s,AID,now)
|
|
self.assertEqual(len(scores),3);self.assertEqual(scores[0].days,1)
|
|
chosen=choose_family('auto','3',scores,minimum_days=1,margin_chf=0,now=now)
|
|
self.assertEqual(chosen['family'],'23');self.assertEqual(chosen['mode'],'economic_replay')
|
|
self.assertTrue(e.status(self.s,AID)['connected']);self.assertFalse(e.status(self.s,AID)['isBillingEvidence'])
|
|
def test_evaluation_idempotent(self):
|
|
self.capture();self.populate();now=END+timedelta(minutes=4);e.advance(self.s,AID,now);e.advance(self.s,AID,now)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_results').fetchone()[0],1)
|
|
def test_missing_outcomes_preserve_cohort(self):
|
|
self.capture();e.advance(self.s,AID,END+timedelta(minutes=4))
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_results').fetchone()[0],0)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],1)
|
|
def test_changed_tariffs_between_families_reject_cohort(self):
|
|
def changed(**kwargs):
|
|
v=self.optimizer(**kwargs)
|
|
if kwargs['family']=='23':v['points'][0]['importPriceChfKwh']+=.5
|
|
return v
|
|
e.capture(self.s,AID,AT,self.assemble,changed)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],0)
|
|
def test_no_grid_charging_unsupported_is_explicit(self):
|
|
self.data['batteries'][0]=__import__('dataclasses').replace(self.data['batteries'][0],grid_charging=False)
|
|
e.capture(self.s,AID,AT,lambda *args:(self.data,END,{}),self.optimizer)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],0)
|
|
def test_short_price_horizon_no_fake_extension(self):
|
|
data=copy.deepcopy(self.data);data['steps']=data['steps'][:12]
|
|
e.capture(self.s,AID,AT,lambda *args:(data,END,{}),self.optimizer)
|
|
self.assertEqual(self.s.con.execute('SELECT COUNT(*) FROM planner_economic_cohorts').fetchone()[0],0)
|
|
def test_other_plant_no_results(self):
|
|
self.capture();self.populate();e.advance(self.s,AID,END+timedelta(minutes=4))
|
|
self.assertEqual(e.scores(self.s,'30509683-4569-49e4-848f-4905e4cc813a',END+timedelta(minutes=4)),[])
|
|
def test_residual_projection_does_not_change_load_profile(self):
|
|
rows=[m.project(record(t,self.c),self.c,AID,TS) for t in range(TS-300,TS+1,30)]
|
|
self.assertEqual(e.actuals(rows,self.c)[0]['loadW'],1000.)
|
|
self.assertEqual(m.reconstruct(rows,self.c)[0]['loadW'],6000.)
|
|
def test_monthly_peak_not_charged_again_every_comparison_day(self):
|
|
self.capture();c=self.cohort()
|
|
for family in c['plans']:
|
|
for point in c['plans'][family]['points']:point['gridTargetW']=6000.
|
|
result={f:e.simulate(c,f,self.outcomes(6000.)) for f in c['plans']}
|
|
self.assertEqual(result['3']['peakCostChf'],5.)
|
|
policy=self.s.con.execute('SELECT policy_id FROM planner_economic_cohorts').fetchone()[0]
|
|
with self.s.con:
|
|
for n in (0,1):
|
|
start=AT+timedelta(days=n);end=END+timedelta(days=n);day=start.astimezone(__import__('zoneinfo').ZoneInfo('Europe/Zurich')).date().isoformat()
|
|
if n:
|
|
self.s.con.execute('INSERT INTO planner_economic_cohorts VALUES(?,?,?,?,?,?)',(AID,day,int(start.timestamp()),int(end.timestamp()),policy,json.dumps(c)))
|
|
v={'results':result,'start':start.isoformat(),'end':end.isoformat()}
|
|
self.s.con.execute('INSERT INTO planner_economic_results VALUES(?,?,?,?,?)',(AID,day,int(end.timestamp())+120,policy,json.dumps(v)))
|
|
scores=e.scores(self.s,AID,END+timedelta(days=1,minutes=4))
|
|
value=next(x.cost_chf for x in scores if x.family=='3')
|
|
expected=2*(result['3']['energyCostChf']+result['3']['throughputCostChf']+result['3']['terminalAdjustmentChf'])+5.
|
|
self.assertAlmostEqual(value,expected)
|
|
|
|
def test_external_sdl_not_assumed_zero(self):
|
|
rows=[m.project(record(t,self.c,sdl=250.),self.c,AID,TS) for t in range(TS-300,TS+1,30)]
|
|
self.assertEqual(e.actuals(rows,self.c)[0]['loadW'],1250.)
|
|
for r in rows:r['raw']['sdl']['valid']=False
|
|
self.assertFalse(e.actuals(rows,self.c)[0]['profileUsable'])
|
|
|
|
if __name__=='__main__':unittest.main()
|