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

1import agentlib as al 

2import numpy as np 

3import pandas as pd 

4 

5class PredictorModuleConfig(al.BaseModuleConfig): 

6 """Module that outputs a prediction of the ambient temp and comfort setpoint 

7 at a specified interval""" 

8 

9 outputs: al.AgentVariables = [ 

10 al.AgentVariable( 

11 name="T_amb", 

12 type="pd.Series", 

13 description="Ambient air temperature", 

14 ), 

15 al.AgentVariable( 

16 name="T_upper", 

17 type="pd.Series", 

18 description="Upper boundary (soft) for T", 

19 ), 

20 al.AgentVariable( 

21 name="T_lower", 

22 type="pd.Series", 

23 description="Lower boundary (soft) for T", 

24 ), 

25 al.AgentVariable( 

26 name="r_pel", 

27 unit="ct/kWh", 

28 type="pd.Series", 

29 description="Weight for P_el in objective function (electricity price)" 

30 ), 

31 ] 

32 

33 parameters: al.AgentVariables = [ 

34 al.AgentVariable( 

35 name="prediction_sampling_time", 

36 value=10, 

37 description="Sampling time for prediction.", 

38 ), 

39 al.AgentVariable( 

40 name="prediction_horizon", 

41 value=10, 

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

43 ), 

44 al.AgentVariable( 

45 name="sampling_time", 

46 value=900, 

47 description="Time between prediction updates.", 

48 ), 

49 al.AgentVariable( 

50 name="comfort_interval", 

51 value=43200, 

52 description="Time between comfort updates.", 

53 ), 

54 al.AgentVariable( 

55 name="upper_comfort_high", 

56 value=299, 

57 description="High value in the comfort set point trajectory.", 

58 ), 

59 al.AgentVariable( 

60 name="upper_comfort_low", 

61 value=297, 

62 description="Low value in the comfort set point trajectory.", 

63 ), 

64 al.AgentVariable( 

65 name="lower_comfort_high", 

66 value=292, 

67 description="High value in the comfort set point trajectory.", 

68 ), 

69 al.AgentVariable( 

70 name="lower_comfort_low", 

71 value=290, 

72 description="Low value in the comfort set point trajectory.", 

73 ), 

74 ] 

75 

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

77 

78class PredictorModule(al.BaseModule): 

79 """Module that outputs a prediction of the ambient temp and comfort setpoint 

80 at a specified interval""" 

81 

82 config: PredictorModuleConfig 

83 

84 def register_callbacks(self): 

85 pass 

86 

87 def process(self): 

88 """Sets a new prediction at each time step""" 

89 self.env.process(self.send_upper_comfort_trajectory()) 

90 self.env.process(self.send_lower_comfort_trajectory()) 

91 self.env.process(self.send_price_var_trajectory()) 

92 

93 while True: 

94 ts = self.get("prediction_sampling_time").value 

95 n = self.get("prediction_horizon").value 

96 now = self.env.now 

97 sample_time = self.get("sampling_time").value 

98 

99 # temperature prediction  

100 grid = np.arange(now, now + n * ts, ts) 

101 values = amb_temp_func(grid, uncertainty=0) 

102 traj = pd.Series(values, index=list(grid)) 

103 self.set("T_amb", traj) 

104 yield self.env.timeout(sample_time) 

105 

106 def send_upper_comfort_trajectory(self): 

107 """Sends the series for the comfort condition""" 

108 while True: 

109 now = self.env.now 

110 comfort_interval = self.get("comfort_interval").value 

111 

112 # temperature prediction 

113 grid = np.arange(now, now + 2 * comfort_interval, 0.5 * comfort_interval) 

114 values = [self.get("upper_comfort_high").value, self.get("upper_comfort_low").value] * 2 

115 traj = pd.Series(values, index=list(grid)) 

116 self.set("T_upper", traj) 

117 yield self.env.timeout(comfort_interval) 

118 

119 def send_lower_comfort_trajectory(self): 

120 """Sends the series for the comfort condition""" 

121 while True: 

122 now = self.env.now 

123 comfort_interval = self.get("comfort_interval").value 

124 

125 # temperature prediction  

126 grid = np.arange(now, now + 2 * comfort_interval, 0.5 * comfort_interval) 

127 values = [self.get("lower_comfort_low").value, self.get("lower_comfort_high").value] * 2 

128 traj = pd.Series(values, index=list(grid)) 

129 self.set("T_lower", traj) 

130 yield self.env.timeout(comfort_interval) 

131 

132 def send_price_var_trajectory(self): 

133 """Sends the series for the price variable""" 

134 while True: 

135 ts = self.get("prediction_sampling_time").value 

136 n = self.get("prediction_horizon").value 

137 now = self.env.now 

138 sample_time = self.get("sampling_time").value 

139 

140 grid = np.arange(now, now + n * ts, ts) 

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

142 self.set("r_pel", traj) 

143 yield self.env.timeout(sample_time) 

144 

145def amb_temp_func(current, uncertainty): 

146 """Returns the ambient temperature in K, given a time in seconds""" 

147 value = np.zeros(shape=current.shape) 

148 for i in range(current.size): 

149 random_factor = 1 + uncertainty * (np.random.random() - 0.5) 

150 value[i] = random_factor * (278.15 + 5 * np.sin(2*np.pi * current[i] / 86400)) 

151 return value