Fix JSON serialization, flexible parameters, and lines

roseau/load_flow_single/io/dict.py

@@ -8,7 +8,7 @@
 
 import copy
 import logging
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING
 
 from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode
 from roseau.load_flow.io.dict import NETWORK_JSON_VERSION
@@ -31,11 +31,8 @@
 logger = logging.getLogger(__name__)
 
 
-_T = TypeVar("_T", bound=AbstractBranch)
-
-
-def _assign_branch_currents(branch: _T, branch_data: JsonDict) -> _T:
-    """Small helper to assign the currents results to a branch object.
+def _assign_branch_currents(branch: AbstractBranch, branch_data: JsonDict):
+    """Assign the currents results to a branch object.
 
     Args:
         branch:
@@ -56,8 +53,6 @@ def _assign_branch_currents(branch: _T, branch_data: JsonDict) -> _T:
         branch._fetch_results = False
         branch._no_results = False
 
-    return branch
-
 
 def network_from_dict(
     data: JsonDict, *, include_results: bool = True
@@ -81,23 +76,19 @@ def network_from_dict(
             the results are also loaded into the network.
 
     Returns:
-        The buses, lines, transformers, switches, loads, sources, grounds and potential refs to construct the electrical
+        The buses, lines, transformers, switches, loads, and sources to construct the electrical
         network and a boolean indicating if the network has results.
     """
     data = copy.deepcopy(data)  # Make a copy to avoid modifying the original
 
     # Check that the network is single phase
     is_multiphase = data.get("is_multiphase", True)
-    assert not is_multiphase, f"Unsupported phase selection {is_multiphase=}."
+    assert not is_multiphase, f"Unsupported {is_multiphase=} for single-phase network."
 
-    # Check on the version
-    # 3 was the first version to support RLFS
+    # Check the version, 3 was the first version to support RLFS
     version = data.get("version", 3)
     if version <= 2:
-        msg = (
-            f"The version {version} of the network file can not be single-phased. This featured appeared in the "
-            f"version 3..."
-        )
+        msg = f"Version {version} of the network file cannot be single-phase. Minimum single-phase version is 3."
         logger.error(msg)
         raise AssertionError(msg)
     # elif version <= NETWORK_JSON_VERSION:
@@ -148,12 +139,15 @@ def network_from_dict(
         id = line_data["id"]
         bus1 = buses[line_data["bus1"]]
         bus2 = buses[line_data["bus2"]]
-        geometry = Line._parse_geometry(line_data.get("geometry"))
         length = line_data["length"]
+        max_loading = line_data["max_loading"]
+        geometry = Line._parse_geometry(line_data.get("geometry"))
         lp = lines_params[line_data["params_id"]]
-        line = Line(id=id, bus1=bus1, bus2=bus2, parameters=lp, length=length, geometry=geometry)
+        line = Line(
+            id=id, bus1=bus1, bus2=bus2, parameters=lp, length=length, max_loading=max_loading, geometry=geometry
+        )
         if include_results:
-            line = _assign_branch_currents(branch=line, branch_data=line_data)
+            _assign_branch_currents(branch=line, branch_data=line_data)
 
         has_results = has_results and not line._no_results
         lines_dict[id] = line
@@ -164,11 +158,15 @@ def network_from_dict(
         id = transformer_data["id"]
         bus1 = buses[transformer_data["bus1"]]
         bus2 = buses[transformer_data["bus2"]]
+        tap = transformer_data["tap"]
+        max_loading = transformer_data["max_loading"]
         geometry = Transformer._parse_geometry(transformer_data.get("geometry"))
         tp = transformers_params[transformer_data["params_id"]]
-        transformer = Transformer(id=id, bus1=bus1, bus2=bus2, parameters=tp, geometry=geometry)
+        transformer = Transformer(
+            id=id, bus1=bus1, bus2=bus2, parameters=tp, tap=tap, max_loading=max_loading, geometry=geometry
+        )
         if include_results:
-            transformer = _assign_branch_currents(branch=transformer, branch_data=transformer_data)
+            _assign_branch_currents(branch=transformer, branch_data=transformer_data)
 
         has_results = has_results and not transformer._no_results
         transformers_dict[id] = transformer
@@ -182,7 +180,7 @@ def network_from_dict(
         geometry = Switch._parse_geometry(switch_data.get("geometry"))
         switch = Switch(id=id, bus1=bus1, bus2=bus2, geometry=geometry)
         if include_results:
-            switch = _assign_branch_currents(branch=switch, branch_data=switch_data)
+            _assign_branch_currents(branch=switch, branch_data=switch_data)
 
         has_results = has_results and not switch._no_results
         switches_dict[id] = switch
@@ -252,13 +250,13 @@ def network_to_dict(en: "ElectricalNetwork", *, include_results: bool) -> JsonDi
     line_params: list[JsonDict] = []
     for lp in lines_params_dict.values():
         line_params.append(lp.to_dict(include_results=include_results))
-    line_params.sort(key=lambda x: x["id"])  # Always keep the same order
+    line_params.sort(key=lambda x: (type(x["id"]).__name__, str(x["id"])))  # Always keep the same order
 
     # Transformer parameters
     transformer_params: list[JsonDict] = []
     for tp in transformers_params_dict.values():
         transformer_params.append(tp.to_dict(include_results=include_results))
-    transformer_params.sort(key=lambda x: x["id"])  # Always keep the same order
+    transformer_params.sort(key=lambda x: (type(x["id"]).__name__, str(x["id"])))  # Always keep the same order
 
     res = {
         "version": NETWORK_JSON_VERSION,

roseau/load_flow_single/models/lines.py

@@ -181,7 +181,7 @@ def max_current(self) -> Q_[float] | None:
         `ampacity` of the parameters."""
         # Do not add a setter. Only `max_loading` can be altered by the user
         amp = self._parameters.ampacity
-        return None if amp is None else Q_(amp[0] * self._max_loading, "A")
+        return None if amp is None else Q_(amp * self._max_loading, "A")
 
     @property
     def with_shunt(self) -> bool:

roseau/load_flow_single/models/loads.py

@@ -4,13 +4,13 @@
 
 import numpy as np
 
-from roseau.load_flow import FlexibleParameter
 from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode
 from roseau.load_flow.typing import Complex, Id, JsonDict
 from roseau.load_flow.units import Q_, ureg_wraps
 from roseau.load_flow_engine.cy_engine import CyAdmittanceLoad, CyCurrentLoad, CyFlexibleLoad, CyPowerLoad
 from roseau.load_flow_single.models.buses import Bus
 from roseau.load_flow_single.models.core import Element
+from roseau.load_flow_single.models.flexible_parameters import FlexibleParameter
 
 logger = logging.getLogger(__name__)
 

roseau/load_flow_single/models/switches.py

@@ -21,13 +21,13 @@ def __init__(self, id: Id, bus1: Bus, bus2: Bus, *, geometry: BaseGeometry | Non
 
         Args:
             id:
-                A unique ID of the switch in the network branches.
+                A unique ID of the switch in the network switches.
 
             bus1:
-                Bus to connect to the switch.
+                First bus to connect to the switch.
 
             bus2:
-                Bus to connect to the switch.
+                Second bus to connect to the switch.
 
             geometry:
                 The geometry of the switch.
@@ -39,7 +39,7 @@ def __init__(self, id: Id, bus1: Bus, bus2: Bus, *, geometry: BaseGeometry | Non
         self._cy_connect()
 
     def _check_loop(self) -> None:
-        """Check that there are no switch loop, raise an exception if it is the case"""
+        """Check that there are no switch loops, raise an exception if it is the case."""
         visited_1: set[Element] = set()
         elements: list[Element] = [self.bus1]
         while elements:
@@ -57,7 +57,11 @@ def _check_loop(self) -> None:
                 if e not in visited_2 and (isinstance(e, (Bus, Switch))) and e != self:
                     elements.append(e)
         if visited_1.intersection(visited_2):
-            msg = f"There is a loop of switch involving the switch {self.id!r}. It is not allowed."
+            msg = (
+                f"Connecting switch {self.id!r} between buses {self.bus1.id!r} and {self.bus2.id!r} "
+                f"creates a switch loop. Current flow in several switch-only branches between buses "
+                f"cannot be computed."
+            )
             logger.error(msg)
             raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.SWITCHES_LOOP)
 
@@ -67,8 +71,8 @@ def _check_elements(self) -> None:
             isinstance(e, VoltageSource) for e in self.bus2._connected_elements
         ):
             msg = (
-                f"The buses {self.bus1.id!r} and {self.bus2.id!r} both have a voltage source and "
-                f"are connected with the switch {self.id!r}. It is not allowed."
+                f"Connecting switch {self.id!r} between buses {self.bus1.id!r} and {self.bus2.id!r} "
+                f"that both have a voltage source is not allowed."
             )
             logger.error(msg)
             raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_VOLTAGES_SOURCES_CONNECTION)

roseau/load_flow_single/models/tests/test_flexible_parameters.py

@@ -2,7 +2,7 @@
 import pytest
 
 from roseau.load_flow import Q_, RoseauLoadFlowException, RoseauLoadFlowExceptionCode
-from roseau.load_flow_single import Control, FlexibleParameter, Projection
+from roseau.load_flow_single.models import Control, FlexibleParameter, Projection
 
 
 def test_control():

roseau/load_flow_single/models/tests/test_switch.py → roseau/load_flow_single/models/tests/test_switches.py

@@ -15,18 +15,21 @@ def test_switch_loop():
 
     with pytest.raises(RoseauLoadFlowException) as e:
         Switch(id="switch2", bus1=bus1, bus2=bus2)
-    assert "There is a loop of switch" in e.value.msg
+    assert e.value.msg == (
+        "Connecting switch 'switch2' between buses 'bus1' and 'bus2' creates a switch loop. Current "
+        "flow in several switch-only branches between buses cannot be computed."
+    )
     assert e.value.code == RoseauLoadFlowExceptionCode.SWITCHES_LOOP
 
     with pytest.raises(RoseauLoadFlowException) as e:
         Switch(id="switch3", bus1=bus2, bus2=bus1)
-    assert "There is a loop of switch" in e.value.msg
+    assert "Connecting switch 'switch3' between buses 'bus2' and 'bus1' creates a switch loop." in e.value.msg
     assert e.value.code == RoseauLoadFlowExceptionCode.SWITCHES_LOOP
 
     Switch(id="switch4", bus1=bus2, bus2=bus3)
     with pytest.raises(RoseauLoadFlowException) as e:
         Switch(id="switch5", bus1=bus1, bus2=bus3)
-    assert "There is a loop of switch" in e.value.msg
+    assert "Connecting switch 'switch5' between buses 'bus1' and 'bus3' creates a switch loop." in e.value.msg
     assert e.value.code == RoseauLoadFlowExceptionCode.SWITCHES_LOOP
 
 
@@ -37,5 +40,7 @@ def test_switch_connection():
     VoltageSource(id="vs2", bus=bus2, voltage=230 + 0j)
     with pytest.raises(RoseauLoadFlowException) as e:
         Switch(id="switch", bus1=bus1, bus2=bus2)
-    assert "are connected with the switch" in e.value.msg
+    assert e.value.msg == (
+        "Connecting switch 'switch' between buses 'bus1' and 'bus2' that both have a voltage source is not allowed."
+    )
     assert e.value.code == RoseauLoadFlowExceptionCode.BAD_VOLTAGES_SOURCES_CONNECTION

roseau/load_flow_single/models/transformers.py

@@ -63,39 +63,37 @@ def __init__(
         self._parameters = parameters
         self.max_loading = max_loading
 
-        if parameters.type == "center-tapped":
-            msg = "Center-tapped transformers are not allowed."
+        if parameters.type != "three-phase":
+            msg = f"{parameters.type.capitalize()} transformers are not allowed in a balanced three-phase load flow."
             logger.error(msg)
-            raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_TRANSFORMER_WINDINGS)
-        elif parameters.type == "single-phase":
-            z2, ym, k_single = parameters._z2, parameters._ym, parameters._k * parameters._orientation
-        else:
-            z2, ym = parameters._z2, parameters._ym
-            k_complex_factor = {
-                ("D", "d", 0): 1,
-                ("Y", "y", 0): 1,
-                ("D", "z", 0): 3,
-                ("D", "d", 6): -1,
-                ("Y", "y", 6): -1,
-                ("D", "z", 6): -3,
-                ("D", "y", 1): 1 - ALPHA,
-                ("Y", "z", 1): 1 - ALPHA,
-                ("Y", "d", 1): 1 / (1 - ALPHA2),
-                ("D", "y", 5): ALPHA2 - 1,
-                ("Y", "z", 5): ALPHA2 - 1,
-                ("Y", "d", 5): 1 / (ALPHA - 1),
-                ("D", "y", 11): 1 - ALPHA2,
-                ("Y", "z", 11): 1 - ALPHA2,
-                ("Y", "d", 11): 1 / (1 - ALPHA),
-            }
-            k_single = (
-                parameters._k
-                * k_complex_factor[parameters.winding1[0], parameters.winding2[0], parameters.phase_displacement]
-            )
-            if parameters.winding1.startswith("D"):
-                ym *= 3.0
-            if parameters.winding2.startswith("d"):
-                z2 /= 3.0
+            raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_TRANSFORMER_TYPE)
+
+        z2, ym = parameters._z2, parameters._ym
+        k_complex_factor = {
+            ("D", "d", 0): 1,
+            ("Y", "y", 0): 1,
+            ("D", "z", 0): 3,
+            ("D", "d", 6): -1,
+            ("Y", "y", 6): -1,
+            ("D", "z", 6): -3,
+            ("D", "y", 1): 1 - ALPHA,
+            ("Y", "z", 1): 1 - ALPHA,
+            ("Y", "d", 1): 1 / (1 - ALPHA2),
+            ("D", "y", 5): ALPHA2 - 1,
+            ("Y", "z", 5): ALPHA2 - 1,
+            ("Y", "d", 5): 1 / (ALPHA - 1),
+            ("D", "y", 11): 1 - ALPHA2,
+            ("Y", "z", 11): 1 - ALPHA2,
+            ("Y", "d", 11): 1 / (1 - ALPHA),
+        }
+        k_single = (
+            parameters._k
+            * k_complex_factor[parameters.winding1[0], parameters.winding2[0], parameters.phase_displacement]
+        )
+        if parameters.winding1.startswith("D"):
+            ym *= 3.0
+        if parameters.winding2.startswith("d"):
+            z2 /= 3.0
 
         self._cy_element = CySingleTransformer(z2=z2, ym=ym, k=k_single * tap)
         self._cy_connect()