diff --git a/roseau/load_flow/converters.py b/roseau/load_flow/converters.py index 26508bcf..1ebeafe3 100644 --- a/roseau/load_flow/converters.py +++ b/roseau/load_flow/converters.py @@ -23,7 +23,7 @@ [1, ALPHA**2, ALPHA], [1, ALPHA, ALPHA**2], ], - dtype=complex, + dtype=np.complex128, ) """numpy.ndarray[complex]: "A" matrix: transformation matrix from phasor to symmetrical components.""" @@ -32,7 +32,7 @@ def phasor_to_sym(v_abc: Sequence[complex]) -> ComplexArray: """Compute the symmetrical components `(0, +, -)` from the phasor components `(a, b, c)`.""" - v_abc_array = np.asarray(v_abc) + v_abc_array = np.array(v_abc) orig_shape = v_abc_array.shape v_012 = _A_INV @ v_abc_array.reshape((3, 1)) return v_012.reshape(orig_shape) @@ -40,7 +40,7 @@ def phasor_to_sym(v_abc: Sequence[complex]) -> ComplexArray: def sym_to_phasor(v_012: Sequence[complex]) -> ComplexArray: """Compute the phasor components `(a, b, c)` from the symmetrical components `(0, +, -)`.""" - v_012_array = np.asarray(v_012) + v_012_array = np.array(v_012) orig_shape = v_012_array.shape v_abc = A @ v_012_array.reshape((3, 1)) return v_abc.reshape(orig_shape) @@ -124,13 +124,13 @@ def calculate_voltages(potentials: ComplexArray, phases: str) -> ComplexArray: Otherwise, the voltages are Phase-Phase. Example: - >>> potentials = 230 * np.array([1, np.exp(-2j*np.pi/3), np.exp(2j*np.pi/3), 0], dtype=complex) + >>> potentials = 230 * np.array([1, np.exp(-2j*np.pi/3), np.exp(2j*np.pi/3), 0], dtype=np.complex128) >>> calculate_voltages(potentials, "abcn") array([ 230. +0.j , -115.-199.18584287j, -115.+199.18584287j]) - >>> potentials = np.array([230, 230 * np.exp(-2j*np.pi/3)], dtype=complex) + >>> potentials = np.array([230, 230 * np.exp(-2j*np.pi/3)], dtype=np.complex128) >>> calculate_voltages(potentials, "ab") array([345.+199.18584287j]) - >>> calculate_voltages(np.array([230, 0], dtype=complex), "an") + >>> calculate_voltages(np.array([230, 0], dtype=np.complex128), "an") array([230.+0.j]) """ assert len(potentials) == len(phases), "Number of potentials must match number of phases." diff --git a/roseau/load_flow/models/branches.py b/roseau/load_flow/models/branches.py index 2aac6a0d..6e515fee 100644 --- a/roseau/load_flow/models/branches.py +++ b/roseau/load_flow/models/branches.py @@ -140,8 +140,8 @@ def to_dict(self, *, _lf_only: bool = False) -> JsonDict: return res def results_from_dict(self, data: JsonDict) -> None: - currents1 = np.array([complex(i[0], i[1]) for i in data["currents1"]], dtype=complex) - currents2 = np.array([complex(i[0], i[1]) for i in data["currents2"]], dtype=complex) + currents1 = np.array([complex(i[0], i[1]) for i in data["currents1"]], dtype=np.complex128) + currents2 = np.array([complex(i[0], i[1]) for i in data["currents2"]], dtype=np.complex128) self._res_currents = (currents1, currents2) def _results_to_dict(self, warning: bool) -> JsonDict: diff --git a/roseau/load_flow/models/buses.py b/roseau/load_flow/models/buses.py index 10aad31b..e2347db7 100644 --- a/roseau/load_flow/models/buses.py +++ b/roseau/load_flow/models/buses.py @@ -1,6 +1,6 @@ import logging -from collections.abc import Iterator, Sequence -from typing import TYPE_CHECKING, Any, Optional +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any, Optional, Union import numpy as np import pandas as pd @@ -10,7 +10,7 @@ from roseau.load_flow.converters import calculate_voltage_phases, calculate_voltages, phasor_to_sym from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode from roseau.load_flow.models.core import Element -from roseau.load_flow.typing import ComplexArray, Id, JsonDict +from roseau.load_flow.typing import ComplexArray, ComplexArrayLike1D, Id, JsonDict from roseau.load_flow.units import Q_, ureg_wraps logger = logging.getLogger(__name__) @@ -36,7 +36,7 @@ def __init__( *, phases: str, geometry: Optional[Point] = None, - potentials: Optional[Sequence[complex]] = None, + potentials: Optional[ComplexArrayLike1D] = None, min_voltage: Optional[float] = None, max_voltage: Optional[float] = None, **kwargs: Any, @@ -57,15 +57,20 @@ def __init__( x-y coordinates of the bus. potentials: - An optional list of initial potentials of each phase of the bus. + An optional array-like of initial potentials of each phase of the bus. If given, + these potentials are used as the starting point of the load flow computation. + Either complex values (V) or a :class:`Quantity ` of + complex values. min_voltage: An optional minimum voltage of the bus (V). It is not used in the load flow. It must be a phase-neutral voltage if the bus has a neutral, phase-phase otherwise. + Either a float (V) or a :class:`Quantity ` of float. max_voltage: An optional maximum voltage of the bus (V). It is not used in the load flow. It must be a phase-neutral voltage if the bus has a neutral, phase-phase otherwise. + Either a float (V) or a :class:`Quantity ` of float. """ super().__init__(id, **kwargs) self._check_phases(id, phases=phases) @@ -88,17 +93,17 @@ def __repr__(self) -> str: @property @ureg_wraps("V", (None,), strict=False) def potentials(self) -> Q_[ComplexArray]: - """The potentials of the bus (V).""" + """An array of initial potentials of the bus (V).""" return self._potentials @potentials.setter @ureg_wraps(None, (None, "V"), strict=False) - def potentials(self, value: Sequence[complex]) -> None: + def potentials(self, value: ComplexArrayLike1D) -> None: if len(value) != len(self.phases): msg = f"Incorrect number of potentials: {len(value)} instead of {len(self.phases)}" logger.error(msg) raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_POTENTIALS_SIZE) - self._potentials = np.asarray(value, dtype=complex) + self._potentials = np.array(value, dtype=np.complex128) self._invalidate_network_results() def _res_potentials_getter(self, warning: bool) -> ComplexArray: @@ -111,7 +116,7 @@ def res_potentials(self) -> Q_[ComplexArray]: return self._res_potentials_getter(warning=True) def _res_voltages_getter(self, warning: bool) -> ComplexArray: - potentials = np.asarray(self._res_potentials_getter(warning=warning)) + potentials = np.array(self._res_potentials_getter(warning=warning)) return calculate_voltages(potentials, self.phases) @property @@ -142,7 +147,7 @@ def min_voltage(self) -> Optional[Q_[float]]: @min_voltage.setter @ureg_wraps(None, (None, "V"), strict=False) - def min_voltage(self, value: Optional[float]) -> None: + def min_voltage(self, value: Optional[Union[float, Q_[float]]]) -> None: if value is not None and self._max_voltage is not None and value > self._max_voltage: msg = ( f"Cannot set min voltage of bus {self.id!r} to {value} V as it is higher than its " @@ -161,7 +166,7 @@ def max_voltage(self) -> Optional[Q_[float]]: @max_voltage.setter @ureg_wraps(None, (None, "V"), strict=False) - def max_voltage(self, value: Optional[float]) -> None: + def max_voltage(self, value: Optional[Union[float, Q_[float]]]) -> None: if value is not None and self._min_voltage is not None and value < self._min_voltage: msg = ( f"Cannot set max voltage of bus {self.id!r} to {value} V as it is lower than its " @@ -334,7 +339,7 @@ def to_dict(self, *, _lf_only: bool = False) -> JsonDict: return res def results_from_dict(self, data: JsonDict) -> None: - self._res_potentials = np.array([complex(v[0], v[1]) for v in data["potentials"]], dtype=complex) + self._res_potentials = np.array([complex(v[0], v[1]) for v in data["potentials"]], dtype=np.complex128) def _results_to_dict(self, warning: bool) -> JsonDict: return { diff --git a/roseau/load_flow/models/lines/lines.py b/roseau/load_flow/models/lines/lines.py index 2c7aee71..3fe052af 100644 --- a/roseau/load_flow/models/lines/lines.py +++ b/roseau/load_flow/models/lines/lines.py @@ -1,6 +1,6 @@ import logging import warnings -from typing import Any, Optional +from typing import Any, Optional, Union import numpy as np from shapely import LineString, Point @@ -144,7 +144,7 @@ def __init__( bus2: Bus, *, parameters: LineParameters, - length: float, + length: Union[float, Q_[float]], phases: Optional[str] = None, ground: Optional[Ground] = None, geometry: Optional[LineString] = None, @@ -231,7 +231,7 @@ def length(self) -> Q_[float]: @length.setter @ureg_wraps(None, (None, "km"), strict=False) - def length(self, value: float) -> None: + def length(self, value: Union[float, Q_[float]]) -> None: if value <= 0: msg = f"A line length must be greater than 0. {value:.2f} km provided." logger.error(msg) @@ -335,7 +335,7 @@ def _res_shunt_values_getter(self, warning: bool) -> tuple[ComplexArray, Complex def _res_shunt_currents_getter(self, warning: bool) -> tuple[ComplexArray, ComplexArray]: if not self.with_shunt: - zeros = np.zeros(len(self.phases), dtype=complex) + zeros = np.zeros(len(self.phases), dtype=np.complex128) return zeros[:], zeros[:] _, _, cur1, cur2 = self._res_shunt_values_getter(warning) return cur1, cur2 @@ -348,7 +348,7 @@ def res_shunt_currents(self) -> tuple[Q_[ComplexArray], Q_[ComplexArray]]: def _res_shunt_power_losses_getter(self, warning: bool) -> ComplexArray: if not self.with_shunt: - return np.zeros(len(self.phases), dtype=complex) + return np.zeros(len(self.phases), dtype=np.complex128) pot1, pot2, cur1, cur2 = self._res_shunt_values_getter(warning) return pot1 * cur1.conj() + pot2 * cur2.conj() diff --git a/roseau/load_flow/models/lines/parameters.py b/roseau/load_flow/models/lines/parameters.py index a6294b2f..53a0f68f 100644 --- a/roseau/load_flow/models/lines/parameters.py +++ b/roseau/load_flow/models/lines/parameters.py @@ -1,6 +1,6 @@ import logging import re -from typing import NoReturn, Optional +from typing import NoReturn, Optional, Union import numpy as np import numpy.linalg as nplin @@ -8,7 +8,7 @@ from typing_extensions import Self from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode -from roseau.load_flow.typing import ComplexArray, Id, JsonDict +from roseau.load_flow.typing import ComplexArray, ComplexArrayLike2D, Id, JsonDict from roseau.load_flow.units import Q_, ureg_wraps from roseau.load_flow.utils import ( CX, @@ -43,8 +43,8 @@ class LineParameters(Identifiable, JsonMixin): def __init__( self, id: Id, - z_line: ComplexArray, - y_shunt: Optional[ComplexArray] = None, + z_line: ComplexArrayLike2D, + y_shunt: Optional[ComplexArrayLike2D] = None, max_current: Optional[float] = None, ) -> None: """LineParameters constructor. @@ -63,13 +63,13 @@ def __init__( An optional maximum current loading of the line (A). It is not used in the load flow. """ super().__init__(id) - self._z_line = np.asarray(z_line, dtype=complex) + self._z_line = np.array(z_line, dtype=np.complex128) if y_shunt is None: self._with_shunt = False - self._y_shunt = np.zeros_like(z_line, dtype=complex) + self._y_shunt = np.zeros_like(self._z_line, dtype=np.complex128) else: self._with_shunt = not np.allclose(y_shunt, 0) - self._y_shunt = np.asarray(y_shunt, dtype=complex) + self._y_shunt = np.array(y_shunt, dtype=np.complex128) self.max_current = max_current self._check_matrix() @@ -112,7 +112,7 @@ def max_current(self) -> Optional[Q_[float]]: @max_current.setter @ureg_wraps(None, (None, "A"), strict=False) - def max_current(self, value: Optional[float]) -> None: + def max_current(self, value: Optional[Union[float, Q_[float]]]) -> None: self._max_current = value @classmethod @@ -122,15 +122,15 @@ def max_current(self, value: Optional[float]) -> None: def from_sym( cls, id: Id, - z0: complex, - z1: complex, - y0: complex, - y1: complex, - zn: Optional[complex] = None, - xpn: Optional[float] = None, - bn: Optional[float] = None, - bpn: Optional[float] = None, - max_current: Optional[float] = None, + z0: Union[complex, Q_[complex]], + z1: Union[complex, Q_[complex]], + y0: Union[complex, Q_[complex]], + y1: Union[complex, Q_[complex]], + zn: Optional[Union[complex, Q_[complex]]] = None, + xpn: Optional[Union[float, Q_[float]]] = None, + bn: Optional[Union[float, Q_[float]]] = None, + bpn: Optional[Union[float, Q_[float]]] = None, + max_current: Optional[Union[float, Q_[float]]] = None, ) -> Self: """Create line parameters from a symmetric model. @@ -245,8 +245,8 @@ def _sym_to_zy( # If all the neutral data have not been filled, the matrix is a 3x3 matrix if any_neutral_na: # No neutral data so retrieve a 3x3 matrix - z_line = np.array([[zs, zm, zm], [zm, zs, zm], [zm, zm, zs]], dtype=complex) - y_shunt = np.array([[ys, ym, ym], [ym, ys, ym], [ym, ym, ys]], dtype=complex) + z_line = np.array([[zs, zm, zm], [zm, zs, zm], [zm, zm, zs]], dtype=np.complex128) + y_shunt = np.array([[ys, ym, ym], [ym, ys, ym], [ym, ym, ys]], dtype=np.complex128) else: # Build the complex # zn: Neutral series impedance (ohm/km) @@ -259,16 +259,16 @@ def _sym_to_zy( f"The line model {id!r} does not have neutral elements. It will be modelled as a 3 wires line " f"instead." ) - z_line = np.array([[zs, zm, zm], [zm, zs, zm], [zm, zm, zs]], dtype=complex) - y_shunt = np.array([[ys, ym, ym], [ym, ys, ym], [ym, ym, ys]], dtype=complex) + z_line = np.array([[zs, zm, zm], [zm, zs, zm], [zm, zm, zs]], dtype=np.complex128) + y_shunt = np.array([[ys, ym, ym], [ym, ys, ym], [ym, ym, ys]], dtype=np.complex128) else: z_line = np.array( [[zs, zm, zm, zpn], [zm, zs, zm, zpn], [zm, zm, zs, zpn], [zpn, zpn, zpn, zn]], - dtype=complex, + dtype=np.complex128, ) y_shunt = np.array( [[ys, ym, ym, ypn], [ym, ys, ym, ypn], [ym, ym, ys, ypn], [ypn, ypn, ypn, yn]], - dtype=complex, + dtype=np.complex128, ) # Check the validity of the resulting matrices @@ -304,11 +304,11 @@ def from_geometry( line_type: LineType, conductor_type: ConductorType, insulator_type: InsulatorType, - section: float, - section_neutral: float, - height: float, - external_diameter: float, - max_current: Optional[float] = None, + section: Union[float, Q_[float]], + section_neutral: Union[float, Q_[float]], + height: Union[float, Q_[float]], + external_diameter: Union[float, Q_[float]], + max_current: Optional[Union[float, Q_[float]]] = None, ) -> Self: """Create line parameters from its geometry. @@ -448,7 +448,7 @@ def _geometry_to_zy( raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_LINE_TYPE) # Distance computation - sections = np.array([section, section, section, section_neutral], dtype=float) * 1e-6 # surfaces (m2) + sections = np.array([section, section, section, section_neutral], dtype=np.float64) * 1e-6 # surfaces (m2) radius = np.sqrt(sections / PI) # radius (m) gmr = radius * np.exp(-0.25) # geometric mean radius (m) # distance between two wires (m) @@ -460,13 +460,13 @@ def _geometry_to_zy( distance_prim = np.sqrt(np.einsum("ijk,ijk->ij", diff, diff)) # Useful matrices - mask_diagonal = np.eye(4, dtype=bool) + mask_diagonal = np.eye(4, dtype=np.bool_) mask_off_diagonal = ~mask_diagonal - minus = -np.ones((4, 4), dtype=float) + minus = -np.ones((4, 4), dtype=np.float64) np.fill_diagonal(minus, 1) # Electrical parameters - r = RHO[conductor_type].m_as("ohm*m") / sections * np.eye(4, dtype=float) * 1e3 # resistance (ohm/km) + r = RHO[conductor_type].m_as("ohm*m") / sections * np.eye(4, dtype=np.float64) * 1e3 # resistance (ohm/km) distance[mask_diagonal] = gmr inductance = MU_0.m_as("H/m") / (2 * PI) * np.log(1 / distance) * 1e3 # H/m->H/km distance[mask_diagonal] = radius @@ -474,10 +474,10 @@ def _geometry_to_zy( # Extract the conductivity and the capacities from the lambda (potential coefficients) lambda_inv = nplin.inv(lambdas) * 1e3 # capacities (F/km) - c = np.zeros((4, 4), dtype=float) # capacities (F/km) + c = np.zeros((4, 4), dtype=np.float64) # capacities (F/km) c[mask_diagonal] = np.einsum("ij,ij->i", lambda_inv, minus) c[mask_off_diagonal] = -lambda_inv[mask_off_diagonal] - g = np.zeros((4, 4), dtype=float) # conductance (S/km) + g = np.zeros((4, 4), dtype=np.float64) # conductance (S/km) omega = OMEGA.m_as("rad/s") g[mask_diagonal] = TAN_D[insulator_type].magnitude * np.einsum("ii->i", c) * omega @@ -486,7 +486,7 @@ def _geometry_to_zy( y = g + c * omega * 1j # Compute the shunt admittance matrix from the admittance matrix - y_shunt = np.zeros((4, 4), dtype=complex) + y_shunt = np.zeros((4, 4), dtype=np.complex128) y_shunt[mask_diagonal] = np.einsum("ij->i", y) y_shunt[mask_off_diagonal] = -y[mask_off_diagonal] @@ -497,10 +497,10 @@ def _geometry_to_zy( def from_name_lv( cls, name: str, - section_neutral: Optional[float] = None, - height: Optional[float] = None, - external_diameter: Optional[float] = None, - max_current: Optional[float] = None, + section_neutral: Optional[Union[float, Q_[float]]] = None, + height: Optional[Union[float, Q_[float]]] = None, + external_diameter: Optional[Union[float, Q_[float]]] = None, + max_current: Optional[Union[float, Q_[float]]] = None, ) -> Self: """Method to get the electrical parameters of a LV line from its canonical name. Some hypothesis will be made: the section of the neutral is the same as the other sections, the height and @@ -559,7 +559,8 @@ def from_name_lv( ) @classmethod - def from_name_mv(cls, name: str, max_current: Optional[float] = None) -> Self: + @ureg_wraps(None, (None, None, "A"), strict=False) + def from_name_mv(cls, name: str, max_current: Optional[Union[float, Q_[float]]] = None) -> Self: """Method to get the electrical parameters of a MV line from its canonical name. Args: @@ -603,8 +604,8 @@ def from_name_mv(cls, name: str, max_current: Optional[float] = None) -> Self: b = (c_b1 + c_b2 * section) * 1e-4 * OMEGA b = b.to("S/km") - z_line = (r + x * 1j) * np.eye(3, dtype=float) # in ohms/km - y_shunt = b * 1j * np.eye(3, dtype=float) # in siemens/km + z_line = (r + x * 1j) * np.eye(3, dtype=np.float64) # in ohms/km + y_shunt = b * 1j * np.eye(3, dtype=np.float64) # in siemens/km return cls(name, z_line=z_line, y_shunt=y_shunt, max_current=max_current) # @@ -621,8 +622,8 @@ def from_dict(cls, data: JsonDict) -> Self: Returns: The created line parameters. """ - z_line = np.asarray(data["z_line"][0]) + 1j * np.asarray(data["z_line"][1]) - y_shunt = np.asarray(data["y_shunt"][0]) + 1j * np.asarray(data["y_shunt"][1]) if "y_shunt" in data else None + z_line = np.array(data["z_line"][0]) + 1j * np.array(data["z_line"][1]) + y_shunt = np.array(data["y_shunt"][0]) + 1j * np.array(data["y_shunt"][1]) if "y_shunt" in data else None return cls(id=data["id"], z_line=z_line, y_shunt=y_shunt, max_current=data.get("max_current")) def to_dict(self, *, _lf_only: bool = False) -> JsonDict: diff --git a/roseau/load_flow/models/loads/flexible_parameters.py b/roseau/load_flow/models/loads/flexible_parameters.py index b92ffc3f..44579c16 100644 --- a/roseau/load_flow/models/loads/flexible_parameters.py +++ b/roseau/load_flow/models/loads/flexible_parameters.py @@ -1,13 +1,20 @@ import logging import warnings -from typing import TYPE_CHECKING, NoReturn, Optional +from typing import TYPE_CHECKING, NoReturn, Optional, Union import numpy as np from numpy.typing import NDArray from typing_extensions import Self from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode -from roseau.load_flow.typing import Authentication, ComplexArray, ControlType, JsonDict, ProjectionType +from roseau.load_flow.typing import ( + Authentication, + ComplexArray, + ComplexArrayLike1D, + ControlType, + JsonDict, + ProjectionType, +) from roseau.load_flow.units import Q_, ureg_wraps from roseau.load_flow.utils import JsonMixin, _optional_deps @@ -41,11 +48,11 @@ class Control(JsonMixin): def __init__( self, type: ControlType, - u_min: float, - u_down: float, - u_up: float, - u_max: float, - alpha: float = _DEFAULT_ALPHA, + u_min: Union[float, Q_[float]], + u_down: Union[float, Q_[float]], + u_up: Union[float, Q_[float]], + u_max: Union[float, Q_[float]], + alpha: Union[float, Q_[float]] = _DEFAULT_ALPHA, ) -> None: """Control constructor. @@ -181,7 +188,9 @@ def constant(cls) -> Self: @classmethod @ureg_wraps(None, (None, "V", "V", None), strict=False) - def p_max_u_production(cls, u_up: float, u_max: float, alpha: float = _DEFAULT_ALPHA) -> Self: + def p_max_u_production( + cls, u_up: Union[float, Q_[float]], u_max: Union[float, Q_[float]], alpha: float = _DEFAULT_ALPHA + ) -> Self: """Create a control of the type ``"p_max_u_production"``. See Also: @@ -210,7 +219,9 @@ def p_max_u_production(cls, u_up: float, u_max: float, alpha: float = _DEFAULT_A @classmethod @ureg_wraps(None, (None, "V", "V", None), strict=False) - def p_max_u_consumption(cls, u_min: float, u_down: float, alpha: float = _DEFAULT_ALPHA) -> Self: + def p_max_u_consumption( + cls, u_min: Union[float, Q_[float]], u_down: Union[float, Q_[float]], alpha: float = _DEFAULT_ALPHA + ) -> Self: """Create a control of the type ``"p_max_u_consumption"``. See Also: @@ -239,7 +250,14 @@ def p_max_u_consumption(cls, u_min: float, u_down: float, alpha: float = _DEFAUL @classmethod @ureg_wraps(None, (None, "V", "V", "V", "V", None), strict=False) - def q_u(cls, u_min: float, u_down: float, u_up: float, u_max: float, alpha: float = _DEFAULT_ALPHA) -> Self: + def q_u( + cls, + u_min: Union[float, Q_[float]], + u_down: Union[float, Q_[float]], + u_up: Union[float, Q_[float]], + u_max: Union[float, Q_[float]], + alpha: float = _DEFAULT_ALPHA, + ) -> Self: """Create a control of the type ``"q_u"``. See Also: @@ -446,9 +464,9 @@ def __init__( control_p: Control, control_q: Control, projection: Projection, - s_max: float, - q_min: Optional[float] = None, - q_max: Optional[float] = None, + s_max: Union[float, Q_[float]], + q_min: Optional[Union[float, Q_[float]]] = None, + q_max: Optional[Union[float, Q_[float]]] = None, ) -> None: """FlexibleParameter constructor. @@ -490,7 +508,7 @@ def s_max(self) -> Q_[float]: @s_max.setter @ureg_wraps(None, (None, "VA"), strict=False) - def s_max(self, value: float) -> None: + def s_max(self, value: Union[float, Q_[float]]) -> None: if value <= 0: s_max = Q_(value, "VA") msg = f"'s_max' must be greater than 0 but {s_max:P#~} was provided." @@ -512,7 +530,7 @@ def q_min(self) -> Q_[float]: @q_min.setter @ureg_wraps(None, (None, "VAr"), strict=False) - def q_min(self, value: Optional[float]) -> None: + def q_min(self, value: Optional[Union[float, Q_[float]]]) -> None: if value is not None and value < -self._s_max: q_min = Q_(value, "VAr") msg = f"'q_min' must be greater than -s_max ({-self.s_max:P#~}) but {q_min:P#~} was provided." @@ -533,7 +551,7 @@ def q_max(self) -> Q_[float]: @q_max.setter @ureg_wraps(None, (None, "VAr"), strict=False) - def q_max(self, value: Optional[float]) -> None: + def q_max(self, value: Optional[Union[float, Q_[float]]]) -> None: if value is not None and value > self._s_max: q_max = Q_(value, "VAr") msg = f"'q_max' must be less than s_max ({self.s_max:P#~}) but {q_max:P#~} was provided." @@ -564,9 +582,9 @@ def constant(cls) -> Self: @ureg_wraps(None, (None, "V", "V", "VA", None, None, None, None), strict=False) def p_max_u_production( cls, - u_up: float, - u_max: float, - s_max: float, + u_up: Union[float, Q_[float]], + u_max: Union[float, Q_[float]], + s_max: Union[float, Q_[float]], alpha_control: float = Control._DEFAULT_ALPHA, type_proj: ProjectionType = Projection._DEFAULT_TYPE, alpha_proj: float = Projection._DEFAULT_ALPHA, @@ -620,9 +638,9 @@ def p_max_u_production( @ureg_wraps(None, (None, "V", "V", "VA", None, None, None, None), strict=False) def p_max_u_consumption( cls, - u_min: float, - u_down: float, - s_max: float, + u_min: Union[float, Q_[float]], + u_down: Union[float, Q_[float]], + s_max: Union[float, Q_[float]], alpha_control: float = Control._DEFAULT_ALPHA, type_proj: ProjectionType = Projection._DEFAULT_TYPE, alpha_proj: float = Projection._DEFAULT_ALPHA, @@ -673,13 +691,13 @@ def p_max_u_consumption( @ureg_wraps(None, (None, "V", "V", "V", "V", "VA", "Var", "Var", None, None, None, None), strict=False) def q_u( cls, - u_min: float, - u_down: float, - u_up: float, - u_max: float, - s_max: float, - q_min: Optional[float] = None, - q_max: Optional[float] = None, + u_min: Union[float, Q_[float]], + u_down: Union[float, Q_[float]], + u_up: Union[float, Q_[float]], + u_max: Union[float, Q_[float]], + s_max: Union[float, Q_[float]], + q_min: Optional[Union[float, Q_[float]]] = None, + q_max: Optional[Union[float, Q_[float]]] = None, alpha_control: float = Control._DEFAULT_ALPHA, type_proj: ProjectionType = Projection._DEFAULT_TYPE, alpha_proj: float = Projection._DEFAULT_ALPHA, @@ -747,15 +765,15 @@ def q_u( @ureg_wraps(None, (None, "V", "V", "V", "V", "V", "V", "VA", "VAr", "VAr", None, None, None, None), strict=False) def pq_u_production( cls, - up_up: float, - up_max: float, - uq_min: float, - uq_down: float, - uq_up: float, - uq_max: float, - s_max: float, - q_min: Optional[float] = None, - q_max: Optional[float] = None, + up_up: Union[float, Q_[float]], + up_max: Union[float, Q_[float]], + uq_min: Union[float, Q_[float]], + uq_down: Union[float, Q_[float]], + uq_up: Union[float, Q_[float]], + uq_max: Union[float, Q_[float]], + s_max: Union[float, Q_[float]], + q_min: Optional[Union[float, Q_[float]]] = None, + q_max: Optional[Union[float, Q_[float]]] = None, alpha_control=Control._DEFAULT_ALPHA, type_proj: ProjectionType = Projection._DEFAULT_TYPE, alpha_proj=Projection._DEFAULT_ALPHA, @@ -834,16 +852,16 @@ def pq_u_production( @ureg_wraps(None, (None, "V", "V", "V", "V", "V", "V", "VA", "VAr", "VAr", None, None, None, None), strict=False) def pq_u_consumption( cls, - up_min: float, - up_down: float, - uq_min: float, - uq_down: float, - uq_up: float, - uq_max: float, - s_max: float, - q_min: Optional[float] = None, - q_max: Optional[float] = None, - alpha_control: float = Control._DEFAULT_ALPHA, + up_min: Union[float, Q_[float]], + up_down: Union[float, Q_[float]], + uq_min: Union[float, Q_[float]], + uq_down: Union[float, Q_[float]], + uq_up: Union[float, Q_[float]], + uq_max: Union[float, Q_[float]], + s_max: Union[float, Q_[float]], + q_min: Optional[Union[float, Q_[float]]] = None, + q_max: Optional[Union[float, Q_[float]]] = None, + alpha_control: Union[float, Q_[float]] = Control._DEFAULT_ALPHA, type_proj: ProjectionType = Projection._DEFAULT_TYPE, alpha_proj: float = Projection._DEFAULT_ALPHA, epsilon_proj: float = Projection._DEFAULT_EPSILON, @@ -966,8 +984,8 @@ def results_from_dict(self, data: JsonDict) -> NoReturn: def compute_powers( self, auth: Authentication, - voltages: NDArray[np.float_], - power: complex, + voltages: ComplexArrayLike1D, + power: Union[complex, Q_[complex]], solve_kwargs: Optional[JsonDict] = None, ) -> Q_[ComplexArray]: """Compute the flexible powers for different voltages (norms) @@ -991,14 +1009,14 @@ def compute_powers( return self._compute_powers(auth=auth, voltages=voltages, power=power, solve_kwargs=solve_kwargs) def _compute_powers( - self, auth: Authentication, voltages: NDArray[np.float_], power: complex, solve_kwargs: Optional[JsonDict] + self, auth: Authentication, voltages: ComplexArrayLike1D, power: complex, solve_kwargs: Optional[JsonDict] ) -> ComplexArray: from roseau.load_flow import Bus, ElectricalNetwork, PotentialRef, PowerLoad, VoltageSource # Format the input if solve_kwargs is None: solve_kwargs = {} - voltages = np.array(np.abs(voltages), dtype=float) + voltages = np.array(np.abs(voltages), dtype=np.float64) # Simple network bus = Bus(id="bus", phases="an") @@ -1015,14 +1033,14 @@ def _compute_powers( en.solve_load_flow(auth=auth, **solve_kwargs) res_flexible_powers.append(load.res_flexible_powers.m_as("VA")[0]) - return np.array(res_flexible_powers, dtype=complex) + return np.array(res_flexible_powers, dtype=np.complex128) @ureg_wraps((None, "VA"), (None, None, "V", "VA", None, None, None, "VA"), strict=False) def plot_pq( self, auth: Authentication, - voltages: NDArray[np.float_], - power: complex, + voltages: Union[NDArray[np.float64], Q_[NDArray[np.float64]]], + power: Union[complex, Q_[complex]], ax: Optional["Axes"] = None, solve_kwargs: Optional[JsonDict] = None, voltages_labels_mask: Optional[NDArray[np.bool_]] = None, @@ -1051,7 +1069,7 @@ def plot_pq( A mask to activate the plot of voltages labels. By default, no voltages annotations. res_flexible_powers: - If None, is provided, the `res_flexible_powers` are computed. Other + If None is provided, the `res_flexible_powers` are computed. Otherwise, the provided values are used. Returns: The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else @@ -1066,9 +1084,9 @@ def plot_pq( # Initialise some variables if voltages_labels_mask is None: - voltages_labels_mask = np.zeros_like(voltages, dtype=bool) + voltages_labels_mask = np.zeros_like(voltages, dtype=np.bool_) else: - voltages_labels_mask = np.array(voltages_labels_mask, dtype=bool) + voltages_labels_mask = np.array(voltages_labels_mask, dtype=np.bool_) s_max = self._s_max v_min = voltages.min() v_max = voltages.max() @@ -1134,8 +1152,8 @@ def plot_pq( def plot_control_p( self, auth: Authentication, - voltages: NDArray[np.float_], - power: complex, + voltages: Union[NDArray[np.float64], Q_[NDArray[np.float64]]], + power: Union[complex, Q_[complex]], ax: Optional["Axes"] = None, solve_kwargs: Optional[JsonDict] = None, res_flexible_powers: Optional[ComplexArray] = None, @@ -1159,7 +1177,7 @@ def plot_control_p( The keywords arguments of the :meth:`~roseau.load_flow.ElectricalNetwork.solve_load_flow` method. res_flexible_powers: - If None, is provided, the `res_flexible_powers` are computed. Other + If None is provided, the `res_flexible_powers` are computed. Otherwise, the provided values are used. Returns: The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else @@ -1200,8 +1218,8 @@ def plot_control_p( def plot_control_q( self, auth: Authentication, - voltages: NDArray[np.float_], - power: complex, + voltages: Union[NDArray[np.float64], Q_[NDArray[np.float64]]], + power: Union[complex, Q_[complex]], ax: Optional["Axes"] = None, solve_kwargs: Optional[JsonDict] = None, res_flexible_powers: Optional[ComplexArray] = None, @@ -1225,7 +1243,7 @@ def plot_control_q( The keywords arguments of the :meth:`~roseau.load_flow.ElectricalNetwork.solve_load_flow` method res_flexible_powers: - If None, is provided, the `res_flexible_powers` are computed. Other + If None is provided, the `res_flexible_powers` are computed. Otherwise, the provided values are used. Returns: The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else @@ -1268,7 +1286,7 @@ def plot_control_q( @staticmethod def _theoretical_control_data( control: Control, v_min: float, v_max: float, power: float, s_max: float - ) -> tuple[NDArray[np.float_], NDArray[np.float_], NDArray[np.object_]]: + ) -> tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.object_]]: """Helper to get data for the different plots of the class. It provides the theoretical control curve abscissas and ordinates values. It also provides ticks for the abscissa axis. @@ -1293,27 +1311,27 @@ def _theoretical_control_data( """ # Depending on the type of the control, several options if control.type == "constant": - x = np.array([v_min, v_max], dtype=float) - y = np.array([power, power], dtype=float) - x_ticks = np.array([f"{v_min:.1f}", f"{v_max:.1f}"], dtype=object) + x = np.array([v_min, v_max], dtype=np.float64) + y = np.array([power, power], dtype=np.float64) + x_ticks = np.array([f"{v_min:.1f}", f"{v_max:.1f}"], dtype=np.object_) elif control.type == "p_max_u_production": u_up = control._u_up u_max = control._u_max - x = np.array([u_up, u_max, v_min, v_max], dtype=float) - y = np.zeros_like(x, dtype=float) + x = np.array([u_up, u_max, v_min, v_max], dtype=np.float64) + y = np.zeros_like(x, dtype=np.float64) y[x < u_up] = -s_max mask = np.logical_and(u_up <= x, x < u_max) y[mask] = -s_max * (x[mask] - u_max) / (u_up - u_max) y[x >= u_max] = 0 x_ticks = np.array( [f"{u_up:.1f}\n$U^{{\\mathrm{{up}}}}$", f"{u_max:.1f}\n$U^{{\\max}}$", f"{v_min:.1f}", f"{v_max:.1f}"], - dtype=object, + dtype=np.object_, ) elif control.type == "p_max_u_consumption": u_min = control._u_min u_down = control._u_down - x = np.array([u_min, u_down, v_min, v_max], dtype=float) - y = np.zeros_like(x, dtype=float) + x = np.array([u_min, u_down, v_min, v_max], dtype=np.float64) + y = np.zeros_like(x, dtype=np.float64) y[x < u_min] = 0 y[x >= u_down] = s_max mask = np.logical_and(u_min <= x, x < u_down) @@ -1325,15 +1343,15 @@ def _theoretical_control_data( f"{v_min:.1f}", f"{v_max:.1f}", ], - dtype=object, + dtype=np.object_, ) elif control.type == "q_u": u_min = control._u_min u_down = control._u_down u_up = control._u_up u_max = control._u_max - x = np.array([u_min, u_down, u_up, u_max, v_min, v_max], dtype=float) - y = np.zeros_like(x, dtype=float) + x = np.array([u_min, u_down, u_up, u_max, v_min, v_max], dtype=np.float64) + y = np.zeros_like(x, dtype=np.float64) y[x < u_min] = -s_max mask = np.logical_and(u_min <= x, x < u_down) y[mask] = -s_max * (x[mask] - u_down) / (u_min - u_down) @@ -1350,7 +1368,7 @@ def _theoretical_control_data( f"{v_min:.1f}", f"{v_max:.1f}", ], - dtype=object, + dtype=np.object_, ) else: # pragma: no-cover msg = f"Unsupported control type {control.type!r}" diff --git a/roseau/load_flow/models/loads/loads.py b/roseau/load_flow/models/loads/loads.py index fbe0e582..b178b767 100644 --- a/roseau/load_flow/models/loads/loads.py +++ b/roseau/load_flow/models/loads/loads.py @@ -1,6 +1,5 @@ import logging from abc import ABC -from collections.abc import Sequence from typing import Any, Literal, Optional import numpy as np @@ -10,7 +9,7 @@ from roseau.load_flow.models.buses import Bus from roseau.load_flow.models.core import Element from roseau.load_flow.models.loads.flexible_parameters import FlexibleParameter -from roseau.load_flow.typing import ComplexArray, Id, JsonDict +from roseau.load_flow.typing import ComplexArray, ComplexArrayLike1D, Id, JsonDict from roseau.load_flow.units import Q_, ureg_wraps logger = logging.getLogger(__name__) @@ -104,7 +103,7 @@ def res_currents(self) -> Q_[ComplexArray]: """The load flow result of the load currents (A).""" return self._res_currents_getter(warning=True) - def _validate_value(self, value: Sequence[complex]) -> ComplexArray: + def _validate_value(self, value: ComplexArrayLike1D) -> ComplexArray: if len(value) != self._size: msg = f"Incorrect number of {self._type}s: {len(value)} instead of {self._size}" logger.error(msg) @@ -116,7 +115,7 @@ def _validate_value(self, value: Sequence[complex]) -> ComplexArray: msg = f"An impedance of the load {self.id!r} is null" logger.error(msg) raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_Z_VALUE) - return np.asarray(value, dtype=complex) + return np.array(value, dtype=np.complex128) def _res_potentials_getter(self, warning: bool) -> ComplexArray: self._raise_disconnected_error() @@ -190,7 +189,7 @@ def from_dict(cls, data: JsonDict) -> "AbstractLoad": raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_LOAD_TYPE) def results_from_dict(self, data: JsonDict) -> None: - self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=complex) + self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=np.complex128) def _results_to_dict(self, warning: bool) -> JsonDict: return { @@ -210,7 +209,7 @@ def __init__( id: Id, bus: Bus, *, - powers: Sequence[complex], + powers: ComplexArrayLike1D, phases: Optional[str] = None, flexible_params: Optional[list[FlexibleParameter]] = None, **kwargs: Any, @@ -225,7 +224,8 @@ def __init__( The bus to connect the load to. powers: - List of power for each phase (VA). + An array-like of the powers for each phase component. Either complex values (VA) + or a :class:`Quantity ` of complex values. phases: The phases of the load. A string like ``"abc"`` or ``"an"`` etc. The order of the @@ -272,7 +272,7 @@ def powers(self) -> Q_[ComplexArray]: @powers.setter @ureg_wraps(None, (None, "VA"), strict=False) - def powers(self, value: Sequence[complex]) -> None: + def powers(self, value: ComplexArrayLike1D) -> None: value = self._validate_value(value) if self.is_flexible: for power, fp in zip(value, self._flexible_params): @@ -332,7 +332,7 @@ def to_dict(self, *, _lf_only: bool = False) -> JsonDict: def results_from_dict(self, data: JsonDict) -> None: super().results_from_dict(data=data) if self.is_flexible: - self._res_flexible_powers = np.array([complex(p[0], p[1]) for p in data["powers"]], dtype=complex) + self._res_flexible_powers = np.array([complex(p[0], p[1]) for p in data["powers"]], dtype=np.complex128) def _results_to_dict(self, warning: bool) -> JsonDict: if self.is_flexible: @@ -350,7 +350,7 @@ class CurrentLoad(AbstractLoad): _type = "current" def __init__( - self, id: Id, bus: Bus, *, currents: Sequence[complex], phases: Optional[str] = None, **kwargs: Any + self, id: Id, bus: Bus, *, currents: ComplexArrayLike1D, phases: Optional[str] = None, **kwargs: Any ) -> None: """CurrentLoad constructor. @@ -362,7 +362,8 @@ def __init__( The bus to connect the load to. currents: - List of currents for each phase (Amps). + An array-like of the currents for each phase component. Either complex values (A) + or a :class:`Quantity ` of complex values. phases: The phases of the load. A string like ``"abc"`` or ``"an"`` etc. The order of the @@ -381,7 +382,7 @@ def currents(self) -> Q_[ComplexArray]: @currents.setter @ureg_wraps(None, (None, "A"), strict=False) - def currents(self, value: Sequence[complex]) -> None: + def currents(self, value: ComplexArrayLike1D) -> None: self._currents = self._validate_value(value) self._invalidate_network_results() @@ -401,7 +402,7 @@ class ImpedanceLoad(AbstractLoad): _type = "impedance" def __init__( - self, id: Id, bus: Bus, *, impedances: Sequence[complex], phases: Optional[str] = None, **kwargs: Any + self, id: Id, bus: Bus, *, impedances: ComplexArrayLike1D, phases: Optional[str] = None, **kwargs: Any ) -> None: """ImpedanceLoad constructor. @@ -413,7 +414,8 @@ def __init__( The bus to connect the load to. impedances: - List of impedances for each phase (Ohms). + An array-like of the impedances for each phase component. Either complex values + (Ohms) or a :class:`Quantity ` of complex values. phases: The phases of the load. A string like ``"abc"`` or ``"an"`` etc. The order of the @@ -432,7 +434,7 @@ def impedances(self) -> Q_[ComplexArray]: @impedances.setter @ureg_wraps(None, (None, "ohm"), strict=False) - def impedances(self, impedances: Sequence[complex]) -> None: + def impedances(self, impedances: ComplexArrayLike1D) -> None: self._impedances = self._validate_value(impedances) self._invalidate_network_results() diff --git a/roseau/load_flow/models/sources.py b/roseau/load_flow/models/sources.py index 27492b13..819a78b7 100644 --- a/roseau/load_flow/models/sources.py +++ b/roseau/load_flow/models/sources.py @@ -1,5 +1,4 @@ import logging -from collections.abc import Sequence from typing import Any, Optional import numpy as np @@ -9,7 +8,7 @@ from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode from roseau.load_flow.models.buses import Bus from roseau.load_flow.models.core import Element -from roseau.load_flow.typing import ComplexArray, Id, JsonDict +from roseau.load_flow.typing import ComplexArray, ComplexArrayLike1D, Id, JsonDict from roseau.load_flow.units import Q_, ureg_wraps logger = logging.getLogger(__name__) @@ -23,7 +22,7 @@ class VoltageSource(Element): _floating_neutral_allowed: bool = False def __init__( - self, id: Id, bus: Bus, *, voltages: Sequence[complex], phases: Optional[str] = None, **kwargs: Any + self, id: Id, bus: Bus, *, voltages: ComplexArrayLike1D, phases: Optional[str] = None, **kwargs: Any ) -> None: """Voltage source constructor. @@ -35,9 +34,10 @@ def __init__( The bus of the voltage source. voltages: - The voltages of the source. They will be fixed on the connected bus. If the source - has a neutral connection, the voltages are the phase-to-neutral voltages, otherwise - they are the phase-to-phase voltages. + An array-like of the voltages of the source. They will be set on the connected bus. + If the source has a neutral connection, the voltages are considered phase-to-neutral + voltages, otherwise they are the phase-to-phase voltages. Either complex values (V) + or a :class:`Quantity ` of complex values. phases: The phases of the source. A string like ``"abc"`` or ``"an"`` etc. The order of the @@ -91,12 +91,12 @@ def voltages(self) -> Q_[ComplexArray]: @voltages.setter @ureg_wraps(None, (None, "V"), strict=False) - def voltages(self, voltages: Sequence[complex]) -> None: + def voltages(self, voltages: ComplexArrayLike1D) -> None: if len(voltages) != self._size: msg = f"Incorrect number of voltages: {len(voltages)} instead of {self._size}" logger.error(msg) raise RoseauLoadFlowException(msg, code=RoseauLoadFlowExceptionCode.BAD_VOLTAGES_SIZE) - self._voltages = np.asarray(voltages, dtype=complex) + self._voltages = np.array(voltages, dtype=np.complex128) self._invalidate_network_results() @property @@ -167,7 +167,7 @@ def to_dict(self, *, _lf_only: bool = False) -> JsonDict: } def results_from_dict(self, data: JsonDict) -> None: - self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=complex) + self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=np.complex128) def _results_to_dict(self, warning: bool) -> JsonDict: return { diff --git a/roseau/load_flow/models/tests/test_branches.py b/roseau/load_flow/models/tests/test_branches.py index 1b0bebf6..72f7829e 100644 --- a/roseau/load_flow/models/tests/test_branches.py +++ b/roseau/load_flow/models/tests/test_branches.py @@ -225,9 +225,9 @@ def test_powers_equal(network_with_results): def test_lines_results(phases, z_line, y_shunt, len_line, bus_pot, line_cur, ground_pot, expected_pow): bus1 = Bus("bus1", phases=phases["bus1"]) bus2 = Bus("bus2", phases=phases["bus2"]) - y_shunt = np.asarray(y_shunt, dtype=complex) if y_shunt is not None else None + y_shunt = np.array(y_shunt, dtype=np.complex128) if y_shunt is not None else None ground = Ground("gnd") - lp = LineParameters("lp", z_line=np.asarray(z_line, dtype=complex), y_shunt=y_shunt) + lp = LineParameters("lp", z_line=np.array(z_line, dtype=np.complex128), y_shunt=y_shunt) line = Line( "line", bus1, diff --git a/roseau/load_flow/models/transformers/parameters.py b/roseau/load_flow/models/transformers/parameters.py index 9c68e541..2e389a0a 100644 --- a/roseau/load_flow/models/transformers/parameters.py +++ b/roseau/load_flow/models/transformers/parameters.py @@ -42,14 +42,14 @@ def __init__( self, id: Id, type: str, - uhv: float, - ulv: float, - sn: float, - p0: float, - i0: float, - psc: float, - vsc: float, - max_power: Optional[float] = None, + uhv: Union[float, Q_[float]], + ulv: Union[float, Q_[float]], + sn: Union[float, Q_[float]], + p0: Union[float, Q_[float]], + i0: Union[float, Q_[float]], + psc: Union[float, Q_[float]], + vsc: Union[float, Q_[float]], + max_power: Optional[Union[float, Q_[float]]] = None, ) -> None: """TransformerParameters constructor. @@ -204,7 +204,7 @@ def max_power(self) -> Optional[Q_[float]]: @max_power.setter @ureg_wraps(None, (None, "VA"), strict=False) - def max_power(self, value: Optional[float]) -> None: + def max_power(self, value: Optional[Union[float, Q_[float]]]) -> None: self._max_power = value @ureg_wraps(("ohm", "S", "", None), (None,), strict=False) diff --git a/roseau/load_flow/network.py b/roseau/load_flow/network.py index 73eabe17..72c2d162 100644 --- a/roseau/load_flow/network.py +++ b/roseau/load_flow/network.py @@ -6,7 +6,7 @@ import re import textwrap import warnings -from collections.abc import Sized +from collections.abc import Mapping, Sized from importlib import resources from itertools import cycle from pathlib import Path @@ -37,7 +37,7 @@ VoltageSource, ) from roseau.load_flow.solvers import check_solver_params -from roseau.load_flow.typing import Authentication, Id, JsonDict, Solver, StrPath +from roseau.load_flow.typing import Authentication, Id, JsonDict, MapOrSeq, Solver, StrPath from roseau.load_flow.utils import CatalogueMixin, JsonMixin, _optional_deps, console, palette from roseau.load_flow.utils.types import _DTYPES, VoltagePhaseDtype @@ -46,7 +46,7 @@ logger = logging.getLogger(__name__) -_T = TypeVar("_T", bound=Element) +_E = TypeVar("_E", bound=Element) class ElectricalNetwork(JsonMixin, CatalogueMixin[JsonDict]): @@ -153,12 +153,12 @@ class ElectricalNetwork(JsonMixin, CatalogueMixin[JsonDict]): # def __init__( self, - buses: Union[list[Bus], dict[Id, Bus]], - branches: Union[list[AbstractBranch], dict[Id, AbstractBranch]], - loads: Union[list[AbstractLoad], dict[Id, AbstractLoad]], - sources: Union[list[VoltageSource], dict[Id, VoltageSource]], - grounds: Union[list[Ground], dict[Id, Ground]], - potential_refs: Union[list[PotentialRef], dict[Id, PotentialRef]], + buses: MapOrSeq[Bus], + branches: MapOrSeq[AbstractBranch], + loads: MapOrSeq[AbstractLoad], + sources: MapOrSeq[VoltageSource], + grounds: MapOrSeq[Ground], + potential_refs: MapOrSeq[PotentialRef], **kwargs, ) -> None: self.buses = self._elements_as_dict(buses, RoseauLoadFlowExceptionCode.BAD_BUS_ID) @@ -194,25 +194,24 @@ def count_repr(__o: Sized, /, singular: str, plural: Optional[str] = None) -> st ) @staticmethod - def _elements_as_dict( - elements: Union[list[_T], dict[Id, _T]], error_code: RoseauLoadFlowExceptionCode - ) -> dict[Id, _T]: - """Convert a list of elements to a dictionary of elements with their IDs as keys.""" + def _elements_as_dict(elements: MapOrSeq[_E], error_code: RoseauLoadFlowExceptionCode) -> dict[Id, _E]: + """Convert a sequence or a mapping of elements to a dictionary of elements with their IDs as keys.""" typ = error_code.name.removeprefix("BAD_").removesuffix("_ID").replace("_", " ") - if isinstance(elements, dict): + elements_dict: dict[Id, _E] = {} + if isinstance(elements, Mapping): for element_id, element in elements.items(): if element.id != element_id: msg = f"{typ.capitalize()} ID mismatch: {element_id!r} != {element.id!r}." logger.error(msg) raise RoseauLoadFlowException(msg, code=error_code) - return elements - elements_dict: dict[Id, _T] = {} - for element in elements: - if element.id in elements_dict: - msg = f"Duplicate ID for an {typ.lower()} in this network: {element.id!r}." - logger.error(msg) - raise RoseauLoadFlowException(msg, code=error_code) - elements_dict[element.id] = element + elements_dict[element_id] = element + else: + for element in elements: + if element.id in elements_dict: + msg = f"Duplicate ID for an {typ.lower()} in this network: {element.id!r}." + logger.error(msg) + raise RoseauLoadFlowException(msg, code=error_code) + elements_dict[element.id] = element return elements_dict @classmethod diff --git a/roseau/load_flow/typing.py b/roseau/load_flow/typing.py index 31a5cb33..fc202af2 100644 --- a/roseau/load_flow/typing.py +++ b/roseau/load_flow/typing.py @@ -1,9 +1,14 @@ """ Type Aliases used by Roseau Load Flow. +.. warning:: + + Types defined in this module are not part of the public API. You can use these types in your + code, but they are not guaranteed to be stable. + .. class:: Id - The type of the identifier of an element. + The type of the identifier of an element. An element's ID can be an integer or a string. .. class:: JsonDict @@ -11,15 +16,15 @@ .. class:: StrPath - The accepted type for files of roseau.load_flow.io. + The accepted type for file paths in roseau.load_flow. This is a string or a path-like object. .. class:: ControlType - Available types of control for flexible loads. + Available control types for flexible loads. .. class:: ProjectionType - Available types of projections for flexible loads control. + Available projections types for flexible loads control. .. class:: Solver @@ -27,20 +32,39 @@ .. class:: Authentication - Valid authentication types. + Valid authentication types used to connect to the Roseau Load Flow solver API. + +.. class:: MapOrSeq + + A mapping from element IDs to elements or a sequence of elements of unique IDs. .. class:: ComplexArray A numpy array of complex numbers. + +.. class:: ComplexArrayLike1D + + A 1D array-like of complex numbers or a quantity of complex numbers. An array-like is a + sequence or a numpy array. + +.. class:: ComplexArrayLike2D + + A 2D array-like of complex numbers or a quantity of complex numbers. An array-like is a + sequence or a numpy array. """ import os -from typing import Any, Literal, Union +from collections.abc import Mapping, Sequence +from typing import Any, Literal, TypeVar, Union import numpy as np from numpy.typing import NDArray from requests.auth import HTTPBasicAuth from typing_extensions import TypeAlias +from roseau.load_flow.units import Q_ + +T = TypeVar("T") + Id: TypeAlias = Union[int, str] JsonDict: TypeAlias = dict[str, Any] StrPath: TypeAlias = Union[str, os.PathLike[str]] @@ -48,7 +72,21 @@ ProjectionType: TypeAlias = Literal["euclidean", "keep_p", "keep_q"] Solver: TypeAlias = Literal["newton", "newton_goldstein"] Authentication: TypeAlias = Union[tuple[str, str], HTTPBasicAuth] -ComplexArray: TypeAlias = NDArray[np.complex_] +MapOrSeq: TypeAlias = Union[Mapping[Id, T], Sequence[T]] +ComplexArray: TypeAlias = NDArray[np.complex128] +# TODO: improve the types below when shape-typing becomes supported +ComplexArrayLike1D: TypeAlias = Union[ + ComplexArray, + Q_[ComplexArray], + Q_[Sequence[complex]], + Sequence[Union[complex, Q_[complex]]], +] +ComplexArrayLike2D: TypeAlias = Union[ + ComplexArray, + Q_[ComplexArray], + Q_[Sequence[Sequence[complex]]], + Sequence[Sequence[Union[complex, Q_[complex]]]], +] __all__ = [ @@ -59,5 +97,8 @@ "ProjectionType", "Solver", "Authentication", + "MapOrSeq", "ComplexArray", + "ComplexArrayLike1D", + "ComplexArrayLike2D", ] diff --git a/roseau/load_flow/units.py b/roseau/load_flow/units.py index 0c497a44..a1e5ec51 100644 --- a/roseau/load_flow/units.py +++ b/roseau/load_flow/units.py @@ -3,11 +3,20 @@ .. class:: ureg - The :class:`~pint.UnitRegistry` object to use in this project. + The :class:`pint.UnitRegistry` object to use in this project. You should not need to use it + directly. .. class:: Q_ - The :class:`~pint.Quantity` class to use in this project. + The :class:`pint.Quantity` class to use in this project. You can use it to provide quantities + in units different from the default ones. For example, to create a constant power load of 1 MVA, + you can do: + + >>> load = lf.PowerLoad("load", bus=bus, powers=Q_([1, 1, 1], "MVA")) + + which is equivalent to: + + >>> load = lf.PowerLoad("load", bus=bus, powers=[1000000, 1000000, 1000000]) # in VA .. _pint: https://pint.readthedocs.io/en/stable/getting/overview.html """