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
« 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
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 ]
29 shared_variable_fields:list[str] = ["outputs"]
32class PredictorModule(al.BaseModule):
33 """Module that outputs a prediction of the heat load at a specified
34 interval."""
36 config: PredictorModuleConfig
38 def register_callbacks(self):
39 pass
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
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)
52 yield self.env.timeout(sample_time)