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

25 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2025-09-19 15:08 +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 

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

30 

31 

32class PredictorModule(al.BaseModule): 

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

34 interval.""" 

35 

36 config: PredictorModuleConfig 

37 

38 def register_callbacks(self): 

39 pass 

40 

41 def process(self): 

42 while True: 

43 sample_time = self.env.config.t_sample 

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

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

46 now = self.env.now 

47 

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

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

50 self.set("r_pel", p_traj) 

51 

52 yield self.env.timeout(sample_time)