1"""
2This module contains classes for modeling chemical/species injection action spaces
3-- i.e. a chemical (EPANET) or species (EPANET-MSX) injection that has to be
4controlled by the agent.
5"""
6from typing import Optional
7from epyt_flow.gym import ScenarioControlEnv
8from epyt_flow.simulation import EpanetConstants
9from gymnasium.spaces import Space, Box
10
11from .actions import Action
12
13
[docs]
14class ChemicalInjectionAction(Action):
15 """
16 Action for controlling the injection of a chemical -- only for EPANET control scenarios.
17
18 Parameters
19 ----------
20 node_id : `str`
21 ID of the node at which the injection is going to happen.
22 pattern_id : `str`
23 ID of the pattern that is used for the injection.
24 source_type_id : `int`
25 Type of the injection source -- must be one of
26 the following EPANET constants:
27
28 - EN_CONCEN = 0
29 - EN_MASS = 1
30 - EN_SETPOINT = 2
31 - EN_FLOWPACED = 3
32
33 Description:
34
35 - E_CONCEN Sets the concentration of external inflow entering a node
36 - EN_MASS Injects a given mass/minute into a node
37 - EN_SETPOINT Sets the concentration leaving a node to a given value
38 - EN_FLOWPACED Adds a given value to the concentration leaving a node
39 upper_bound : `float`, optional
40 Upper bound on the amount that can be injected by the agent.
41 If None, there is no upper bound on the amount.
42
43 The default is None.
44 """
45 def __init__(self, node_id: str, pattern_id: str,
46 source_type_id: int, upper_bound: Optional[float] = None,
47 **kwds):
48 if not isinstance(node_id, str):
49 raise TypeError(f"'node_id' must be an instance of 'str' but not of '{type(node_id)}'")
50 if not isinstance(pattern_id, str):
51 raise TypeError("'pattern_id' must be an instance of 'str' " +
52 f"but not of '{type(pattern_id)}'")
53 if not isinstance(source_type_id, int):
54 raise TypeError("'source_type_id' must be an instance of 'int' " +
55 f"but not of '{type(source_type_id)}'")
56 if source_type_id not in [EpanetConstants.EN_MASS, EpanetConstants.EN_CONCEN,
57 EpanetConstants.EN_SETPOINT, EpanetConstants.EN_FLOWPACED]:
58 raise ValueError("Invalid 'source_type_id'")
59 if upper_bound is not None:
60 if not isinstance(upper_bound, float):
61 raise TypeError("'upper_bound' must be an instance of 'float' " +
62 f"but not of '{type(upper_bound)}'")
63 if upper_bound <= 0:
64 raise ValueError("'upper_bound' must be positive")
65
66 self._node_id = node_id
67 self._pattern_id = pattern_id
68 self._source_type_id = source_type_id
69 self._upper_bound = upper_bound
70
71 super().__init__(**kwds)
72
73 @property
74 def node_id(self) -> str:
75 """
76 Return the ID of the node at which the injection is going to happen.
77
78 Returns
79 -------
80 `str`
81 ID of the node.
82 """
83 return self._node_id
84
85 @property
86 def pattern_id(self) -> str:
87 """
88 Returns the ID of the pattern that is used for the injection.
89
90 Returns
91 -------
92 `str`
93 ID of the pattern.
94 """
95 return self._pattern_id
96
97 @property
98 def source_type_id(self) -> int:
99 """
100 Returns the type (i.e. ID) of the injection source -- will be one of
101 the following EPANET toolkit constants:
102
103 - EN_CONCEN = 0
104 - EN_MASS = 1
105 - EN_SETPOINT = 2
106 - EN_FLOWPACED = 3
107
108 Description:
109
110 - E_CONCEN Sets the concentration of external inflow entering a node
111 - EN_MASS Injects a given mass/minute into a node
112 - EN_SETPOINT Sets the concentration leaving a node to a given value
113 - EN_FLOWPACED Adds a given value to the concentration leaving a node
114
115 Returns
116 -------
117 `int`
118 Type (ID) of the injection source.
119 """
120 return self._source_type_id
121
122 @property
123 def upper_bound(self) -> float:
124 """
125 Returns the upper bound on the amount that can be injected by the agent.
126
127 Returns
128 -------
129 `float`
130 Upper bound.
131 """
132 return self._upper_bound
133
134 def __eq__(self, other) -> bool:
135 return super().__eq__(other) and self._node_id == other.node_id and \
136 self._pattern_id == other.pattern_id and \
137 self._source_type_id == other.source_type_id and \
138 self._upper_bound == other.upper_bound
139
140 def __str__(self) -> str:
141 return f"Node ID: {self._node_id} " +\
142 f"Pattern ID: {self._pattern_id} Source type ID: {self._source_type_id} " +\
143 f"Upper bound: {self._upper_bound}"
144
[docs]
145 def to_gym_action_space(self) -> Space:
146 return Box(low=0, high=self._upper_bound if self._upper_bound is not None
147 else float("inf"))
148
[docs]
149 def apply(self, env: ScenarioControlEnv, action_value: float) -> None:
150 env.set_node_quality_source_value(self._node_id, self._pattern_id, action_value)
151
152
[docs]
153class SpeciesInjectionAction(ChemicalInjectionAction):
154 """
155 Action for controlling the injection of a given species --
156 only for EPANET-MSX control scenarios.
157
158 Parameters
159 ----------
160 species_id : `str`
161 ID of the species that is going to be injected by the agent.
162 node_id : `str`
163 ID of the node at which the injection is going to happen.
164 pattern_id : `str`
165 ID of the pattern that is used for the injection.
166 source_type_id : `int`
167 Types of the injection source -- must be one of
168 the following EPANET toolkit constants:
169
170 - EN_CONCEN = 0
171 - EN_MASS = 1
172 - EN_SETPOINT = 2
173 - EN_FLOWPACED = 3
174
175 Description:
176
177 - E_CONCEN Sets the concentration of external inflow entering a node
178 - EN_MASS Injects a given mass/minute into a node
179 - EN_SETPOINT Sets the concentration leaving a node to a given value
180 - EN_FLOWPACED Adds a given value to the concentration leaving a node
181 upper_bound : `float`, optional
182 Upper bound on the amount that can be injected by the agent.
183 If None, there is no upper bound on the amount.
184
185 The default is None.
186 """
187 def __init__(self, species_id: str, node_id: str, pattern_id: str,
188 source_type_id: int, upper_bound: Optional[float] = None,
189 **kwds):
190 if not isinstance(species_id, str):
191 raise TypeError("'species_id' must be an instance of 'str' " +
192 f"but not of '{type(species_id)}'")
193
194 self._species_id = species_id
195
196 super().__init__(node_id=node_id, pattern_id=pattern_id,
197 source_type_id=source_type_id, upper_bound=upper_bound, **kwds)
198
199 @property
200 def species_id(self) -> str:
201 """
202 Returns the ID of the species that is going to be injected.
203
204 Returns
205 -------
206 `str`
207 ID of the species.
208 """
209 return self._species_id
210
211 def __eq__(self, other) -> bool:
212 return super().__eq__(other) and self._species_id == other.species_id
213
214 def __str__(self) -> str:
215 return super().__str__() + f"Species ID: {self._species_id}"
216
[docs]
217 def apply(self, env: ScenarioControlEnv, action_value: float) -> None:
218 env.set_node_species_source_value(self._species_id, self._node_id, self._source_type_id,
219 self._pattern_id, action_value)