Feature/Check subsystem graphs in sanity check

epowcore/gdf/core_model.py

@@ -1,9 +1,10 @@
-from ast import literal_eval as make_tuple
-from dataclasses import dataclass, field, asdict
 import importlib
+from ast import literal_eval as make_tuple
+from dataclasses import asdict, dataclass, field
 from typing import TypeVar
 
 import networkx as nx
+
 from epowcore.generic.component_graph import ComponentGraph
 from epowcore.generic.configuration import Configuration
 from epowcore.generic.constants import GDF_VERSION, Platform
@@ -269,8 +270,8 @@ def get_neighbors(
         :return: A list of components connected to to given [component].
         :rtype: list[Component]
         """
-        from epowcore.gdf.subsystem import Subsystem
         from epowcore.gdf.port import Port
+        from epowcore.gdf.subsystem import Subsystem
 
         _, graph = self.get_component_by_id(component.uid)
         if graph is None:
@@ -334,31 +335,44 @@ def get_valid_id(self) -> int:
         return max_id + 1
 
     def sanity_check(self) -> bool:
-        """Checks the validity of the model.
+        """Checks the validity of the model, including subsystem graphs."""
 
-        :return: True if the model is valid, else False.
-        """
+        from epowcore.gdf.subsystem import Subsystem
+
+        def check_graph(graph: ComponentGraph) -> bool:
+            graph_sanity = graph.sanity_check()
 
-        graph_sanity = self.graph.sanity_check()
-        # Check if the edges have the required connectors
-        connector_check = all(
-            map(
-                lambda node: len(node.connector_names) == 0
+            connector_check = all(
+                len(node.connector_names) == 0
                 or all(
-                    map(
-                        lambda x: self.has_connected_to(node, x),
-                        node.connector_names,
-                    )
-                ),
-                self.graph.nodes,
+                    connector_name in [
+                        connector
+                        for _, _, data in graph.edges.data(node)
+                        for connector in data.get(node.uid, [])
+                    ]
+                    for connector_name in node.connector_names
+                )
+                for node in graph.nodes
+            )
+
+            unique_ids_check = len(graph.nodes) == len(
+                {node.uid for node in graph.nodes}
+            )
+
+            subsystem_check = all(
+                check_graph(node.graph)
+                for node in graph.nodes
+                if isinstance(node, Subsystem)
+            )
+
+            return (
+                graph_sanity
+                and connector_check
+                and unique_ids_check
+                and subsystem_check
             )
-        )
 
-        # Check if the nodes have unique IDs
-        unique_ids_check = len(self.graph.nodes) == len(
-            set((node.uid for node in self.graph.nodes))
-        )
-        return graph_sanity and connector_check and unique_ids_check
+        return check_graph(self.graph)
 
     def export_dict(self) -> dict:
         """Export the whole model as a dictionary.

epowcore/generic/component_graph.py

@@ -1,11 +1,12 @@
-from collections.abc import Iterable
 import copy as cp
+from collections.abc import Iterable
 from functools import cached_property
+from pprint import pp
 from typing import Iterator
 
 import networkx as nx
-from epowcore.gdf.component import Component
 
+from epowcore.gdf.component import Component
 from epowcore.generic.component_views import ComponentEdgeView, ComponentNodeView
 
 
@@ -142,15 +143,20 @@ def sanity_check(self) -> bool:
         # Check the types of the components
         node_type_check = all(isinstance(node, Component) for node in self._graph.nodes)
 
-        # Check the types of the edge data
-        data_keys = [
-            a for b in map(lambda edge: list(edge[2].keys()), self._graph.edges.data()) for a in b
-        ]
-        data_values = [
-            a for b in map(lambda edge: list(edge[2].values()), self._graph.edges.data()) for a in b
-        ]
-        edge_type_check = all(isinstance(x, list) for x in data_keys) and all(
-            isinstance(x, list) for x in data_values
+        # The edge data ist a list of tuples,
+        # each tuple contains the two components on index 0 and 1 and a dictionary on index 2.
+        # This dictionary maps the integer uid of both components each to a list of connector name strings.
+        # The different connectors are connected in order of the lists.
+        # The goal of the following code is to verify the value types of the dictionary.
+        edge_data = list(self._graph.edges.data())
+        data_components = [item for tuple in edge_data for item in tuple[:2]]
+        data_keys = [a for b in map(lambda edge: list(edge[2].keys()), edge_data) for a in b]
+        data_values = [a for b in map(lambda edge: list(edge[2].values()), edge_data) for a in b]
+        edge_type_check = (
+            all(isinstance(component, Component) for component in data_components)
+            and all(isinstance(edge, tuple) for edge in edge_data)
+            and all(isinstance(x, int) for x in data_keys)
+            and all(isinstance(x, list) and all(isinstance(y, str) for y in x) for x in data_values)
         )
 
         return node_type_check and edge_type_check

tests/core/gdf/utils_test.py

@@ -1,10 +1,12 @@
+import json
 import pathlib
 import unittest
 
 from helpers.gdf_component_creator import GdfTestComponentCreator
 
 from epowcore.gdf.bus import Bus, LFBusType
 from epowcore.gdf.core_model import CoreModel
+from epowcore.gdf.subsystem import Subsystem
 from epowcore.gdf.utils import get_connected_bus
 
 PATH = pathlib.Path(__file__).parent.resolve()
@@ -13,7 +15,6 @@
 class UtilsTest(unittest.TestCase):
     """Test the utility functions of the gdf package."""
 
-    # @unittest.skip("tmp")
     def test_get_connected_bus(self) -> None:
         core_model = CoreModel(base_frequency=50.0)
 
@@ -33,6 +34,83 @@ def test_get_connected_bus(self) -> None:
         bus = get_connected_bus(core_model.graph, tline)
         self.assertIn(bus, (bus_a, bus_b))
 
+    def test_sanity_check_IEEE39(self) -> None:
+        path = pathlib.Path(__file__).parent.parent.resolve()
+
+        with open(
+            path.parent.parent / "tests/models/gdf/IEEE39_gdf.json",
+            "r",
+            encoding="utf-8",
+        ) as file:
+            data_str = file.read()
+            data = json.loads(data_str)
+            core_model = CoreModel.import_dict(data)
+
+        self.assertFalse(core_model.sanity_check())
+
+    def test_sanity_check_IEEE39_flat(self) -> None:
+        path = pathlib.Path(__file__).parent.parent.resolve()
+
+        with open(
+            path.parent.parent / "tests/models/gdf/IEEE39-flat_gdf.json",
+            "r",
+            encoding="utf-8",
+        ) as file:
+            data_str = file.read()
+            data = json.loads(data_str)
+            core_model = CoreModel.import_dict(data)
+
+        self.assertFalse(core_model.sanity_check())
+
+    def test_sanity_check_valid_subsystem(self) -> None:
+        """Sanity check succeeds when a subsystem graph is valid."""
+
+        creator = GdfTestComponentCreator(50.0)
+        core_model = creator.core_model
+
+        tline = creator.create_tline("Line")
+        bus_a = creator.create_bus("Bus A")
+        bus_b = creator.create_bus("Bus B")
+
+        core_model.add_connection(tline, bus_a, "A", "")
+        core_model.add_connection(tline, bus_b, "B", "")
+
+        Subsystem.from_components(core_model, [tline])
+
+        self.assertTrue(core_model.sanity_check())
+
+    def test_sanity_check_invalid_subsystem(self) -> None:
+        """Sanity check fails when a required connector is missing inside a subsystem."""
+
+        creator = GdfTestComponentCreator(50.0)
+        core_model = creator.core_model
+
+        tline = creator.create_tline("Line")
+        bus_a = creator.create_bus("Bus A")
+        bus_b = creator.create_bus("Bus B")
+
+        core_model.add_connection(tline, bus_a, "A", "")
+        core_model.add_connection(tline, bus_b, "B", "")
+
+        subsystem = Subsystem.from_components(core_model, [tline])
+
+        port = next(iter(subsystem.graph.neighbors(tline)))
+        subsystem.graph.edges[tline, port][tline.uid] = []
+
+        self.assertFalse(core_model.sanity_check())
+
+    def test_sanity_check_IEEE399_valid(self) -> None:
+        path = pathlib.Path(__file__).parent.parent.resolve()
+
+        with open(
+            path.parent.parent / "tests/models/gdf/IEEE399_gdf.json",
+            "r",
+            encoding="utf-8",
+        ) as file:
+            data = json.load(file)
+            core_model = CoreModel.import_dict(data)
+
+        self.assertTrue(core_model.sanity_check())
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()