Implement networkx graphs

conda/environment.yml

@@ -14,3 +14,4 @@ dependencies:
   - typing-extensions >=4.6.2
   - rich >=13.5.1
   - matplotlib >=3.7.2
+  - networkx >=3.0.0

conda/meta.yaml

@@ -36,6 +36,7 @@ requirements:
     - typing-extensions >=4.6.2
     - rich >=13.5.1
     - matplotlib >=3.7.2
+    - networkx >=3.0.0
 
 test:
   imports:

doc/Changelog.md

@@ -4,6 +4,14 @@
 
 **In development**
 
+- {gh-pr}`141` {gh-issue}`137` Add `ElectricalNetwork.to_graph()` to get a `networkx.Graph` object
+  representing the electrical network for graph theory studies. Install with the `"graph"` extra to
+  get _networkx_.
+  `ElectricalNetwork` also gained a new `buses_clusters` property that returns a list of sets of
+  IDs of buses that are connected by a line or a switch. This can be useful to isolate parts of the
+  network for localized analysis. For example, to study a LV subnetwork of a MV feeder. Alternatively,
+  to get the cluster certain bus belongs to, you can use `Bus.find_neighbors()`.
+- {gh-pr}`141` Add official support for Python 3.12. This is the last release to support Python 3.9.
 - {gh-pr}`138` Add network constraints for analysis of the results.
   - Buses can define minimum and maximum voltages. Use `bus.res_violated` to see if the bus has
     over- or under-voltage.

doc/conf.py

@@ -130,6 +130,8 @@
     "pint": ("https://pint.readthedocs.io/en/stable/", None),
     "typing_extensions": ("https://typing-extensions.readthedocs.io/en/stable/", None),
     "rich": ("https://rich.readthedocs.io/en/stable/", None),
+    "matplotlib": ("https://matplotlib.org/stable/", None),
+    "networkx": ("https://networkx.org/documentation/stable/", None),
 }
 
 # -- Options for sphinx_copybutton -------------------------------------------

doc/usage/Extras.md

@@ -2,6 +2,39 @@
 
 `roseau-load-flow` comes with some extra features that can be useful for some users.
 
+## Graph theory
+
+{meth}`ElectricalNetwork.to_graph() <roseau.load_flow.ElectricalNetwork.to_graph>` can be used to
+get a {class}`networkx.Graph` object from the electrical network.
+
+The graph contains the geometries of the buses in the nodes data and the geometries and branch
+types in the edges data.
+
+```{note}
+This method requires *networkx* which is not installed by default in pip managed installs. You can
+install it with the `"graph"` extra if you are using pip: `pip install "roseau-load-flow[graph]"`.
+```
+
+In addition, you can use the property
+{meth}`ElectricalNetwork.buses_clusters <roseau.load_flow.ElectricalNetwork.buses_clusters>` to
+get a list of sets of IDs of buses connected by a line or a switch. For example, with a network
+with a MV feeder, this property returns a list containing a set of MV buses IDs and all sets of
+LV subnetworks buses IDs. If you want to get the cluster of only one bus, you can use
+{meth}`Bus.find_neighbors <roseau.load_flow.models.Bus.find_neighbors>`
+
+If we take the example network from the [Getting Started page](gs-creating-network):
+
+```pycon
+>>> set(source_bus.find_neighbors())
+{'sb', 'lb'}
+>>> set(load_bus.find_neighbors())
+{'sb', 'lb'}
+>>> en.buses_clusters
+[{'sb', 'lb'}]
+```
+
+As there are no transformers between the two buses, they all belong to the same cluster.
+
 ## Conversion to symmetrical components
 
 {mod}`roseau.load_flow.converters` contains helpers to convert between phasor and symmetrical

pyproject.toml

@@ -78,7 +78,7 @@ ruff = "==0.1.3" # keep in sync with .pre-commit-config.yaml
 sphinx = "^7.0.1"
 myst-parser = ">=0.16.1"
 sphinx-math-dollar = "^1.2.1"
-sphinx-autoapi = "^2.0.0"
+sphinx-autoapi = "^3.0.0"
 sphinx-copybutton = ">=0.5.1"
 sphinx-inline-tabs = ">=2022.1.2b11"
 furo = ">=2022.9.29"

roseau/load_flow/models/buses.py

@@ -1,5 +1,5 @@
 import logging
-from collections.abc import Sequence
+from collections.abc import Iterator, Sequence
 from typing import TYPE_CHECKING, Any, Optional
 
 import numpy as np
@@ -253,6 +253,29 @@ def propagate_limits(self, force: bool = False) -> None:
             bus._min_voltage = self._min_voltage
             bus._max_voltage = self._max_voltage
 
+    def find_neighbors(self) -> Iterator[Id]:
+        """Find the buses connected to this bus via a line or switch recursively."""
+        from roseau.load_flow.models.lines import Line, Switch
+
+        visited_buses = {self.id}
+        yield self.id
+
+        visited: set[Element] = set()
+        remaining = set(self._connected_elements)
+
+        while remaining:
+            branch = remaining.pop()
+            visited.add(branch)
+            if not isinstance(branch, (Line, Switch)):
+                continue
+            for element in branch._connected_elements:
+                if not isinstance(element, Bus) or element.id in visited_buses:
+                    continue
+                visited_buses.add(element.id)
+                yield element.id
+                to_add = set(element._connected_elements).difference(visited)
+                remaining.update(to_add)
+
     #
     # Json Mixin interface
     #

roseau/load_flow/models/loads/flexible_parameters.py

@@ -8,27 +8,14 @@
 from roseau.load_flow.exceptions import RoseauLoadFlowException, RoseauLoadFlowExceptionCode
 from roseau.load_flow.typing import Authentication, ControlType, JsonDict, ProjectionType
 from roseau.load_flow.units import Q_, ureg_wraps
-from roseau.load_flow.utils import JsonMixin
+from roseau.load_flow.utils import JsonMixin, _optional_deps
 
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
     from matplotlib.axes import Axes
 
 
-def _import_matplotlib_pyplot():
-    try:
-        import matplotlib.pyplot
-    except ImportError as e:
-        msg = (
-            'matplotlib is required for plotting. Install it with the "plot" extra using '
-            '`pip install -U "roseau-load-flow[plot]"`'
-        )
-        logger.error(msg)
-        raise RoseauLoadFlowException(msg=msg, code=RoseauLoadFlowExceptionCode.IMPORT_ERROR) from e
-    return matplotlib.pyplot
-
-
 class Control(JsonMixin):
     """Control class for flexible loads.
 
@@ -1065,12 +1052,12 @@ def plot_pq(
             The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else
             the computed values).
         """
-        plt = _import_matplotlib_pyplot()  # this line first for better error handling
+        plt = _optional_deps.pyplot  # this line first for better error handling
         from matplotlib import colormaps, patheffects
 
         # Get the axes
         if ax is None:
-            ax: "Axes" = plt.gca()
+            ax = plt.gca()
 
         # Initialise some variables
         if voltages_labels_mask is None:
@@ -1173,11 +1160,11 @@ def plot_control_p(
             The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else
             the computed values).
         """
-        plt = _import_matplotlib_pyplot()
+        plt = _optional_deps.pyplot
 
         # Get the axes
         if ax is None:
-            ax: "Axes" = plt.gca()
+            ax = plt.gca()
 
         # Depending on the type of the control, several options
         x, y, x_ticks = self._theoretical_control_data(
@@ -1239,11 +1226,11 @@ def plot_control_q(
             The axis on which the plot has been drawn and the resulting flexible powers (the input if not `None` else
             the computed values).
         """
-        plt = _import_matplotlib_pyplot()
+        plt = _optional_deps.pyplot
 
         # Get the axes
         if ax is None:
-            ax: "Axes" = plt.gca()
+            ax = plt.gca()
 
         # Depending on the type of the control, several options
         x, y, x_ticks = self._theoretical_control_data(

roseau/load_flow/models/tests/test_buses.py

@@ -281,3 +281,46 @@ def test_propagate_limits():  # noqa: C901
     for bus in (b1_lv, b2_lv):
         assert bus.min_voltage == Q_(217, "V")
         assert bus.max_voltage == Q_(253, "V")
+
+
+def test_find_neighbors():
+    b1_mv = Bus("b1_mv", phases="abc")
+    b2_mv = Bus("b2_mv", phases="abc")
+    b3_mv = Bus("b3_mv", phases="abc")
+    b4_mv = Bus("b4_mv", phases="abc")
+    b1_lv = Bus("b1_lv", phases="abcn")
+    b2_lv = Bus("b2_lv", phases="abcn")
+    b3_lv = Bus("b3_lv", phases="abcn")
+
+    PotentialRef("pref_mv", element=b1_mv)
+    g = Ground("g")
+    PotentialRef("pref_lv", element=g)
+
+    lp_mv = LineParameters("lp_mv", z_line=np.eye(3), y_shunt=0.1 * np.eye(3))
+    lp_lv = LineParameters("lp_lv", z_line=np.eye(4))
+    tp = TransformerParameters.from_catalogue(id="SE_Minera_A0Ak_100kVA", manufacturer="SE")
+
+    Line("l1_mv", b1_mv, b2_mv, length=1.5, parameters=lp_mv, ground=g)
+    Line("l2_mv", b2_mv, b3_mv, length=2, parameters=lp_mv, ground=g)
+    Line("l3_mv", b2_mv, b4_mv, length=0.5, parameters=lp_mv, ground=g)  # creates a loop
+    Switch("sw_mv", b3_mv, b4_mv)
+    Transformer("tr", b3_mv, b1_lv, parameters=tp)
+    Line("l1_lv", b1_lv, b2_lv, length=1, parameters=lp_lv)
+    Switch("sw_lv", b2_lv, b3_lv)
+
+    voltages = 20_000 * np.exp([0, -2 / 3 * np.pi * 1j, 2 / 3 * np.pi * 1j])
+    VoltageSource("s_mv", bus=b1_mv, voltages=voltages)
+
+    PowerLoad("pl1_mv", bus=b2_mv, powers=[10e3, 10e3, 10e3])
+    PowerLoad("pl2_mv", bus=b3_mv, powers=[10e3, 10e3, 10e3])
+    PowerLoad("pl1_lv", bus=b1_lv, powers=[1e3, 1e3, 1e3])
+    PowerLoad("pl2_lv", bus=b2_lv, powers=[1e3, 1e3, 1e3])
+
+    mv_buses = (b1_mv, b2_mv, b3_mv, b4_mv)
+    mv_bus_ids = sorted(b.id for b in mv_buses)
+    lv_buses = (b1_lv, b2_lv, b3_lv)
+    lv_bus_ids = sorted(b.id for b in lv_buses)
+    for mvb in mv_buses:
+        assert sorted(mvb.find_neighbors()) == mv_bus_ids
+    for lvb in lv_buses:
+        assert sorted(lvb.find_neighbors()) == lv_bus_ids