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

26 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2025-08-01 15:10 +0000

1import agentlib as al 

2import numpy as np 

3import pandas as pd 

4from agentlib.core import Agent 

5from typing import List 

6import json 

7import csv 

8from datetime import datetime 

9 

10class PredictorModuleConfig(al.BaseModuleConfig): 

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

12 interval.""" 

13 outputs: al.AgentVariables = [ 

14 al.AgentVariable( 

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

16 ), 

17 ] 

18 parameters: al.AgentVariables = [ 

19 al.AgentVariable( 

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

21 ), 

22 al.AgentVariable( 

23 name="prediction_horizon", 

24 value=8, 

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

26 ) 

27 ] 

28 

29 

30 shared_variable_fields:List[str] = ["outputs"] 

31 

32 

33class PredictorModule(al.BaseModule): 

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

35 interval.""" 

36 

37 config: PredictorModuleConfig 

38 

39 def register_callbacks(self): 

40 pass 

41 

42 def process(self): 

43 while True: 

44 sample_time = self.env.config.t_sample 

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

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

47 now = self.env.now 

48 

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

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

51 self.set("r_pel", p_traj) 

52 

53 yield self.env.timeout(sample_time)