1"""
2This module contains a class for modeling pump speed action spaces.
3"""
4from typing import Optional
5from epyt_flow.gym import ScenarioControlEnv
6from gymnasium.spaces import Space, Box
7
8from .actions import Action
9
10
[docs]
11class PumpSpeedAction(Action):
12 """
13 Action for controlling the speed of a pump.
14
15 Parameters
16 ----------
17 pump_id : `str`
18 ID of the pump.
19 speed_lower_bound : `float`, optional
20 Lower bound on the pump speed.
21
22 The default is zero.
23 speed_upper_bound : `float`, optional
24 Upper bound on the pump speed.
25 If None, no upper bound restriction for the agent.
26
27 The default is None.
28 """
29 def __init__(self, pump_id: str, speed_lower_bound: float = 0.,
30 speed_upper_bound: Optional[float] = None, **kwds):
31 if not isinstance(pump_id, str):
32 raise TypeError("'pump_id' must be an instance of 'str' " +
33 f"but not of '{type(pump_id)}'")
34 if not isinstance(speed_lower_bound, float):
35 raise TypeError("'speed_lower_bound' must be an instance of 'float' " +
36 f"but not of '{type(speed_lower_bound)}'")
37 if speed_lower_bound < 0:
38 raise ValueError("'speed_lower_bound' can not be negative!")
39 if speed_upper_bound is not None:
40 if not isinstance(speed_upper_bound, float):
41 raise TypeError("'speed_upper_bound' must be an instance of 'float' " +
42 f"but not of '{type(speed_upper_bound)}'")
43 if speed_upper_bound < 0:
44 raise ValueError("'speed_upper_bound' can not be negative!")
45
46 self._pump_id = pump_id
47 self._speed_lower_bound = speed_lower_bound
48 self._speed_upper_bound = speed_upper_bound \
49 if speed_upper_bound is not None else float("inf")
50
51 super().__init__(**kwds)
52
53 def __eq__(self, other) -> bool:
54 return super().__eq__(other) and self._pump_id == other.pump_id and \
55 self._speed_lower_bound == other.speed_lower_bound and\
56 self._speed_upper_bound == other.speed_upper_bound
57
58 def __str__(self) -> str:
59 return super().__str__() + f"Pump ID: {self._pump_id} " +\
60 f"Speed lower bound: {self._speed_lower_bound} " +\
61 f"Speed upper bound: {self._speed_upper_bound}"
62
63 @property
64 def pump_id(self) -> str:
65 """
66 Return the ID of the pump.
67
68 Returns
69 -------
70 `str`
71 ID of the pump.
72 """
73 return self._pump_id
74
75 @property
76 def speed_lower_bound(self) -> float:
77 """
78 Returns the lower bound of the pump speed that can be set by the agent.
79
80 Returns
81 -------
82 `float`
83 Lower bound on the pump speed.
84 """
85 return self._speed_lower_bound
86
87 @property
88 def speed_upper_bound(self) -> float:
89 """
90 Returns the upper bound of the pump speed that can be set by the agent.
91
92 Returns
93 -------
94 `float`
95 Upper bound on the pump speed.
96 """
97 return self._speed_upper_bound
98
[docs]
99 def to_gym_action_space(self) -> Space:
100 return Box(low=self._speed_lower_bound, high=self._speed_upper_bound)
101
[docs]
102 def apply(self, env: ScenarioControlEnv, action_value: float) -> None:
103 env.set_pump_speed(self._pump_id, action_value)