Coverage for examples/OneRoom_CIA/predictor/simple_predictor.py: 100%

25 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2025-10-20 14:09 +0000

1import agentlib as al 

2import numpy as np 

3import pandas as pd 

4from agentlib.core import Agent 

5import json 

6import csv 

7from datetime import datetime 

8 

9class PredictorModuleConfig(al.BaseModuleConfig): 

10 """Module that outputs a prediction of the heat load at a specified 

11 interval.""" 

12 outputs: al.AgentVariables = [ 

13 al.AgentVariable( 

14 name="r_pel", unit="ct/kWh", type="pd.Series", description="Weight for P_el in objective function" 

15 ), 

16 ] 

17 parameters: al.AgentVariables = [ 

18 al.AgentVariable( 

19 name="time_step", value=900, description="Sampling time for prediction." 

20 ), 

21 al.AgentVariable( 

22 name="prediction_horizon", 

23 value=8, 

24 description="Number of sampling points for prediction.", 

25 ) 

26 ] 

27 

28 shared_variable_fields:list[str] = ["outputs"] 

29 

30 

31class PredictorModule(al.BaseModule): 

32 """Module that outputs a prediction of the heat load at a specified 

33 interval.""" 

34 

35 config: PredictorModuleConfig 

36 

37 def register_callbacks(self): 

38 pass 

39 

40 def process(self): 

41 while True: 

42 sample_time = self.env.config.t_sample 

43 ts = self.get("time_step").value 

44 k = self.get("prediction_horizon").value 

45 now = self.env.now 

46 

47 grid = np.arange(now, now + k * ts + 1, sample_time) 

48 p_traj = pd.Series([1 for i in grid], index=list(grid)) 

49 self.set("r_pel", p_traj) 

50 

51 yield self.env.timeout(sample_time)