Add ComponentConnection

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/client/microgrid/component/__init__.py

@@ -15,6 +15,7 @@
 from ._category import ComponentCategory
 from ._chp import Chp
 from ._component import Component
+from ._connection import ComponentConnection
 from ._converter import Converter
 from ._crypto_miner import CryptoMiner
 from ._electrolyzer import Electrolyzer
@@ -67,6 +68,7 @@
     "Chp",
     "Component",
     "ComponentCategory",
+    "ComponentConnection",
     "ComponentStatus",
     "ComponentTypes",
     "Converter",

src/frequenz/client/microgrid/component/_connection.py

@@ -0,0 +1,72 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Component connection."""
+
+import dataclasses
+from datetime import datetime, timezone
+
+from frequenz.client.common.microgrid.components import ComponentId
+
+from .._lifetime import Lifetime
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class ComponentConnection:
+    """A single electrical link between two components within a microgrid.
+
+    A 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 message is not about data flow but rather about the physical
+        electrical connections between components. Therefore, the IDs for the
+        source and destination 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 allows for
+        tracking the changes over time to a microgrid, providing insights into
+        when and how the microgrid infrastructure has been modified.
+    """
+
+    source: ComponentId
+    """The unique identifier of the 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: ComponentId
+    """The unique ID of the component where the connection terminates.
+
+    This is the 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 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/microgrid/component/_connection_proto.py

@@ -0,0 +1,104 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of ComponentConnection objects from protobuf messages."""
+
+import logging
+
+from frequenz.api.common.v1.microgrid.components import components_pb2
+from frequenz.client.common.microgrid.components import ComponentId
+
+from .._lifetime import Lifetime
+from .._lifetime_proto import lifetime_from_proto
+from ._connection import ComponentConnection
+
+_logger = logging.getLogger(__name__)
+
+
+def component_connection_from_proto(
+    message: components_pb2.ComponentConnection,
+) -> ComponentConnection | None:
+    """Create a `ComponentConnection` from a protobuf message."""
+    major_issues: list[str] = []
+    minor_issues: list[str] = []
+
+    connection = component_connection_from_proto_with_issues(
+        message, major_issues=major_issues, minor_issues=minor_issues
+    )
+
+    if major_issues:
+        _logger.warning(
+            "Found issues in component connection: %s | Protobuf message:\n%s",
+            ", ".join(major_issues),
+            message,
+        )
+    if minor_issues:
+        _logger.debug(
+            "Found minor issues in component connection: %s | Protobuf message:\n%s",
+            ", ".join(minor_issues),
+            message,
+        )
+
+    return connection
+
+
+def component_connection_from_proto_with_issues(
+    message: components_pb2.ComponentConnection,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> ComponentConnection | None:
+    """Create a `ComponentConnection` 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:
+        A `ComponentConnection` object created from the protobuf message, or
+            `None` if the protobuf message is completely invalid and a
+            `ComponentConnection` cannot be created.
+    """
+    source_component_id = ComponentId(message.source_component_id)
+    destination_component_id = ComponentId(message.destination_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 ComponentConnection(
+        source=source_component_id,
+        destination=destination_component_id,
+        operational_lifetime=lifetime,
+    )
+
+
+def _get_operational_lifetime_from_proto(
+    message: components_pb2.ComponentConnection,
+    *,
+    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()

tests/component/test_connection.py

@@ -0,0 +1,81 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for ComponentConnection class and related functionality."""
+
+from datetime import datetime, timezone
+from unittest.mock import Mock, patch
+
+import pytest
+from frequenz.client.common.microgrid.components import ComponentId
+
+from frequenz.client.microgrid import Lifetime
+from frequenz.client.microgrid.component import ComponentConnection
+
+
+def test_creation() -> None:
+    """Test basic ComponentConnection creation and validation."""
+    now = datetime.now(timezone.utc)
+    lifetime = Lifetime(start=now)
+    connection = ComponentConnection(
+        source=ComponentId(1), destination=ComponentId(2), operational_lifetime=lifetime
+    )
+
+    assert connection.source == ComponentId(1)
+    assert connection.destination == ComponentId(2)
+    assert connection.operational_lifetime == lifetime
+
+
+def test_validation() -> None:
+    """Test validation of source and destination components."""
+    with pytest.raises(
+        ValueError, match="Source and destination components must be different"
+    ):
+        ComponentConnection(source=ComponentId(1), destination=ComponentId(1))
+
+
+def test_str() -> None:
+    """Test string representation of ComponentConnection."""
+    connection = ComponentConnection(source=ComponentId(1), destination=ComponentId(2))
+    assert str(connection) == "CID1->CID2"
+
+
+@pytest.mark.parametrize(
+    "lifetime_active", [True, False], ids=["operational", "not-operational"]
+)
+def test_is_operational_at(lifetime_active: bool) -> None:
+    """Test active_at behavior with lifetime.active values."""
+    mock_lifetime = Mock(spec=Lifetime)
+    mock_lifetime.is_operational_at.return_value = lifetime_active
+
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+        operational_lifetime=mock_lifetime,
+    )
+
+    now = datetime.now(timezone.utc)
+    assert connection.is_operational_at(now) == lifetime_active
+    mock_lifetime.is_operational_at.assert_called_once_with(now)
+
+
+@patch("frequenz.client.microgrid.component._connection.datetime")
+@pytest.mark.parametrize(
+    "lifetime_active", [True, False], ids=["operational", "not-operational"]
+)
+def test_is_operational_now(mock_datetime: Mock, lifetime_active: bool) -> None:
+    """Test if the connection is operational at the current time."""
+    now = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+    mock_datetime.now.side_effect = lambda tz: now.replace(tzinfo=tz)
+    mock_lifetime = Mock(spec=Lifetime)
+    mock_lifetime.is_operational_at.return_value = lifetime_active
+
+    connection = ComponentConnection(
+        source=ComponentId(1),
+        destination=ComponentId(2),
+        operational_lifetime=mock_lifetime,
+    )
+
+    assert connection.is_operational_now() is lifetime_active
+    mock_lifetime.is_operational_at.assert_called_once_with(now)
+    mock_datetime.now.assert_called_once_with(timezone.utc)