Skip to content

Index

frequenz.client.common.microgrid ¤

Frequenz microgrid definition.

Classes¤

frequenz.client.common.microgrid.BaseLifetime dataclass ¤

A base class for well-formed and malformed operational lifetimes.

This class cannot be instantiated directly. Use Lifetime for a valid period or InvalidLifetime to preserve malformed wire data.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
@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)
Attributes¤
end_time class-attribute instance-attribute ¤
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.

start_time class-attribute instance-attribute ¤
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.

Methods:¤
__new__ ¤
__new__(*args: Any, **kwargs: Any) -> Self

Prevent instantiation of this class.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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)

frequenz.client.common.microgrid.EnterpriseId ¤

Bases: BaseId

A unique identifier for an enterprise account.

Source code in src/frequenz/client/common/microgrid/_ids.py
@final
class EnterpriseId(BaseId, str_prefix="EID"):
    """A unique identifier for an enterprise account."""
Attributes¤
str_prefix property ¤
str_prefix: str

The prefix used for the string representation of this ID.

Methods:¤
__eq__ ¤
__eq__(other: object) -> bool

Check if this instance is equal to another object.

Equality is defined as being of the exact same type and having the same underlying ID.

PARAMETER DESCRIPTION
other

The object to compare against.

TYPE: object

RETURNS DESCRIPTION
bool

True if other is of the same type and has the same underlying ID,

bool

NotImplemented if other is of a different type.

Source code in frequenz/core/id.py
def __eq__(self, other: object) -> bool:
    """Check if this instance is equal to another object.

    Equality is defined as being of the exact same type and having the same
    underlying ID.

    Args:
        other: The object to compare against.

    Returns:
        True if `other` is of the same type and has the same underlying ID,
        `NotImplemented` if `other` is of a different type.
    """
    # pylint thinks this is not an unidiomatic typecheck, but in this case
    # it is not. isinstance() returns True for subclasses, which is not
    # what we want here, as different ID types should never be equal.
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    # We already checked type(other) is type(self), but mypy doesn't
    # understand that, so we need to cast it to Self.
    other_id = cast(Self, other)
    return self._id == other_id._id
__hash__ ¤
__hash__() -> int

Return the hash of this instance.

The hash is based on the exact type and the underlying ID to ensure that IDs of different types but with the same numeric value have different hashes.

RETURNS DESCRIPTION
int

The hash of this instance.

Source code in frequenz/core/id.py
def __hash__(self) -> int:
    """Return the hash of this instance.

    The hash is based on the exact type and the underlying ID to ensure
    that IDs of different types but with the same numeric value have different hashes.

    Returns:
        The hash of this instance.
    """
    return hash((type(self), self._id))
__init__ ¤
__init__(id_: int) -> None

Initialize this instance.

PARAMETER DESCRIPTION
id_

The numeric unique identifier.

TYPE: int

RAISES DESCRIPTION
ValueError

If the ID is negative.

Source code in frequenz/core/id.py
def __init__(self, id_: int, /) -> None:
    """Initialize this instance.

    Args:
        id_: The numeric unique identifier.

    Raises:
        ValueError: If the ID is negative.
    """
    if id_ < 0:
        raise ValueError(f"{type(self).__name__} can't be negative.")
    self._id = id_
__init_subclass__ ¤
__init_subclass__(
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any
) -> None

Initialize a subclass, set its string prefix, and perform checks.

PARAMETER DESCRIPTION
str_prefix

The string prefix for the ID type (e.g., "MID"). Must be unique across all ID types.

TYPE: str

allow_custom_name

If True, bypasses the check that the class name must end with "Id". Defaults to False.

TYPE: bool DEFAULT: False

**kwargs

Forwarded to the parent's init_subclass.

TYPE: Any DEFAULT: {}

RAISES DESCRIPTION
TypeError

If allow_custom_name is False and the class name does not end with "Id".

Source code in frequenz/core/id.py
def __init_subclass__(
    cls,
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any,
) -> None:
    """Initialize a subclass, set its string prefix, and perform checks.

    Args:
        str_prefix: The string prefix for the ID type (e.g., "MID").
            Must be unique across all ID types.
        allow_custom_name: If True, bypasses the check that the class name
            must end with "Id". Defaults to False.
        **kwargs: Forwarded to the parent's __init_subclass__.

    Raises:
        TypeError: If `allow_custom_name` is False and the class name
            does not end with "Id".
    """
    super().__init_subclass__(**kwargs)

    if str_prefix in BaseId._registered_prefixes:
        # We want to raise an exception here, but currently can't due to
        # https://github.com/frequenz-floss/frequenz-repo-config-python/issues/421
        _logger.warning(
            "Prefix '%s' is already registered. ID prefixes must be unique.",
            str_prefix,
        )
    BaseId._registered_prefixes.add(str_prefix)

    if not allow_custom_name and not cls.__name__.endswith("Id"):
        raise TypeError(
            f"Class name '{cls.__name__}' for an ID class must end with 'Id' "
            "(e.g., 'SomeId'), or use `allow_custom_name=True`."
        )

    cls._str_prefix = str_prefix
__int__ ¤
__int__() -> int

Return the numeric ID of this instance.

Source code in frequenz/core/id.py
def __int__(self) -> int:
    """Return the numeric ID of this instance."""
    return self._id
__lt__ ¤
__lt__(other: object) -> bool

Check if this instance is less than another object.

Comparison is only defined between instances of the exact same type.

PARAMETER DESCRIPTION
other

The object to compare against.

TYPE: object

RETURNS DESCRIPTION
bool

True if this instance is less than other, NotImplemented if

bool

other is of a different type.

Source code in frequenz/core/id.py
def __lt__(self, other: object) -> bool:
    """Check if this instance is less than another object.

    Comparison is only defined between instances of the exact same type.

    Args:
        other: The object to compare against.

    Returns:
        True if this instance is less than `other`, `NotImplemented` if
        `other` is of a different type.
    """
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    other_id = cast(Self, other)
    return self._id < other_id._id
__new__ ¤
__new__(*_: Any, **__: Any) -> Self

Create a new instance of the ID class, only if it is a subclass of BaseId.

Source code in frequenz/core/id.py
def __new__(cls, *_: Any, **__: Any) -> Self:
    """Create a new instance of the ID class, only if it is a subclass of BaseId."""
    if cls is BaseId:
        raise TypeError("BaseId cannot be instantiated directly. Use a subclass.")
    return super().__new__(cls)
__repr__ ¤
__repr__() -> str

Return the string representation of this instance.

Source code in frequenz/core/id.py
def __repr__(self) -> str:
    """Return the string representation of this instance."""
    return f"{type(self).__name__}({self._id!r})"
__str__ ¤
__str__() -> str

Return the short string representation of this instance.

Source code in frequenz/core/id.py
def __str__(self) -> str:
    """Return the short string representation of this instance."""
    return f"{self._str_prefix}{self._id}"

frequenz.client.common.microgrid.InvalidLifetime dataclass ¤

Bases: 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, 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.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
@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}]>"
Attributes¤
end_time class-attribute instance-attribute ¤
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.

start_time class-attribute instance-attribute ¤
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.

Methods:¤
__new__ ¤
__new__(*args: Any, **kwargs: Any) -> Self

Prevent instantiation of this class.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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)
__str__ ¤
__str__() -> str

Return a compact string representation of this invalid lifetime.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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}]>"

frequenz.client.common.microgrid.InvalidLifetimeError ¤

Bases: InvalidAttributeError

Raised when a semantic accessor sees an invalid lifetime.

The offending InvalidLifetime is available as the lifetime attribute so callers can inspect the raw wire data.

This is also a ValueError for convenience.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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} for attribute {attr_name!r} in {instance}"
            ),
        )
Attributes¤
attr_name instance-attribute ¤
attr_name: str = attr_name

The name of the attribute that had an invalid value.

instance instance-attribute ¤
instance: object = instance

The object instance that had an invalid value.

lifetime instance-attribute ¤
lifetime: InvalidLifetime = lifetime

The invalid lifetime that caused this error.

Methods:¤
__init__ ¤
__init__(
    instance: object,
    attr_name: str,
    lifetime: InvalidLifetime,
    message: str | None = None,
) -> None

Initialize this error.

PARAMETER DESCRIPTION
instance

The instance that was being accessed when this error was raised.

TYPE: object

attr_name

The name of the attribute that was being accessed.

TYPE: str

lifetime

The invalid lifetime instance.

TYPE: InvalidLifetime

message

A custom error message. If None, a default message mentioning the invalid lifetime is used.

TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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} for attribute {attr_name!r} in {instance}"
        ),
    )

frequenz.client.common.microgrid.Lifetime dataclass ¤

Bases: BaseLifetime

An active operational period of an asset.

When both start_time and end_time are None, the lifetime is unbounded and the asset is considered operational at every timestamp.

Warning

The end_time timestamp indicates that the asset has been permanently removed from service.

Note

Raises a ValueError if start_time is later than the end_time timestamp. Use InvalidLifetime to represent malformed lifetime data received from the wire.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
@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))
Attributes¤
end_time class-attribute instance-attribute ¤
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.

start_time class-attribute instance-attribute ¤
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.

Methods:¤
__new__ ¤
__new__(*args: Any, **kwargs: Any) -> Self

Prevent instantiation of this class.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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)
__post_init__ ¤
__post_init__() -> None

Validate this lifetime.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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})"
        )
__str__ ¤
__str__() -> str

Return a compact string representation of this lifetime.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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}]"
is_operational_at ¤
is_operational_at(timestamp: datetime) -> bool

Check whether this lifetime is active at a specific timestamp.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
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
is_operational_now ¤
is_operational_now() -> bool

Whether this lifetime is currently active.

Source code in src/frequenz/client/common/microgrid/_lifetime.py
def is_operational_now(self) -> bool:
    """Whether this lifetime is currently active."""
    return self.is_operational_at(datetime.now(timezone.utc))

frequenz.client.common.microgrid.Microgrid dataclass ¤

A localized grouping of electricity generation, energy storage, and loads.

A microgrid is a localized grouping of electricity generation, energy storage, and loads that normally operates connected to a traditional centralized grid.

Each microgrid has a unique identifier and is associated with an enterprise account.

A key feature is that it has a physical location and is situated in a delivery area.

Key Concepts
  • Physical Location: Geographical coordinates specify the exact physical location of the microgrid.
  • Delivery Area: Each microgrid is part of a broader delivery area, which is crucial for energy trading and compliance.
Source code in src/frequenz/client/common/microgrid/_microgrid.py
@dataclass(frozen=True, kw_only=True)
class Microgrid:  # pylint: disable=too-many-instance-attributes
    """A localized grouping of electricity generation, energy storage, and loads.

    A microgrid is a localized grouping of electricity generation, energy storage, and
    loads that normally operates connected to a traditional centralized grid.

    Each microgrid has a unique identifier and is associated with an enterprise account.

    A key feature is that it has a physical location and is situated in a delivery area.

    Note: Key Concepts
        - Physical Location: Geographical coordinates specify the exact physical
          location of the microgrid.
        - Delivery Area: Each microgrid is part of a broader delivery area, which is
          crucial for energy trading and compliance.
    """

    id: MicrogridId
    """The unique identifier of the microgrid."""

    enterprise_id: EnterpriseId
    """The unique identifier linking this microgrid to its parent enterprise account."""

    name: str
    """The name of the microgrid."""

    delivery_area: DeliveryArea | InvalidDeliveryArea | None
    """The delivery area where the microgrid is located.

    `None` means the field was not set on the wire. An
    [`InvalidDeliveryArea`][....grid.InvalidDeliveryArea] means the wire
    carried a delivery area that fails its invariants.

    Tip:
        This is the lower-level field; prefer [`get_delivery_area()`][..get_delivery_area]
        to obtain a valid [`DeliveryArea`][....grid.DeliveryArea] or a clear error.
    """

    location: Location | None
    """The physical location of the microgrid, in geographical co-ordinates."""

    create_time: datetime.datetime
    """The UTC timestamp indicating when the microgrid was initially created."""

    _active: bool | int
    """Whether the microgrid is active.

    This stores the low-level representation of the microgrid state. It holds a `bool`
    for a known active/inactive status, the raw `int` `0` when the status is
    unspecified, or any other raw `int` not yet known to this client. Users should use
    [`Microgrid.is_active()`][.is_active] to obtain a clear boolean or a clear error.
    """

    _allow_construction: bool = field(
        default=False, repr=False, compare=False, hash=False
    )
    """Internal guard allowing construction only via the `microgrid_from_proto` converter."""

    def __post_init__(self) -> None:
        """Reject direct construction of this read-only type.

        Raises:
            TypeError: If the instance was not created via the
                [`microgrid_from_proto`][...proto.v1alpha8.microgrid_from_proto]
                converter.
        """
        if not self._allow_construction:
            raise TypeError(
                f"{type(self).__name__} cannot be constructed directly; obtain "
                "instances via the microgrid_from_proto converter."
            )

    def is_active(self) -> bool:
        """Return whether the microgrid is active.

        Returns:
            Whether the microgrid is active.

        Raises:
            UnspecifiedEnumValueError: If the status is unspecified.
            UnrecognizedEnumValueError: If the status is not recognized. The raw
                status value is available on the error's `value` attribute.
        """
        match self._active:
            case bool() as active:
                return active
            case 0:
                raise UnspecifiedEnumValueError(
                    self, "_active", f"status of microgrid {self} is unspecified"
                )
            case int() as value:
                raise UnrecognizedEnumValueError(
                    self,
                    "_active",
                    value,
                    f"unrecognized status of microgrid {self}: {value}",
                )
            case unknown:
                assert_never(unknown)

    def get_delivery_area(self) -> DeliveryArea:
        """Return the delivery area as a well-formed `DeliveryArea`.

        This is the higher-level accessor for the [`delivery_area`][..delivery_area]
        attribute: it resolves the field to a valid
        [`DeliveryArea`][....grid.DeliveryArea] or raises a clear, catchable error.

        Returns:
            The delivery area, when it is a well-formed
                [`DeliveryArea`][....grid.DeliveryArea].

        Raises:
            MissingFieldError: If the delivery area is not set (`None`).
            InvalidDeliveryAreaError: If the delivery area is an
                [`InvalidDeliveryArea`][....grid.InvalidDeliveryArea]. The
                offending instance is available on the exception's
                `delivery_area` attribute.
        """
        match self.delivery_area:
            case None:
                raise MissingFieldError(self, "delivery_area")
            case InvalidDeliveryArea() as invalid:
                raise InvalidDeliveryAreaError(self, "delivery_area", invalid)
            case DeliveryArea() as valid:
                return valid
            case unknown:
                assert_never(unknown)

    def get_delivery_area_or_none(self) -> DeliveryArea | None:
        """Return the delivery area as a well-formed `DeliveryArea`, or `None`.

        This is the higher-level accessor for the [`delivery_area`][..delivery_area]
        attribute that tolerates a missing field: it resolves the field to a
        valid [`DeliveryArea`][....grid.DeliveryArea], returns `None` when the
        field was not set on the wire, or raises a clear, catchable error when
        the field carries an invalid delivery area.

        Returns:
            The delivery area when it is a well-formed
                [`DeliveryArea`][....grid.DeliveryArea], or `None` when it is
                not set.

        Raises:
            InvalidDeliveryAreaError: If the delivery area is an
                [`InvalidDeliveryArea`][....grid.InvalidDeliveryArea]. The
                offending instance is available on the exception's
                `delivery_area` attribute.
        """
        match self.delivery_area:
            case None:
                return None
            case InvalidDeliveryArea() as invalid:
                raise InvalidDeliveryAreaError(self, "delivery_area", invalid)
            case DeliveryArea() as valid:
                return valid
            case unknown:
                assert_never(unknown)

    def get_location(self) -> Location:
        """Return the location as a [`Location`][....types.Location].

        This is the higher-level accessor for the [`location`][..location]
        attribute: it resolves the field to a
        [`Location`][....types.Location] or raises a clear, catchable error.

        The returned instance may still carry raw wire values that fail the
        [`Location`][....types.Location] field invariants; use its own
        `get_*()` accessors to obtain validated coordinates and country code.

        Returns:
            The location, when it is set.

        Raises:
            MissingFieldError: If the location is not set (`None`).
        """
        if self.location is None:
            raise MissingFieldError(self, "location")
        return self.location

    def __str__(self) -> str:
        """Return the ID of this microgrid as a string."""
        name = f":{self.name}" if self.name else ""
        return f"{self.id}{name}"
Attributes¤
create_time instance-attribute ¤
create_time: datetime

The UTC timestamp indicating when the microgrid was initially created.

delivery_area instance-attribute ¤
delivery_area: DeliveryArea | InvalidDeliveryArea | None

The delivery area where the microgrid is located.

None means the field was not set on the wire. An InvalidDeliveryArea means the wire carried a delivery area that fails its invariants.

Tip

This is the lower-level field; prefer get_delivery_area() to obtain a valid DeliveryArea or a clear error.

enterprise_id instance-attribute ¤
enterprise_id: EnterpriseId

The unique identifier linking this microgrid to its parent enterprise account.

id instance-attribute ¤

The unique identifier of the microgrid.

location instance-attribute ¤
location: Location | None

The physical location of the microgrid, in geographical co-ordinates.

name instance-attribute ¤
name: str

The name of the microgrid.

Methods:¤
__post_init__ ¤
__post_init__() -> None

Reject direct construction of this read-only type.

RAISES DESCRIPTION
TypeError

If the instance was not created via the microgrid_from_proto converter.

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def __post_init__(self) -> None:
    """Reject direct construction of this read-only type.

    Raises:
        TypeError: If the instance was not created via the
            [`microgrid_from_proto`][...proto.v1alpha8.microgrid_from_proto]
            converter.
    """
    if not self._allow_construction:
        raise TypeError(
            f"{type(self).__name__} cannot be constructed directly; obtain "
            "instances via the microgrid_from_proto converter."
        )
__str__ ¤
__str__() -> str

Return the ID of this microgrid as a string.

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def __str__(self) -> str:
    """Return the ID of this microgrid as a string."""
    name = f":{self.name}" if self.name else ""
    return f"{self.id}{name}"
get_delivery_area ¤
get_delivery_area() -> DeliveryArea

Return the delivery area as a well-formed DeliveryArea.

This is the higher-level accessor for the delivery_area attribute: it resolves the field to a valid DeliveryArea or raises a clear, catchable error.

RETURNS DESCRIPTION
DeliveryArea

The delivery area, when it is a well-formed DeliveryArea.

RAISES DESCRIPTION
MissingFieldError

If the delivery area is not set (None).

InvalidDeliveryAreaError

If the delivery area is an InvalidDeliveryArea. The offending instance is available on the exception's delivery_area attribute.

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def get_delivery_area(self) -> DeliveryArea:
    """Return the delivery area as a well-formed `DeliveryArea`.

    This is the higher-level accessor for the [`delivery_area`][..delivery_area]
    attribute: it resolves the field to a valid
    [`DeliveryArea`][....grid.DeliveryArea] or raises a clear, catchable error.

    Returns:
        The delivery area, when it is a well-formed
            [`DeliveryArea`][....grid.DeliveryArea].

    Raises:
        MissingFieldError: If the delivery area is not set (`None`).
        InvalidDeliveryAreaError: If the delivery area is an
            [`InvalidDeliveryArea`][....grid.InvalidDeliveryArea]. The
            offending instance is available on the exception's
            `delivery_area` attribute.
    """
    match self.delivery_area:
        case None:
            raise MissingFieldError(self, "delivery_area")
        case InvalidDeliveryArea() as invalid:
            raise InvalidDeliveryAreaError(self, "delivery_area", invalid)
        case DeliveryArea() as valid:
            return valid
        case unknown:
            assert_never(unknown)
get_delivery_area_or_none ¤
get_delivery_area_or_none() -> DeliveryArea | None

Return the delivery area as a well-formed DeliveryArea, or None.

This is the higher-level accessor for the delivery_area attribute that tolerates a missing field: it resolves the field to a valid DeliveryArea, returns None when the field was not set on the wire, or raises a clear, catchable error when the field carries an invalid delivery area.

RETURNS DESCRIPTION
DeliveryArea | None

The delivery area when it is a well-formed DeliveryArea, or None when it is not set.

RAISES DESCRIPTION
InvalidDeliveryAreaError

If the delivery area is an InvalidDeliveryArea. The offending instance is available on the exception's delivery_area attribute.

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def get_delivery_area_or_none(self) -> DeliveryArea | None:
    """Return the delivery area as a well-formed `DeliveryArea`, or `None`.

    This is the higher-level accessor for the [`delivery_area`][..delivery_area]
    attribute that tolerates a missing field: it resolves the field to a
    valid [`DeliveryArea`][....grid.DeliveryArea], returns `None` when the
    field was not set on the wire, or raises a clear, catchable error when
    the field carries an invalid delivery area.

    Returns:
        The delivery area when it is a well-formed
            [`DeliveryArea`][....grid.DeliveryArea], or `None` when it is
            not set.

    Raises:
        InvalidDeliveryAreaError: If the delivery area is an
            [`InvalidDeliveryArea`][....grid.InvalidDeliveryArea]. The
            offending instance is available on the exception's
            `delivery_area` attribute.
    """
    match self.delivery_area:
        case None:
            return None
        case InvalidDeliveryArea() as invalid:
            raise InvalidDeliveryAreaError(self, "delivery_area", invalid)
        case DeliveryArea() as valid:
            return valid
        case unknown:
            assert_never(unknown)
get_location ¤
get_location() -> Location

Return the location as a Location.

This is the higher-level accessor for the location attribute: it resolves the field to a Location or raises a clear, catchable error.

The returned instance may still carry raw wire values that fail the Location field invariants; use its own get_*() accessors to obtain validated coordinates and country code.

RETURNS DESCRIPTION
Location

The location, when it is set.

RAISES DESCRIPTION
MissingFieldError

If the location is not set (None).

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def get_location(self) -> Location:
    """Return the location as a [`Location`][....types.Location].

    This is the higher-level accessor for the [`location`][..location]
    attribute: it resolves the field to a
    [`Location`][....types.Location] or raises a clear, catchable error.

    The returned instance may still carry raw wire values that fail the
    [`Location`][....types.Location] field invariants; use its own
    `get_*()` accessors to obtain validated coordinates and country code.

    Returns:
        The location, when it is set.

    Raises:
        MissingFieldError: If the location is not set (`None`).
    """
    if self.location is None:
        raise MissingFieldError(self, "location")
    return self.location
is_active ¤
is_active() -> bool

Return whether the microgrid is active.

RETURNS DESCRIPTION
bool

Whether the microgrid is active.

RAISES DESCRIPTION
UnspecifiedEnumValueError

If the status is unspecified.

UnrecognizedEnumValueError

If the status is not recognized. The raw status value is available on the error's value attribute.

Source code in src/frequenz/client/common/microgrid/_microgrid.py
def is_active(self) -> bool:
    """Return whether the microgrid is active.

    Returns:
        Whether the microgrid is active.

    Raises:
        UnspecifiedEnumValueError: If the status is unspecified.
        UnrecognizedEnumValueError: If the status is not recognized. The raw
            status value is available on the error's `value` attribute.
    """
    match self._active:
        case bool() as active:
            return active
        case 0:
            raise UnspecifiedEnumValueError(
                self, "_active", f"status of microgrid {self} is unspecified"
            )
        case int() as value:
            raise UnrecognizedEnumValueError(
                self,
                "_active",
                value,
                f"unrecognized status of microgrid {self}: {value}",
            )
        case unknown:
            assert_never(unknown)

frequenz.client.common.microgrid.MicrogridId ¤

Bases: BaseId

A unique identifier for a microgrid.

Source code in src/frequenz/client/common/microgrid/_ids.py
@final
class MicrogridId(BaseId, str_prefix="MID"):
    """A unique identifier for a microgrid."""
Attributes¤
str_prefix property ¤
str_prefix: str

The prefix used for the string representation of this ID.

Methods:¤
__eq__ ¤
__eq__(other: object) -> bool

Check if this instance is equal to another object.

Equality is defined as being of the exact same type and having the same underlying ID.

PARAMETER DESCRIPTION
other

The object to compare against.

TYPE: object

RETURNS DESCRIPTION
bool

True if other is of the same type and has the same underlying ID,

bool

NotImplemented if other is of a different type.

Source code in frequenz/core/id.py
def __eq__(self, other: object) -> bool:
    """Check if this instance is equal to another object.

    Equality is defined as being of the exact same type and having the same
    underlying ID.

    Args:
        other: The object to compare against.

    Returns:
        True if `other` is of the same type and has the same underlying ID,
        `NotImplemented` if `other` is of a different type.
    """
    # pylint thinks this is not an unidiomatic typecheck, but in this case
    # it is not. isinstance() returns True for subclasses, which is not
    # what we want here, as different ID types should never be equal.
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    # We already checked type(other) is type(self), but mypy doesn't
    # understand that, so we need to cast it to Self.
    other_id = cast(Self, other)
    return self._id == other_id._id
__hash__ ¤
__hash__() -> int

Return the hash of this instance.

The hash is based on the exact type and the underlying ID to ensure that IDs of different types but with the same numeric value have different hashes.

RETURNS DESCRIPTION
int

The hash of this instance.

Source code in frequenz/core/id.py
def __hash__(self) -> int:
    """Return the hash of this instance.

    The hash is based on the exact type and the underlying ID to ensure
    that IDs of different types but with the same numeric value have different hashes.

    Returns:
        The hash of this instance.
    """
    return hash((type(self), self._id))
__init__ ¤
__init__(id_: int) -> None

Initialize this instance.

PARAMETER DESCRIPTION
id_

The numeric unique identifier.

TYPE: int

RAISES DESCRIPTION
ValueError

If the ID is negative.

Source code in frequenz/core/id.py
def __init__(self, id_: int, /) -> None:
    """Initialize this instance.

    Args:
        id_: The numeric unique identifier.

    Raises:
        ValueError: If the ID is negative.
    """
    if id_ < 0:
        raise ValueError(f"{type(self).__name__} can't be negative.")
    self._id = id_
__init_subclass__ ¤
__init_subclass__(
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any
) -> None

Initialize a subclass, set its string prefix, and perform checks.

PARAMETER DESCRIPTION
str_prefix

The string prefix for the ID type (e.g., "MID"). Must be unique across all ID types.

TYPE: str

allow_custom_name

If True, bypasses the check that the class name must end with "Id". Defaults to False.

TYPE: bool DEFAULT: False

**kwargs

Forwarded to the parent's init_subclass.

TYPE: Any DEFAULT: {}

RAISES DESCRIPTION
TypeError

If allow_custom_name is False and the class name does not end with "Id".

Source code in frequenz/core/id.py
def __init_subclass__(
    cls,
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any,
) -> None:
    """Initialize a subclass, set its string prefix, and perform checks.

    Args:
        str_prefix: The string prefix for the ID type (e.g., "MID").
            Must be unique across all ID types.
        allow_custom_name: If True, bypasses the check that the class name
            must end with "Id". Defaults to False.
        **kwargs: Forwarded to the parent's __init_subclass__.

    Raises:
        TypeError: If `allow_custom_name` is False and the class name
            does not end with "Id".
    """
    super().__init_subclass__(**kwargs)

    if str_prefix in BaseId._registered_prefixes:
        # We want to raise an exception here, but currently can't due to
        # https://github.com/frequenz-floss/frequenz-repo-config-python/issues/421
        _logger.warning(
            "Prefix '%s' is already registered. ID prefixes must be unique.",
            str_prefix,
        )
    BaseId._registered_prefixes.add(str_prefix)

    if not allow_custom_name and not cls.__name__.endswith("Id"):
        raise TypeError(
            f"Class name '{cls.__name__}' for an ID class must end with 'Id' "
            "(e.g., 'SomeId'), or use `allow_custom_name=True`."
        )

    cls._str_prefix = str_prefix
__int__ ¤
__int__() -> int

Return the numeric ID of this instance.

Source code in frequenz/core/id.py
def __int__(self) -> int:
    """Return the numeric ID of this instance."""
    return self._id
__lt__ ¤
__lt__(other: object) -> bool

Check if this instance is less than another object.

Comparison is only defined between instances of the exact same type.

PARAMETER DESCRIPTION
other

The object to compare against.

TYPE: object

RETURNS DESCRIPTION
bool

True if this instance is less than other, NotImplemented if

bool

other is of a different type.

Source code in frequenz/core/id.py
def __lt__(self, other: object) -> bool:
    """Check if this instance is less than another object.

    Comparison is only defined between instances of the exact same type.

    Args:
        other: The object to compare against.

    Returns:
        True if this instance is less than `other`, `NotImplemented` if
        `other` is of a different type.
    """
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    other_id = cast(Self, other)
    return self._id < other_id._id
__new__ ¤
__new__(*_: Any, **__: Any) -> Self

Create a new instance of the ID class, only if it is a subclass of BaseId.

Source code in frequenz/core/id.py
def __new__(cls, *_: Any, **__: Any) -> Self:
    """Create a new instance of the ID class, only if it is a subclass of BaseId."""
    if cls is BaseId:
        raise TypeError("BaseId cannot be instantiated directly. Use a subclass.")
    return super().__new__(cls)
__repr__ ¤
__repr__() -> str

Return the string representation of this instance.

Source code in frequenz/core/id.py
def __repr__(self) -> str:
    """Return the string representation of this instance."""
    return f"{type(self).__name__}({self._id!r})"
__str__ ¤
__str__() -> str

Return the short string representation of this instance.

Source code in frequenz/core/id.py
def __str__(self) -> str:
    """Return the short string representation of this instance."""
    return f"{self._str_prefix}{self._id}"