Improve Lifetime validation and error handling

RELEASE_NOTES.md

@@ -90,7 +90,7 @@
 
 * Added `frequenz.client.common.grid.proto.v1alpha8.delivery_area_from_proto2` returning `DeliveryArea | InvalidDeliveryArea`. This is the replacement for the now-deprecated `delivery_area_from_proto`.
 
-* Added a new `frequenz.client.common.types.Lifetime` type together with the `frequenz.client.common.types.proto.v1alpha8.lifetime_from_proto` conversion function.
+* Added a new `frequenz.client.common.microgrid.Lifetime` type together with the `frequenz.client.common.microgrid.proto.v1alpha8.lifetime_from_proto` conversion function.
 
 * Added a new `frequenz.client.common.types.Location` type together with the `frequenz.client.common.types.proto.v1alpha8.location_from_proto` conversion function.
 

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

@@ -4,10 +4,20 @@
 """Frequenz microgrid definition."""
 
 from ._ids import EnterpriseId, MicrogridId
+from ._lifetime import (
+    BaseLifetime,
+    InvalidLifetime,
+    InvalidLifetimeError,
+    Lifetime,
+)
 from ._microgrid import Microgrid
 
 __all__ = [
+    "BaseLifetime",
     "EnterpriseId",
+    "InvalidLifetime",
+    "InvalidLifetimeError",
+    "Lifetime",
     "Microgrid",
     "MicrogridId",
 ]

src/frequenz/client/common/microgrid/_lifetime.py

@@ -0,0 +1,155 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Lifetime of an asset."""
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Self
+
+from .._exception import InvalidAttributeError
+
+
+@dataclass(frozen=True, kw_only=True)
+class BaseLifetime:
+    """A base class for well-formed and malformed operational lifetimes.
+
+    This class cannot be instantiated directly. Use [`Lifetime`][..Lifetime]
+    for a valid period or [`InvalidLifetime`][..InvalidLifetime] to preserve
+    malformed wire data.
+    """
+
+    start_time: datetime | None = None
+    """The moment when the asset became operationally active.
+
+    If `None`, the asset is considered to be active in any past moment previous to the
+    [`end_time`][..end_time].
+    """
+
+    end_time: datetime | None = None
+    """The moment when the asset's operational activity ceased.
+
+    If `None`, the asset is considered to be active with no plans to be deactivated.
+    """
+
+    # pylint: disable-next=unused-argument
+    def __new__(cls, *args: Any, **kwargs: Any) -> Self:
+        """Prevent instantiation of this class."""
+        if cls is BaseLifetime:
+            raise TypeError(f"Cannot instantiate {cls.__name__} directly")
+        return super().__new__(cls)
+
+
+@dataclass(frozen=True, kw_only=True)
+class Lifetime(BaseLifetime):
+    """An active operational period of an asset.
+
+    When both [`start_time`][.start_time] and [`end_time`][.end_time] are
+    `None`, the lifetime is unbounded and the asset is considered operational
+    at every timestamp.
+
+    Warning:
+        The [`end_time`][.end_time] timestamp indicates that the asset has been
+        permanently removed from service.
+
+    Note:
+        Raises a `ValueError` if [`start_time`][.start_time] is later than the
+        [`end_time`][.end_time] timestamp. Use
+        [`InvalidLifetime`][..InvalidLifetime] to represent malformed lifetime
+        data received from the wire.
+    """
+
+    def __post_init__(self) -> None:
+        """Validate this lifetime."""
+        if (
+            self.start_time is not None
+            and self.end_time is not None
+            and self.start_time > self.end_time
+        ):
+            raise ValueError(
+                f"Start ({self.start_time}) must be before or equal to end "
+                f"({self.end_time})"
+            )
+
+    def __str__(self) -> str:
+        """Return a compact string representation of this lifetime."""
+        start_str = (
+            self.start_time.isoformat() if self.start_time is not None else "-inf"
+        )
+        end_str = self.end_time.isoformat() if self.end_time is not None else "+inf"
+        return f"({start_str},{end_str}]"
+
+    def is_operational_at(self, timestamp: datetime) -> bool:
+        """Check whether this lifetime is active at a specific timestamp."""
+        # Handle start time - it's not active if start_time is in the future
+        if self.start_time is not None and self.start_time > timestamp:
+            return False
+        # Handle end time - active up to and including end_time
+        if self.end_time is not None:
+            return self.end_time >= timestamp
+        # self.end_time is None, and either self.start_time is None or
+        # self.start_time <= timestamp, so it is active at this timestamp
+        return True
+
+    def is_operational_now(self) -> bool:
+        """Whether this lifetime is currently active."""
+        return self.is_operational_at(datetime.now(timezone.utc))
+
+
+@dataclass(frozen=True, kw_only=True)
+class InvalidLifetime(BaseLifetime):
+    """An operational lifetime with malformed data received from the wire.
+
+    This class preserves lifetime data that fails the invariants required for
+    a well-formed [`Lifetime`][..Lifetime], allowing callers to inspect the raw
+    timestamps without accidentally using them for operational checks. Use a
+    semantic accessor, such as `ElectricalComponent.get_operational_lifetime()`,
+    to receive a clear [`InvalidLifetimeError`][..InvalidLifetimeError].
+    """
+
+    def __str__(self) -> str:
+        """Return a compact string representation of this invalid lifetime."""
+        start_str = (
+            self.start_time.isoformat() if self.start_time is not None else "-inf"
+        )
+        end_str = self.end_time.isoformat() if self.end_time is not None else "+inf"
+        return f"<invalid:({start_str},{end_str}]>"
+
+
+class InvalidLifetimeError(InvalidAttributeError):
+    """Raised when a semantic accessor sees an invalid lifetime.
+
+    The offending [`InvalidLifetime`][..InvalidLifetime] is available as the
+    [`lifetime`][.lifetime] attribute so callers can inspect the raw wire data.
+
+    This is also a [`ValueError`][] for convenience.
+    """
+
+    def __init__(
+        self,
+        instance: object,
+        attr_name: str,
+        lifetime: InvalidLifetime,
+        message: str | None = None,
+    ) -> None:
+        """Initialize this error.
+
+        Args:
+            instance: The instance that was being accessed when this error was raised.
+            attr_name: The name of the attribute that was being accessed.
+            lifetime: The invalid lifetime instance.
+            message: A custom error message. If `None`, a default message mentioning
+                the invalid lifetime is used.
+        """
+        self.lifetime: InvalidLifetime = lifetime
+        """The invalid lifetime that caused this error."""
+
+        super().__init__(
+            instance,
+            attr_name,
+            (
+                message
+                if message is not None
+                else f"invalid lifetime {lifetime!r} for attribute {attr_name!r} in {instance}"
+            ),
+        )

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

@@ -10,8 +10,8 @@
 
 from ..._exception import UnrecognizedEnumValueError, UnspecifiedEnumValueError
 from ...metrics import Bounds, Metric
-from ...types import Lifetime
 from .. import MicrogridId
+from .._lifetime import InvalidLifetime, InvalidLifetimeError, Lifetime
 from ._ids import ElectricalComponentId
 
 
@@ -28,14 +28,23 @@ class ElectricalComponent:  # pylint: disable=too-many-instance-attributes
     name: str
     """The name of this electrical component."""
 
-    model: str | None = None
+    model: str
     """The model of this electrical component.
 
     This includes both the manufacturer and the model name.
     """
 
-    operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime)
-    """The operational lifetime of this electrical component."""
+    operational_lifetime: Lifetime | InvalidLifetime = dataclasses.field(
+        default_factory=Lifetime
+    )
+    """The operational lifetime of this electrical component.
+
+    An [`InvalidLifetime`][....InvalidLifetime] preserves malformed wire data.
+
+    Tip:
+        Prefer [`get_operational_lifetime()`][..get_operational_lifetime] when
+        a valid lifetime is required.
+    """
 
     _provides_telemetry: bool | int
     """Whether this component provides telemetry data.
@@ -183,22 +192,51 @@ def accepts_control(self) -> bool:
             case unknown:
                 assert_never(unknown)
 
-    def is_operational_at(self, timestamp: datetime) -> bool:
+    def get_operational_lifetime(self) -> Lifetime:
+        """Return the operational lifetime as a valid `Lifetime`.
+
+        Returns:
+            The valid operational lifetime.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
+        """
+        match self.operational_lifetime:
+            case InvalidLifetime() as invalid:
+                raise InvalidLifetimeError(self, "operational_lifetime", invalid)
+            case Lifetime() as valid:
+                return valid
+            case unknown:
+                assert_never(unknown)
+
+    def is_operational_at(self, timestamp: datetime) -> bool:  # noqa: DOC502
         """Check whether this electrical component is operational at a specific timestamp.
 
         Args:
             timestamp: The timestamp to check.
 
         Returns:
             Whether this electrical component is operational at the given timestamp.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
         """
-        return self.operational_lifetime.is_operational_at(timestamp)
+        return self.get_operational_lifetime().is_operational_at(timestamp)
 
-    def is_operational_now(self) -> bool:
+    def is_operational_now(self) -> bool:  # noqa: DOC502
         """Check whether this electrical component is currently operational.
 
         Returns:
             Whether this electrical component is operational at the current time.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
         """
         return self.is_operational_at(datetime.now(timezone.utc))
 

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

@@ -5,9 +5,9 @@
 
 import dataclasses
 from datetime import datetime, timezone
-from typing import Any, Self
+from typing import Any, Self, assert_never
 
-from ...types import Lifetime
+from .._lifetime import InvalidLifetime, InvalidLifetimeError, Lifetime
 from ._ids import ElectricalComponentId
 
 
@@ -55,8 +55,17 @@ class BaseElectricalComponentConnection:
     This is the electrical component towards which the current flows.
     """
 
-    operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime)
-    """The operational lifetime of the connection."""
+    operational_lifetime: Lifetime | InvalidLifetime = dataclasses.field(
+        default_factory=Lifetime
+    )
+    """The operational lifetime of the connection.
+
+    An [`InvalidLifetime`][....InvalidLifetime] preserves malformed wire data.
+
+    Tip:
+        Prefer [`get_operational_lifetime()`][..get_operational_lifetime] when
+        a valid lifetime is required.
+    """
 
     # pylint: disable-next=unused-argument
     def __new__(cls, *args: Any, **kwargs: Any) -> Self:
@@ -65,12 +74,52 @@ def __new__(cls, *args: Any, **kwargs: Any) -> Self:
             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."""
-        return self.operational_lifetime.is_operational_at(timestamp)
+    def get_operational_lifetime(self) -> Lifetime:
+        """Return the operational lifetime as a valid `Lifetime`.
+
+        Returns:
+            The valid operational lifetime.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
+        """
+        match self.operational_lifetime:
+            case InvalidLifetime() as invalid:
+                raise InvalidLifetimeError(self, "operational_lifetime", invalid)
+            case Lifetime() as valid:
+                return valid
+            case unknown:
+                assert_never(unknown)
+
+    def is_operational_at(self, timestamp: datetime) -> bool:  # noqa: DOC502
+        """Check whether this connection is operational at a specific timestamp.
+
+        Args:
+            timestamp: The timestamp to check against the operational lifetime.
 
-    def is_operational_now(self) -> bool:
-        """Whether this connection is currently operational."""
+        Returns:
+            Whether this connection is operational at the given timestamp.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
+        """
+        return self.get_operational_lifetime().is_operational_at(timestamp)
+
+    def is_operational_now(self) -> bool:  # noqa: DOC502
+        """Whether this connection is currently operational.
+
+        Returns:
+            Whether this connection is operational at the current time.
+
+        Raises:
+            InvalidLifetimeError: If malformed lifetime data was received. The
+                offending value is available on the exception's `lifetime`
+                attribute.
+        """
         return self.is_operational_at(datetime.now(timezone.utc))
 
     def __str__(self) -> str: