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