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