Add problematic electrical component connections hierarchy

RELEASE_NOTES.md

@@ -78,7 +78,7 @@
 
 * Added a new `frequenz.client.common.microgrid.Microgrid` type, together with the `frequenz.client.common.microgrid.proto.v1alpha8.microgrid_from_proto` conversion function.
 
-* Added a new `frequenz.client.common.microgrid.electrical_components` package, featuring a `ElectricalComponent` class hierarchy and its families (battery, inverter, EV charger, etc.), and `ElectricalComponentConnection`, including `v1alpha8` proto conversion functions.
+* Added a new `frequenz.client.common.microgrid.electrical_components` package, featuring a `ElectricalComponent` class hierarchy and its families (battery, inverter, EV charger, etc.), and `ElectricalComponentConnection` class hierarchy, including `v1alpha8` proto conversion functions.
 
     The class of a component is its identity; components don't carry category or type attributes. The only exceptions are the error-recovery classes `UnrecognizedElectricalComponent` and `MismatchedCategoryElectricalComponent` (with a raw protobuf `category` value) and `UnrecognizedBattery`, `UnrecognizedInverter` and `UnrecognizedEvCharger` (with a raw protobuf `type` value), which preserve the raw protobuf values received from the protocol version used to load them.
 

src/frequenz/client/common/microgrid/electrical_components/__init__.py

@@ -19,7 +19,10 @@
 from ._crypto_miner import CryptoMiner
 from ._diagnostic_code import ElectricalComponentDiagnosticCode
 from ._electrical_component import ElectricalComponent
-from ._electrical_component_connection import ElectricalComponentConnection
+from ._electrical_component_connection import (
+    BaseElectricalComponentConnection,
+    ElectricalComponentConnection,
+)
 from ._electrolyzer import Electrolyzer
 from ._ev_charger import (
     AcEvCharger,
@@ -52,11 +55,17 @@
     UnrecognizedElectricalComponent,
     UnspecifiedElectricalComponent,
 )
+from ._problematic_connection import (
+    ProblematicElectricalComponentConnection,
+    SelfReferencingElectricalComponentConnection,
+)
 from ._state_code import ElectricalComponentStateCode
 from ._static_transfer_switch import StaticTransferSwitch
 from ._steam_boiler import SteamBoiler
 from ._types import (
+    ElectricalComponentConnectionTypes,
     ElectricalComponentTypes,
+    ProblematicElectricalComponentConnectionTypes,
     ProblematicElectricalComponentTypes,
     UnrecognizedElectricalComponentTypes,
     UnspecifiedElectricalComponentTypes,
@@ -66,6 +75,7 @@
 
 __all__ = [
     "AcEvCharger",
+    "BaseElectricalComponentConnection",
     "Battery",
     "BatteryInverter",
     "BatteryTypes",
@@ -78,6 +88,7 @@
     "ElectricalComponent",
     "ElectricalComponentCategory",
     "ElectricalComponentConnection",
+    "ElectricalComponentConnectionTypes",
     "ElectricalComponentDiagnosticCode",
     "ElectricalComponentId",
     "ElectricalComponentStateCode",
@@ -99,8 +110,11 @@
     "PowerTransformer",
     "Precharger",
     "ProblematicElectricalComponent",
+    "ProblematicElectricalComponentConnection",
+    "ProblematicElectricalComponentConnectionTypes",
     "ProblematicElectricalComponentTypes",
     "PvInverter",
+    "SelfReferencingElectricalComponentConnection",
     "StaticTransferSwitch",
     "SteamBoiler",
     "UninterruptiblePowerSupply",

src/frequenz/client/common/microgrid/electrical_components/_electrical_component_connection.py

@@ -5,18 +5,24 @@
 
 import dataclasses
 from datetime import datetime, timezone
+from typing import Any, Self
 
 from ...types import Lifetime
 from ._ids import ElectricalComponentId
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
-class ElectricalComponentConnection:
-    """A single electrical link between two electrical components within a microgrid.
-
-    An electrical component connection represents the physical wiring as viewed from the
-    grid connection point, if one exists, or from the islanding point, in case of an
-    islanded microgrid.
+class BaseElectricalComponentConnection:
+    """A base class for all electrical component connections.
+
+    This is the common supertype of every kind of connection, both
+    well-formed (see
+    [`ElectricalComponentConnection`][..ElectricalComponentConnection]) and
+    problematic (see
+    [`ProblematicElectricalComponentConnection`][..ProblematicElectricalComponentConnection]).
+    It cannot be instantiated directly; use one of its concrete subclasses
+    instead, or obtain instances via the corresponding `*_from_proto`
+    converter.
 
     Note: Physical Representation
         This object is not about data flow but rather about the physical
@@ -52,10 +58,12 @@ class ElectricalComponentConnection:
     operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime)
     """The operational lifetime of the connection."""
 
-    def __post_init__(self) -> None:
-        """Ensure that the source and destination electrical components are different."""
-        if self.source_id == self.destination_id:
-            raise ValueError("Source and destination components must be different")
+    # pylint: disable-next=unused-argument
+    def __new__(cls, *args: Any, **kwargs: Any) -> Self:
+        """Prevent instantiation of this class."""
+        if cls is BaseElectricalComponentConnection:
+            raise TypeError(f"Cannot instantiate {cls.__name__} directly")
+        return super().__new__(cls)
 
     def is_operational_at(self, timestamp: datetime) -> bool:
         """Check whether this connection is operational at a specific timestamp."""
@@ -68,3 +76,30 @@ def is_operational_now(self) -> bool:
     def __str__(self) -> str:
         """Return a human-readable string representation of this instance."""
         return f"{self.source_id}->{self.destination_id}"
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class ElectricalComponentConnection(BaseElectricalComponentConnection):
+    """A single electrical link between two distinct electrical components in a microgrid.
+
+    This is the well-formed case of an electrical component connection: the
+    source and destination are guaranteed to be different components.
+    Malformed cases (e.g. self-loops) are represented by dedicated
+    subclasses of
+    [`ProblematicElectricalComponentConnection`][..ProblematicElectricalComponentConnection]
+    instead.
+    """
+
+    def __post_init__(self) -> None:
+        """Ensure that the source and destination electrical components are different.
+
+        Raises:
+            ValueError: If
+                [`source_id`][...BaseElectricalComponentConnection.source_id]
+                and
+                [`destination_id`][...BaseElectricalComponentConnection.destination_id]
+                are equal, since that would describe a self-loop rather than
+                a well-formed connection.
+        """
+        if self.source_id == self.destination_id:
+            raise ValueError("Source and destination components must be different")

src/frequenz/client/common/microgrid/electrical_components/_problematic_connection.py

@@ -0,0 +1,61 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Problematic electrical component connections."""
+
+import dataclasses
+from typing import Any, Self
+
+from ._electrical_component_connection import BaseElectricalComponentConnection
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class ProblematicElectricalComponentConnection(BaseElectricalComponentConnection):
+    """An abstract electrical component connection with a problem.
+
+    This is the base class for connections that carry a data-integrity issue.
+
+    Problematic connections are siblings of the well-formed
+    [`ElectricalComponentConnection`][..ElectricalComponentConnection]: both
+    extend
+    [`BaseElectricalComponentConnection`][..BaseElectricalComponentConnection].
+    """
+
+    # pylint: disable-next=unused-argument
+    def __new__(cls, *args: Any, **kwargs: Any) -> Self:
+        """Prevent instantiation of this class."""
+        if cls is ProblematicElectricalComponentConnection:
+            raise TypeError(f"Cannot instantiate {cls.__name__} directly")
+        return super().__new__(cls)
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class SelfReferencingElectricalComponentConnection(
+    ProblematicElectricalComponentConnection
+):
+    """An electrical component connection whose source and destination are the same.
+
+    This represents a self-loop in the microgrid topology, which is physically
+    impossible and normally invalid. Instances of this class are produced by
+    the ``*_from_proto`` converters when they receive a connection whose
+    source and destination component IDs are identical, so that the
+    problematic data is exposed to the caller instead of being silently
+    discarded.
+    """
+
+    def __post_init__(self) -> None:
+        """Ensure that source and destination refer to the same component.
+
+        Raises:
+            ValueError: If
+                [`source_id`][...BaseElectricalComponentConnection.source_id]
+                and
+                [`destination_id`][...BaseElectricalComponentConnection.destination_id]
+                are different, since a self-referencing connection is defined
+                by them being the same.
+        """
+        if self.source_id != self.destination_id:
+            raise ValueError(
+                "Source and destination components must be the same for a "
+                "self-referencing electrical component connection"
+            )

src/frequenz/client/common/microgrid/electrical_components/_types.py

@@ -11,6 +11,7 @@
 from ._chp import Chp
 from ._converter import Converter
 from ._crypto_miner import CryptoMiner
+from ._electrical_component_connection import ElectricalComponentConnection
 from ._electrolyzer import Electrolyzer
 from ._ev_charger import (
     EvChargerTypes,
@@ -33,6 +34,7 @@
     UnrecognizedElectricalComponent,
     UnspecifiedElectricalComponent,
 )
+from ._problematic_connection import SelfReferencingElectricalComponentConnection
 from ._static_transfer_switch import StaticTransferSwitch
 from ._steam_boiler import SteamBoiler
 from ._uninterruptible_power_supply import UninterruptiblePowerSupply
@@ -61,6 +63,21 @@
 )
 """All possible electrical component types that have a problem."""
 
+ProblematicElectricalComponentConnectionTypes: TypeAlias = (
+    SelfReferencingElectricalComponentConnection
+)
+"""All possible electrical component connection types that have a problem."""
+
+ElectricalComponentConnectionTypes: TypeAlias = (
+    ElectricalComponentConnection | ProblematicElectricalComponentConnectionTypes
+)
+"""All concrete electrical component connection types.
+
+These are the concrete leaf types of electrical component connections that can
+be actually instantiated. Match against this union to exhaustively handle
+every kind of connection returned by the `*_from_proto` converters.
+"""
+
 ElectricalComponentTypes: TypeAlias = (
     BatteryTypes
     | Breaker

src/frequenz/client/common/microgrid/electrical_components/proto/v1alpha8/_electrical_component_connection.py

@@ -11,23 +11,26 @@
 
 from .....types import Lifetime
 from .....types.proto.v1alpha8 import lifetime_from_proto
-from ... import ElectricalComponentConnection, ElectricalComponentId
+from ... import (
+    ElectricalComponentConnection,
+    ElectricalComponentConnectionTypes,
+    ElectricalComponentId,
+    SelfReferencingElectricalComponentConnection,
+)
 
 _logger = logging.getLogger(__name__)
 
 
 def electrical_component_connection_from_proto(
     message: electrical_components_pb2.ElectricalComponentConnection,
-) -> ElectricalComponentConnection | None:
-    """Create an `ElectricalComponentConnection` from a protobuf message.
+) -> ElectricalComponentConnectionTypes:
+    """Create an electrical component connection from a protobuf message.
 
     Args:
         message: The protobuf message to convert.
 
     Returns:
-        The corresponding
-            [`ElectricalComponentConnection`][....ElectricalComponentConnection]
-            object, or `None` if the protobuf message is completely invalid.
+        One of the concrete connection types.
     """
     major_issues: list[str] = []
     minor_issues: list[str] = []
@@ -57,8 +60,8 @@ def electrical_component_connection_from_proto_with_issues(
     *,
     major_issues: list[str],
     minor_issues: list[str],
-) -> ElectricalComponentConnection | None:
-    """Create an `ElectricalComponentConnection` from a protobuf message, collecting issues.
+) -> ElectricalComponentConnectionTypes:
+    """Create an electrical component connection from a protobuf message, collecting issues.
 
     This function is useful when you want to collect issues during the parsing
     of multiple connections, rather than logging them immediately.
@@ -69,25 +72,27 @@ def electrical_component_connection_from_proto_with_issues(
         minor_issues: A list to collect minor issues found during parsing.
 
     Returns:
-        The corresponding
-            [`ElectricalComponentConnection`][....ElectricalComponentConnection]
-            object, or `None` if the protobuf message is completely invalid and
-            cannot be converted.
+        One of the concrete connection types.
     """
     source_component_id = ElectricalComponentId(message.source_electrical_component_id)
     destination_component_id = ElectricalComponentId(
         message.destination_electrical_component_id
     )
-    if source_component_id == destination_component_id:
-        major_issues.append(
-            f"connection ignored: source and destination are the same ({source_component_id})",
-        )
-        return None
-
     lifetime = _get_operational_lifetime_from_proto(
         message, major_issues=major_issues, minor_issues=minor_issues
     )
 
+    if source_component_id == destination_component_id:
+        major_issues.append(
+            "self-referencing connection: source and destination are the same "
+            f"({source_component_id})",
+        )
+        return SelfReferencingElectricalComponentConnection(
+            source_id=source_component_id,
+            destination_id=destination_component_id,
+            operational_lifetime=lifetime,
+        )
+
     return ElectricalComponentConnection(
         source_id=source_component_id,
         destination_id=destination_component_id,