Un-deprecate results_to_dict and remove results_from_dict

doc/Changelog.md

@@ -15,6 +15,10 @@ myst:
 
 ## Unreleased
 
+- {gh-pr}`206` {gh-issue}`187` Un-deprecate `results_to_dict/json` methods and remove deprecated
+  `results_from_dict/json` methods.
+- {gh-pr}`205` {gh-issue}`200` Fix error when propagating the potentials from a voltage source with fewer phases
+  than the bus.
 - {gh-pr}`204` {gh-issue}`193` Remove restrictions on geometry types. Allow specifying the CRS of the geometries.
 - {gh-pr}`203` {gh-issue}`186` Detect invalid element overrides when connecting a new element with the
   same ID and type of an existing element.

roseau/load_flow/models/branches.py

@@ -2,7 +2,6 @@
 from functools import cached_property
 from typing import ClassVar, Literal
 
-import numpy as np
 from shapely.geometry.base import BaseGeometry
 from typing_extensions import Self
 
@@ -187,13 +186,6 @@ def _to_dict(self, include_results: bool) -> 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=np.complex128)
-        currents2 = np.array([complex(i[0], i[1]) for i in data["currents2"]], dtype=np.complex128)
-        self._res_currents = (currents1, currents2)
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         currents1, currents2 = self._res_currents_getter(warning)
         return {

roseau/load_flow/models/buses.py

@@ -366,11 +366,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"] = {"potentials": [[v.real, v.imag] for v in potentials]}
         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=np.complex128)
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         return {
             "id": self.id,

roseau/load_flow/models/grounds.py

@@ -117,11 +117,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"] = {"potential": [v.real, v.imag]}
         return res
 
-    def _results_from_dict(self, data: JsonDict) -> None:
-        self._res_potential = complex(*data["potential"])
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         v = self._res_potential_getter(warning)
         return {"id": self.id, "potential": [v.real, v.imag]}

roseau/load_flow/models/lines/parameters.py

@@ -1018,11 +1018,6 @@ def _results_to_dict(self, warning: bool) -> NoReturn:
         logger.error(msg)
         raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
 
-    def _results_from_dict(self, data: JsonDict) -> None:
-        msg = f"The {type(self).__name__} has no results to import."
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
-
     #
     # Utility
     #

roseau/load_flow/models/loads/flexible_parameters.py

@@ -405,11 +405,6 @@ def _results_to_dict(self, warning: bool) -> NoReturn:
         logger.error(msg)
         raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
 
-    def _results_from_dict(self, data: JsonDict) -> NoReturn:
-        msg = f"The {type(self).__name__} has no results to import."
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
-
 
 class Projection(JsonMixin):
     """This class defines the projection on the feasible circle for a flexible load.
@@ -500,11 +495,6 @@ def _results_to_dict(self, warning: bool) -> NoReturn:
         logger.error(msg)
         raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
 
-    def _results_from_dict(self, data: JsonDict) -> NoReturn:
-        msg = f"The {type(self).__name__} has no results to import."
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
-
 
 class FlexibleParameter(JsonMixin):
     """Flexible parameters of a flexible load.
@@ -1089,11 +1079,6 @@ def _results_to_dict(self, warning: bool) -> NoReturn:
         logger.error(msg)
         raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
 
-    def _results_from_dict(self, data: JsonDict) -> NoReturn:
-        msg = f"The {type(self).__name__} has no results to import."
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
-
     #
     # Equivalent Python method
     #

roseau/load_flow/models/loads/loads.py

@@ -234,11 +234,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"] = {"currents": [[i.real, i.imag] for i in currents]}
         return res
 
-    def _results_from_dict(self, data: JsonDict) -> None:
-        self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=np.complex128)
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         return {
             "id": self.id,
@@ -411,11 +406,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"]["powers"] = [[s.real, s.imag] for s in flexible_powers]
         return res
 
-    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=np.complex128)
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         if self.is_flexible:
             return {

roseau/load_flow/models/potential_refs.py

@@ -123,11 +123,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"] = {"current": [i.real, i.imag]}
         return res
 
-    def _results_from_dict(self, data: JsonDict) -> None:
-        self._res_current = complex(*data["current"])
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         i = self._res_current_getter(warning)
         return {"id": self.id, "current": [i.real, i.imag]}

roseau/load_flow/models/sources.py

@@ -206,11 +206,6 @@ def _to_dict(self, include_results: bool) -> JsonDict:
             res["results"] = {"currents": [[i.real, i.imag] for i in currents]}
         return res
 
-    def _results_from_dict(self, data: JsonDict) -> None:
-        self._res_currents = np.array([complex(i[0], i[1]) for i in data["currents"]], dtype=np.complex128)
-        self._fetch_results = False
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         return {
             "id": self.id,

roseau/load_flow/models/transformers/parameters.py

@@ -317,11 +317,6 @@ def _results_to_dict(self, warning: bool) -> NoReturn:
         logger.error(msg)
         raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
 
-    def _results_from_dict(self, data: JsonDict) -> NoReturn:
-        msg = f"The {type(self).__name__} has no results to import."
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.JSON_NO_RESULTS)
-
     #
     # Catalogue Mixin
     #

roseau/load_flow/network.py

@@ -1425,66 +1425,8 @@ def _to_dict(self, include_results: bool) -> JsonDict:
         return network_to_dict(self, include_results=include_results)
 
     #
-    # Results saving/loading
+    # Results saving
     #
-    def _results_from_dict(self, data: JsonDict) -> None:
-        """Load the results of a load flow from a dict created by :meth:`results_to_dict`.
-
-        The results are stored in the network elements.
-
-        Args:
-            data:
-                The dictionary containing the load flow results.
-        """
-        # Checks on the provided data
-        for key, self_elements, name in (
-            ("buses", self.buses, "Bus"),
-            ("branches", self.branches, "Branch"),
-            ("loads", self.loads, "Load"),
-            ("sources", self.sources, "Source"),
-            ("grounds", self.grounds, "Ground"),
-            ("potential_refs", self.potential_refs, "PotentialRef"),
-        ):
-            seen = set()
-            for element_data in data[key]:
-                element_id = element_data["id"]
-                if element_id not in self_elements:
-                    msg = f"{name} {element_id!r} appears in the results but is not present in the network."
-                    logger.error(msg)
-                    raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_LOAD_FLOW_RESULT)
-                seen.add(element_id)
-            if missing_elements := self_elements.keys() - seen:
-                msg = (
-                    f"The following {key} are present in the network but not in the results: "
-                    f"{sorted(missing_elements)}."
-                )
-                logger.error(msg)
-                raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.BAD_LOAD_FLOW_RESULT)
-
-        # The results are assigned to all elements
-        for bus_data in data["buses"]:
-            bus = self.buses[bus_data["id"]]
-            bus._results_from_dict(bus_data)
-        for branch_data in data["branches"]:
-            branch = self.branches[branch_data["id"]]
-            branch._results_from_dict(branch_data)
-        for load_data in data["loads"]:
-            load = self.loads[load_data["id"]]
-            load._results_from_dict(load_data)
-        for source_data in data["sources"]:
-            source = self.sources[source_data["id"]]
-            source._results_from_dict(data=source_data)
-        for ground_data in data["grounds"]:
-            ground = self.grounds[ground_data["id"]]
-            ground._results_from_dict(ground_data)
-        for p_ref_data in data["potential_refs"]:
-            p_ref = self.potential_refs[p_ref_data["id"]]
-            p_ref._results_from_dict(p_ref_data)
-
-        # The results are now valid
-        self._results_valid = True
-        self._no_results = False
-
     def _results_to_dict(self, warning: bool) -> JsonDict:
         """Get the voltages and currents computed by the load flow and return them as a dict."""
         if warning:

roseau/load_flow/tests/test_electrical_network.py

@@ -1,5 +1,6 @@
 import contextlib
 import itertools as it
+import json
 import re
 import warnings
 from contextlib import contextmanager
@@ -1852,7 +1853,7 @@ def assert_results(en_dict: dict, included: bool):
     assert_results(en_dict_with_results, included=True)
     assert_results(en_dict_without_results, included=False)
     assert en_dict_with_results != en_dict_without_results
-    # round triping
+    # round tripping
     assert ElectricalNetwork.from_dict(en_dict_with_results).to_dict() == en_dict_with_results
     assert ElectricalNetwork.from_dict(en_dict_without_results).to_dict() == en_dict_without_results
     # default is to include the results
@@ -1868,47 +1869,60 @@ def assert_results(en_dict: dict, included: bool):
     )
     assert e.value.code == RoseauLoadFlowExceptionCode.BAD_LOAD_FLOW_RESULT
     en_dict_without_results = en.to_dict(include_results=False)
-    # round triping without the results should still work
+    # round tripping without the results should still work
     assert ElectricalNetwork.from_dict(en_dict_without_results).to_dict() == en_dict_without_results
 
 
-def test_deprecated_results_methods(small_network_with_results, tmp_path):
+def test_results_to_dict(small_network_with_results):
     en = small_network_with_results
-    en_dict_res = en.to_dict(include_results=True)
-    en_dict_no_res = en.to_dict(include_results=False)
-
-    with pytest.warns(DeprecationWarning) as record:
-        res_dict = en.results_to_dict()
-    assert len(record) == 1
-    assert record[0].message.args[0] == (
-        "Method `results_to_dict()` is deprecated. Method `to_dict()` now includes the results by default."
-    )
-
-    new_en = ElectricalNetwork.from_dict(en_dict_no_res)
-    with pytest.warns(DeprecationWarning) as record:
-        new_en.results_from_dict(res_dict)
-    assert len(record) == 1
-    assert record[0].message.args[0] == (
-        "Method `results_from_dict()` is deprecated. Method `from_dict()` now includes the results by default."
-    )
-    assert new_en.to_dict() == en_dict_res
 
+    res_network = en.results_to_dict()
+    assert set(res_network) == {"buses", "branches", "loads", "sources", "grounds", "potential_refs"}
+    for v in res_network.values():
+        assert isinstance(v, list)
+    for res_bus in res_network["buses"]:
+        bus = en.buses[res_bus["id"]]
+        assert res_bus["phases"] == bus.phases
+        complex_potentials = [v_r + 1j * v_i for v_r, v_i in res_bus["potentials"]]
+        np.testing.assert_allclose(complex_potentials, bus.res_potentials.m)
+    for res_branch in res_network["branches"]:
+        branch = en.branches[res_branch["id"]]
+        assert res_branch["phases1"] == branch.phases1
+        assert res_branch["phases2"] == branch.phases2
+        complex_currents1 = [i_r + 1j * i_i for i_r, i_i in res_branch["currents1"]]
+        np.testing.assert_allclose(complex_currents1, branch.res_currents[0].m)
+        complex_currents2 = [i_r + 1j * i_i for i_r, i_i in res_branch["currents2"]]
+        np.testing.assert_allclose(complex_currents2, branch.res_currents[1].m)
+    for res_load in res_network["loads"]:
+        load = en.loads[res_load["id"]]
+        assert res_load["phases"] == load.phases
+        complex_currents = [i_r + 1j * i_i for i_r, i_i in res_load["currents"]]
+        np.testing.assert_allclose(complex_currents, load.res_currents.m)
+    for res_source in res_network["sources"]:
+        source = en.sources[res_source["id"]]
+        assert res_source["phases"] == source.phases
+        complex_currents = [i_r + 1j * i_i for i_r, i_i in res_source["currents"]]
+        np.testing.assert_allclose(complex_currents, source.res_currents.m)
+    for res_ground in res_network["grounds"]:
+        ground = en.grounds[res_ground["id"]]
+        complex_potential = complex(*res_ground["potential"])
+        np.testing.assert_allclose(complex_potential, ground.res_potential.m)
+    for res_potential_ref in res_network["potential_refs"]:
+        potential_ref = en.potential_refs[res_potential_ref["id"]]
+        complex_current = complex(*res_potential_ref["current"])
+        np.testing.assert_allclose(complex_current, potential_ref.res_current.m)
+
+
+def test_results_to_json(small_network_with_results, tmp_path):
+    en = small_network_with_results
+    res_network_expected = en.results_to_dict()
     tmp_file = tmp_path / "results.json"
-    with pytest.warns(DeprecationWarning) as record:
-        en.results_to_json(tmp_file)
-    assert len(record) == 1
-    assert record[0].message.args[0] == (
-        "Method `results_to_json()` is deprecated. Method `to_json()` now includes the results by default."
-    )
+    en.results_to_json(tmp_file)
 
-    new_en = ElectricalNetwork.from_dict(en_dict_no_res)
-    with pytest.warns(DeprecationWarning) as record:
-        new_en.results_from_json(tmp_file)
-    assert len(record) == 1
-    assert record[0].message.args[0] == (
-        "Method `results_from_json()` is deprecated. Method `from_json()` now includes the results by default."
-    )
-    assert new_en.to_dict() == en_dict_res
+    with open(tmp_file) as fp:
+        res_network = json.load(fp)
+
+    assert res_network == res_network_expected
 
 
 def test_propagate_potentials_center_transformers():