1+ from loguru import logger
2+ from typing import Union , List , Optional , Dict
3+ import numpy as np
4+
5+ import grid2op
6+ from grid2op .dtypes import dt_float
7+ from grid2op .Exceptions import Grid2OpException
8+
9+ class ThermalLimits :
10+ """
11+ Class for managing the thermal limits of power grid lines.
12+ """
13+
14+ def __init__ (
15+ self ,
16+ _thermal_limit_a : Optional [np .ndarray ] = None ,
17+ line_names : Optional [List [str ]] = None ,
18+ n_line : Optional [int ] = None
19+ ):
20+ """
21+ Initializes the thermal limits manager.
22+
23+ :param thermal_limits: Optional[np.ndarray]
24+ Array of thermal limits for each power line. Must have the same length as the number of lines.
25+ :param line_names: Optional[List[str]]
26+ List of power line names.
27+ :param n_line: Optional[int]
28+ Number of lines (can be passed explicitly or inferred from `thermal_limits` or `line_names`).
29+
30+ :raises ValueError:
31+ If neither `thermal_limits` nor `n_line` and `line_names` are provided.
32+ """
33+ if _thermal_limit_a is None and (n_line is None and line_names is None ):
34+ raise ValueError ("Must provide thermal_limits or both n_line and line_names." )
35+
36+ self ._thermal_limit_a = _thermal_limit_a
37+ self ._n_line = n_line
38+ self ._name_line = line_names
39+
40+ logger .info (f"ThermalLimits initialized with { self .n_line } )
41+
42+ @property
43+ def n_line (self ) -> int :
44+ return self ._n_line
45+
46+ @n_line .setter
47+ def n_line (self , new_n_line : int ) -> None :
48+ if new_n_line <= 0 :
49+ raise ValueError ("Number of lines must be a positive integer." )
50+ self ._n_line = new_n_line
51+ logger .info (f"Number of lines updated to { self ._n_line } )
52+
53+ @property
54+ def name_line (self ) -> Union [List [str ], np .ndarray ]:
55+ return self ._name_line
56+
57+ @name_line .setter
58+ def name_line (self , new_name_line : Union [List [str ], np .ndarray ]) -> None :
59+ if isinstance (new_name_line , np .ndarray ):
60+ if not np .all ([isinstance (name , str ) for name in new_name_line ]):
61+ raise ValueError ("All elements in name_line must be strings." )
62+ elif isinstance (new_name_line , list ):
63+ if not all (isinstance (name , str ) for name in new_name_line ):
64+ raise ValueError ("All elements in name_line must be strings." )
65+ else :
66+ raise ValueError ("Line names must be a list or numpy array of non-empty strings." )
67+
68+ if self ._n_line is not None and len (new_name_line ) != self ._n_line :
69+ raise ValueError ("Length of name list must match the number of lines." )
70+
71+ self ._name_line = new_name_line
72+ logger .info (f"Power line names updated" )
73+
74+ @property
75+ def limits (self ) -> np .ndarray :
76+ """
77+ Gets the current thermal limits of the power lines.
78+
79+ :return: np.ndarray
80+ The array containing thermal limits for each power line.
81+ """
82+ return self ._thermal_limit_a
83+
84+ @limits .setter
85+ def limits (self , new_limits : Union [np .ndarray , Dict [str , float ]]):
86+ """
87+ Sets new thermal limits.
88+
89+ :param new_limits: Union[np.ndarray, Dict[str, float]]
90+ Either a numpy array or a dictionary mapping line names to new thermal limits.
91+
92+ :raises ValueError:
93+ If the new limits array size does not match the number of lines.
94+ :raises Grid2OpException:
95+ If invalid power line names are provided in the dictionary.
96+ If the new thermal limit values are invalid (non-positive or non-convertible).
97+ :raises TypeError:
98+ If the input type is not an array or dictionary.
99+ """
100+ if isinstance (new_limits , np .ndarray ):
101+ if new_limits .shape [0 ] == self .n_line :
102+ self ._thermal_limit_a = 1.0 * new_limits .astype (dt_float )
103+ elif isinstance (new_limits , dict ):
104+ for el in new_limits .keys ():
105+ if not el in self .name_line :
106+ raise Grid2OpException (
107+ 'You asked to modify the thermal limit of powerline named "{}" that is not '
108+ "on the grid. Names of powerlines are {}" .format (
109+ el , self .name_line
110+ )
111+ )
112+ for i , el in self .name_line :
113+ if el in new_limits :
114+ try :
115+ tmp = dt_float (new_limits [el ])
116+ except Exception as exc_ :
117+ raise Grid2OpException (
118+ 'Impossible to convert data ({}) for powerline named "{}" into float '
119+ "values" .format (new_limits [el ], el )
120+ ) from exc_
121+ if tmp <= 0 :
122+ raise Grid2OpException (
123+ 'New thermal limit for powerlines "{}" is not positive ({})'
124+ "" .format (el , tmp )
125+ )
126+ self ._thermal_limit_a [i ] = tmp
127+
128+ def env_limits (self , thermal_limit ):
129+ if isinstance (thermal_limit , dict ):
130+ tmp = np .full (self .n_line , fill_value = np .NaN , dtype = dt_float )
131+ for key , val in thermal_limit .items ():
132+ if key not in self .name_line :
133+ raise Grid2OpException (
134+ f"When setting a thermal limit with a dictionary, the keys should be line "
135+ f"names. We found: { key }
136+ f"powerlines are { self .name_line }
137+ )
138+ ind_line = (self .name_line == key ).nonzero ()[0 ][0 ]
139+ if np .isfinite (tmp [ind_line ]):
140+ raise Grid2OpException (
141+ f"Humm, there is a really strange bug, some lines are set twice."
142+ )
143+ try :
144+ val_fl = float (val )
145+ except Exception as exc_ :
146+ raise Grid2OpException (
147+ f"When setting thermal limit with a dictionary, the keys should be "
148+ f"the values of the thermal limit (in amps) you provided something that "
149+ f'cannot be converted to a float. Error was "{ exc_ }
150+ )
151+ tmp [ind_line ] = val_fl
152+
153+ elif isinstance (thermal_limit , (np .ndarray , list )):
154+ try :
155+ tmp = np .array (thermal_limit ).flatten ().astype (dt_float )
156+ except Exception as exc_ :
157+ raise Grid2OpException (
158+ f"Impossible to convert the vector as input into a 1d numpy float array. "
159+ f"Error was: \n { exc_ }
160+ )
161+ if tmp .shape [0 ] != self .n_line :
162+ raise Grid2OpException (
163+ "Attempt to set thermal limit on {} powerlines while there are {}"
164+ "on the grid" .format (tmp .shape [0 ], self .n_line )
165+ )
166+ if (~ np .isfinite (tmp )).any ():
167+ raise Grid2OpException (
168+ "Impossible to use non finite value for thermal limits."
169+ )
170+ else :
171+ raise Grid2OpException (
172+ f"You can only set the thermal limits of the environment with a dictionary (in that "
173+ f"case the keys are the line names, and the values the thermal limits) or with "
174+ f"a numpy array that has as many components of the number of powerlines on "
175+ f'the grid. You provided something with type "{ type (thermal_limit )}
176+ f"is not supported."
177+ )
178+
179+ self ._thermal_limit_a = tmp
180+ logger .info ("Env thermal limits successfully set." )
181+
182+ def update_limits (self , thermal_limit_a : np .ndarray ) -> None :
183+ """
184+ Updates the thermal limits using a numpy array.
185+
186+ :param thermal_limit_a: np.ndarray
187+ The new array of thermal limits (in Amperes).
188+ """
189+ thermal_limit_a = np .array (thermal_limit_a ).astype (dt_float )
190+ self ._thermal_limit_a = thermal_limit_a
191+ logger .info ("Thermal limits updated from vector." )
0 commit comments