Add ElectricalComponentConnection wrapper

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

@@ -18,6 +18,7 @@
 from ._crypto_miner import CryptoMiner
 from ._diagnostic_code import ElectricalComponentDiagnosticCode
 from ._electrical_component import ElectricalComponent
+from ._electrical_component_connection import ElectricalComponentConnection
 from ._electrolyzer import Electrolyzer
 from ._ev_charger import (
     AcEvCharger,
@@ -75,6 +76,7 @@
     "DcEvCharger",
     "ElectricalComponent",
     "ElectricalComponentCategory",
+    "ElectricalComponentConnection",
     "ElectricalComponentDiagnosticCode",
     "ElectricalComponentId",
     "ElectricalComponentStateCode",

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

@@ -0,0 +1,71 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Electrical component connection."""
+
+import dataclasses
+from datetime import datetime, timezone
+
+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 microgrids.
+
+    Note: Physical Representation
+        This object is not about data flow but rather about the physical
+        electrical connections between electrical components. Therefore, the IDs for the
+        source and destination electrical components correspond to the actual setup within
+        the microgrid.
+
+    Note: Direction
+        The direction of the connection follows the flow of current away from the
+        grid connection point, or in case of islands, away from the islanding
+        point. This direction is aligned with positive current according to the
+        [Passive Sign Convention]
+        (https://en.wikipedia.org/wiki/Passive_sign_convention).
+
+    Note: Historical Data
+        The timestamps of when a connection was created and terminated allow for
+        tracking the changes over time to a microgrid, providing insights into
+        when and how the microgrid infrastructure has been modified.
+    """
+
+    source: ElectricalComponentId
+    """The unique identifier of the electrical component where the connection originates.
+
+    This is aligned with the direction of current flow away from the grid connection
+    point, or in case of islands, away from the islanding point.
+    """
+
+    destination: ElectricalComponentId
+    """The unique ID of the electrical component where the connection terminates.
+
+    This is the electrical component towards which the current flows.
+    """
+
+    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 == self.destination:
+            raise ValueError("Source and destination components must be different")
+
+    def is_operational_at(self, timestamp: datetime) -> bool:
+        """Check whether this connection is operational at a specific timestamp."""
+        return self.operational_lifetime.is_operational_at(timestamp)
+
+    def is_operational_now(self) -> bool:
+        """Whether this connection is currently operational."""
+        return self.is_operational_at(datetime.now(timezone.utc))
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of this instance."""
+        return f"{self.source}->{self.destination}"

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

@@ -13,10 +13,16 @@
     electrical_component_state_code_from_proto,
     electrical_component_state_code_to_proto,
 )
+from ._electrical_component_connection import (
+    electrical_component_connection_from_proto,
+    electrical_component_connection_from_proto_with_issues,
+)
 
 __all__ = [
     "electrical_component_category_from_proto",
     "electrical_component_category_to_proto",
+    "electrical_component_connection_from_proto",
+    "electrical_component_connection_from_proto_with_issues",
     "electrical_component_diagnostic_code_from_proto",
     "electrical_component_diagnostic_code_to_proto",
     "electrical_component_from_proto",

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

@@ -0,0 +1,107 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of ElectricalComponentConnection objects from protobuf messages."""
+
+import logging
+
+from frequenz.api.common.v1alpha8.microgrid.electrical_components import (
+    electrical_components_pb2,
+)
+
+from .....types import Lifetime
+from .....types.proto.v1alpha8 import lifetime_from_proto
+from ... import ElectricalComponentConnection, ElectricalComponentId
+
+_logger = logging.getLogger(__name__)
+
+
+def electrical_component_connection_from_proto(
+    message: electrical_components_pb2.ElectricalComponentConnection,
+) -> ElectricalComponentConnection | None:
+    """Create an `ElectricalComponentConnection` from a protobuf message."""
+    major_issues: list[str] = []
+    minor_issues: list[str] = []
+
+    connection = electrical_component_connection_from_proto_with_issues(
+        message, major_issues=major_issues, minor_issues=minor_issues
+    )
+
+    if major_issues:
+        _logger.warning(
+            "Found issues in electrical component connection: %s | Protobuf message:\n%s",
+            ", ".join(major_issues),
+            message,
+        )
+    if minor_issues:
+        _logger.debug(
+            "Found minor issues in electrical component connection: %s | Protobuf message:\n%s",
+            ", ".join(minor_issues),
+            message,
+        )
+
+    return connection
+
+
+def electrical_component_connection_from_proto_with_issues(
+    message: electrical_components_pb2.ElectricalComponentConnection,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> ElectricalComponentConnection | None:
+    """Create an `ElectricalComponentConnection` 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.
+
+    Args:
+        message: The protobuf message to parse.
+        major_issues: A list to collect major issues found during parsing.
+        minor_issues: A list to collect minor issues found during parsing.
+
+    Returns:
+        An `ElectricalComponentConnection` object created from the protobuf message,
+            or `None` if the protobuf message is completely invalid and an
+            `ElectricalComponentConnection` cannot be created.
+    """
+    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
+    )
+
+    return ElectricalComponentConnection(
+        source=source_component_id,
+        destination=destination_component_id,
+        operational_lifetime=lifetime,
+    )
+
+
+def _get_operational_lifetime_from_proto(
+    message: electrical_components_pb2.ElectricalComponentConnection,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> Lifetime:
+    """Get the operational lifetime from a protobuf message."""
+    if message.HasField("operational_lifetime"):
+        try:
+            return lifetime_from_proto(message.operational_lifetime)
+        except ValueError as exc:
+            major_issues.append(
+                f"invalid operational lifetime ({exc}), considering it as missing "
+                "(i.e. always operational)",
+            )
+    else:
+        minor_issues.append(
+            "missing operational lifetime, considering it always operational",
+        )
+    return Lifetime()