Skip to content

Index

frequenz.client.assets ¤

Assets API client.

Classes¤

frequenz.client.assets.AssetsApiClient ¤

Bases: BaseApiClient[PlatformAssetsStub]

A client for the Assets API.

Source code in frequenz/client/assets/_client.py
class AssetsApiClient(
    BaseApiClient[assets_pb2_grpc.PlatformAssetsStub]
):  # pylint: disable=too-many-arguments
    """A client for the Assets API."""

    def __init__(
        self,
        server_url: str,
        *,
        auth_key: str | None = None,
        sign_secret: str | None = None,
        channel_defaults: channel.ChannelOptions = channel.ChannelOptions(),
        connect: bool = True,
    ) -> None:
        """
        Initialize the AssetsApiClient.

        Args:
            server_url: The location of the microgrid API server in the form of a URL.
                The following format is expected:
                "grpc://hostname{:`port`}{?ssl=`ssl`}",
                where the `port` should be an int between 0 and 65535 (defaulting to
                9090) and `ssl` should be a boolean (defaulting to `true`).
                For example: `grpc://localhost:1090?ssl=true`.
            auth_key: The authentication key to use for the connection.
            sign_secret: The secret to use for signing requests.
            channel_defaults: The default options use to create the channel when not
                specified in the URL.
            connect: Whether to connect to the server as soon as a client instance is
                created. If `False`, the client will not connect to the server until
                [connect()][frequenz.client.base.client.BaseApiClient.connect] is
                called.
        """
        super().__init__(
            server_url,
            assets_pb2_grpc.PlatformAssetsStub,
            connect=connect,
            channel_defaults=channel_defaults,
            auth_key=auth_key,
            sign_secret=sign_secret,
        )

    @property
    def stub(self) -> assets_pb2_grpc.PlatformAssetsAsyncStub:
        """
        The gRPC stub for the Assets API.

        Returns:
            The gRPC stub for the Assets API.

        Raises:
            ClientNotConnected: If the client is not connected to the server.
        """
        if self._channel is None or self._stub is None:
            raise ClientNotConnected(server_url=self.server_url, operation="stub")
        # This type: ignore is needed because the stub is a sync stub, but we need to
        # use the async stub, so we cast the sync stub to the async stub.
        return self._stub  # type: ignore

    async def get_microgrid(  # noqa: DOC502 (raises ApiClientError indirectly)
        self, microgrid_id: int
    ) -> Microgrid:
        """
        Get the details of a microgrid.

        Args:
            microgrid_id: The ID of the microgrid to get the details of.

        Returns:
            The details of the microgrid.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        """
        response = await call_stub_method(
            self,
            lambda: self.stub.GetMicrogrid(
                assets_pb2.GetMicrogridRequest(microgrid_id=microgrid_id),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="GetMicrogrid",
        )

        return microgrid_from_proto(response.microgrid)

    async def list_microgrid_electrical_components(
        self, microgrid_id: int
    ) -> list[ElectricalComponent]:
        """
        Get the electrical components of a microgrid.

        Args:
            microgrid_id: The ID of the microgrid to get the electrical components of.

        Returns:
            The electrical components of the microgrid.
        """
        response = await call_stub_method(
            self,
            lambda: self.stub.ListMicrogridElectricalComponents(
                assets_pb2.ListMicrogridElectricalComponentsRequest(
                    microgrid_id=microgrid_id,
                ),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="ListMicrogridElectricalComponents",
        )

        return [
            electrical_component_proto(component) for component in response.components
        ]
Attributes¤
channel property ¤
channel: Channel

The underlying gRPC channel used to communicate with the server.

Warning

This channel is provided as a last resort for advanced users. It is not recommended to use this property directly unless you know what you are doing and you don't care about being tied to a specific gRPC library.

RAISES DESCRIPTION
ClientNotConnected

If the client is not connected to the server.

channel_defaults property ¤
channel_defaults: ChannelOptions

The default options for the gRPC channel.

is_connected property ¤
is_connected: bool

Whether the client is connected to the server.

server_url property ¤
server_url: str

The URL of the server.

stub property ¤
stub: PlatformAssetsAsyncStub

The gRPC stub for the Assets API.

RETURNS DESCRIPTION
PlatformAssetsAsyncStub

The gRPC stub for the Assets API.

RAISES DESCRIPTION
ClientNotConnected

If the client is not connected to the server.

Functions¤
__aenter__ async ¤
__aenter__() -> Self

Enter a context manager.

Source code in frequenz/client/base/client.py
async def __aenter__(self) -> Self:
    """Enter a context manager."""
    self.connect()
    return self
__aexit__ async ¤
__aexit__(
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: Any | None,
) -> bool | None

Exit a context manager.

Source code in frequenz/client/base/client.py
async def __aexit__(
    self,
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: Any | None,
) -> bool | None:
    """Exit a context manager."""
    if self._channel is None:
        return None
    result = await self._channel.__aexit__(_exc_type, _exc_val, _exc_tb)
    self._channel = None
    self._stub = None
    return result
__init__ ¤
__init__(
    server_url: str,
    *,
    auth_key: str | None = None,
    sign_secret: str | None = None,
    channel_defaults: ChannelOptions = ChannelOptions(),
    connect: bool = True
) -> None

Initialize the AssetsApiClient.

PARAMETER DESCRIPTION
server_url

The location of the microgrid API server in the form of a URL. The following format is expected: "grpc://hostname{:port}{?ssl=ssl}", where the port should be an int between 0 and 65535 (defaulting to 9090) and ssl should be a boolean (defaulting to true). For example: grpc://localhost:1090?ssl=true.

TYPE: str

auth_key

The authentication key to use for the connection.

TYPE: str | None DEFAULT: None

sign_secret

The secret to use for signing requests.

TYPE: str | None DEFAULT: None

channel_defaults

The default options use to create the channel when not specified in the URL.

TYPE: ChannelOptions DEFAULT: ChannelOptions()

connect

Whether to connect to the server as soon as a client instance is created. If False, the client will not connect to the server until connect() is called.

TYPE: bool DEFAULT: True

Source code in frequenz/client/assets/_client.py
def __init__(
    self,
    server_url: str,
    *,
    auth_key: str | None = None,
    sign_secret: str | None = None,
    channel_defaults: channel.ChannelOptions = channel.ChannelOptions(),
    connect: bool = True,
) -> None:
    """
    Initialize the AssetsApiClient.

    Args:
        server_url: The location of the microgrid API server in the form of a URL.
            The following format is expected:
            "grpc://hostname{:`port`}{?ssl=`ssl`}",
            where the `port` should be an int between 0 and 65535 (defaulting to
            9090) and `ssl` should be a boolean (defaulting to `true`).
            For example: `grpc://localhost:1090?ssl=true`.
        auth_key: The authentication key to use for the connection.
        sign_secret: The secret to use for signing requests.
        channel_defaults: The default options use to create the channel when not
            specified in the URL.
        connect: Whether to connect to the server as soon as a client instance is
            created. If `False`, the client will not connect to the server until
            [connect()][frequenz.client.base.client.BaseApiClient.connect] is
            called.
    """
    super().__init__(
        server_url,
        assets_pb2_grpc.PlatformAssetsStub,
        connect=connect,
        channel_defaults=channel_defaults,
        auth_key=auth_key,
        sign_secret=sign_secret,
    )
connect ¤
connect(
    server_url: str | None = None,
    *,
    auth_key: str | None | EllipsisType = ...,
    sign_secret: str | None | EllipsisType = ...
) -> None

Connect to the server, possibly using a new URL.

If the client is already connected and the URL is the same as the previous URL, this method does nothing. If you want to force a reconnection, you can call disconnect() first.

PARAMETER DESCRIPTION
server_url

The URL of the server to connect to. If not provided, the previously used URL is used.

TYPE: str | None DEFAULT: None

auth_key

The API key to use when connecting to the service. If an Ellipsis is provided, the previously used auth_key is used.

TYPE: str | None | EllipsisType DEFAULT: ...

sign_secret

The secret to use when creating message HMAC. If an Ellipsis is provided,

TYPE: str | None | EllipsisType DEFAULT: ...

Source code in frequenz/client/base/client.py
def connect(
    self,
    server_url: str | None = None,
    *,
    auth_key: str | None | EllipsisType = ...,
    sign_secret: str | None | EllipsisType = ...,
) -> None:
    """Connect to the server, possibly using a new URL.

    If the client is already connected and the URL is the same as the previous URL,
    this method does nothing. If you want to force a reconnection, you can call
    [disconnect()][frequenz.client.base.client.BaseApiClient.disconnect] first.

    Args:
        server_url: The URL of the server to connect to. If not provided, the
            previously used URL is used.
        auth_key: The API key to use when connecting to the service. If an Ellipsis
            is provided, the previously used auth_key is used.
        sign_secret: The secret to use when creating message HMAC. If an Ellipsis is
            provided,
    """
    reconnect = False
    if server_url is not None and server_url != self._server_url:  # URL changed
        self._server_url = server_url
        reconnect = True
    if auth_key is not ... and auth_key != self._auth_key:
        self._auth_key = auth_key
        reconnect = True
    if sign_secret is not ... and sign_secret != self._sign_secret:
        self._sign_secret = sign_secret
        reconnect = True
    if self.is_connected and not reconnect:  # Desired connection already exists
        return

    interceptors: list[ClientInterceptor] = []
    if self._auth_key is not None:
        interceptors += [
            AuthenticationInterceptorUnaryUnary(self._auth_key),  # type: ignore [list-item]
            AuthenticationInterceptorUnaryStream(self._auth_key),  # type: ignore [list-item]
        ]
    if self._sign_secret is not None:
        interceptors += [
            SigningInterceptorUnaryUnary(self._sign_secret),  # type: ignore [list-item]
            SigningInterceptorUnaryStream(self._sign_secret),  # type: ignore [list-item]
        ]

    self._channel = parse_grpc_uri(
        self._server_url,
        interceptors,
        defaults=self._channel_defaults,
    )
    self._stub = self._create_stub(self._channel)
disconnect async ¤
disconnect() -> None

Disconnect from the server.

If the client is not connected, this method does nothing.

Source code in frequenz/client/base/client.py
async def disconnect(self) -> None:
    """Disconnect from the server.

    If the client is not connected, this method does nothing.
    """
    await self.__aexit__(None, None, None)
get_microgrid async ¤
get_microgrid(microgrid_id: int) -> Microgrid

Get the details of a microgrid.

PARAMETER DESCRIPTION
microgrid_id

The ID of the microgrid to get the details of.

TYPE: int

RETURNS DESCRIPTION
Microgrid

The details of the microgrid.

RAISES DESCRIPTION
ApiClientError

If there are any errors communicating with the Assets API, most likely a subclass of GrpcError.

Source code in frequenz/client/assets/_client.py
async def get_microgrid(  # noqa: DOC502 (raises ApiClientError indirectly)
    self, microgrid_id: int
) -> Microgrid:
    """
    Get the details of a microgrid.

    Args:
        microgrid_id: The ID of the microgrid to get the details of.

    Returns:
        The details of the microgrid.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
    """
    response = await call_stub_method(
        self,
        lambda: self.stub.GetMicrogrid(
            assets_pb2.GetMicrogridRequest(microgrid_id=microgrid_id),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="GetMicrogrid",
    )

    return microgrid_from_proto(response.microgrid)
list_microgrid_electrical_components async ¤
list_microgrid_electrical_components(
    microgrid_id: int,
) -> list[ElectricalComponent]

Get the electrical components of a microgrid.

PARAMETER DESCRIPTION
microgrid_id

The ID of the microgrid to get the electrical components of.

TYPE: int

RETURNS DESCRIPTION
list[ElectricalComponent]

The electrical components of the microgrid.

Source code in frequenz/client/assets/_client.py
async def list_microgrid_electrical_components(
    self, microgrid_id: int
) -> list[ElectricalComponent]:
    """
    Get the electrical components of a microgrid.

    Args:
        microgrid_id: The ID of the microgrid to get the electrical components of.

    Returns:
        The electrical components of the microgrid.
    """
    response = await call_stub_method(
        self,
        lambda: self.stub.ListMicrogridElectricalComponents(
            assets_pb2.ListMicrogridElectricalComponentsRequest(
                microgrid_id=microgrid_id,
            ),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="ListMicrogridElectricalComponents",
    )

    return [
        electrical_component_proto(component) for component in response.components
    ]

frequenz.client.assets.DeliveryArea dataclass ¤

A geographical or administrative region where electricity deliveries occur.

DeliveryArea represents the geographical or administrative region, usually defined and maintained by a Transmission System Operator (TSO), where electricity deliveries for a contract occur.

The concept is important to energy trading as it delineates the agreed-upon delivery location. Delivery areas can have different codes based on the jurisdiction in which they operate.

Jurisdictional Differences

This is typically represented by specific codes according to local jurisdiction.

In Europe, this is represented by an EIC (Energy Identification Code). List of EICs.

Source code in frequenz/client/assets/_delivery_area.py
@dataclass(frozen=True, kw_only=True)
class DeliveryArea:
    """A geographical or administrative region where electricity deliveries occur.

    DeliveryArea represents the geographical or administrative region, usually defined
    and maintained by a Transmission System Operator (TSO), where electricity deliveries
    for a contract occur.

    The concept is important to energy trading as it delineates the agreed-upon delivery
    location. Delivery areas can have different codes based on the jurisdiction in
    which they operate.

    Note: Jurisdictional Differences
        This is typically represented by specific codes according to local jurisdiction.

        In Europe, this is represented by an
        [EIC](https://en.wikipedia.org/wiki/Energy_Identification_Code) (Energy
        Identification Code). [List of
        EICs](https://www.entsoe.eu/data/energy-identification-codes-eic/eic-approved-codes/).
    """

    code: str | None
    """The code representing the unique identifier for the delivery area."""

    code_type: EnergyMarketCodeType | int
    """Type of code used for identifying the delivery area itself.

    This code could be extended in the future, in case an unknown code type is
    encountered, a plain integer value is used to represent it.
    """

    def __str__(self) -> str:
        """Return a human-readable string representation of this instance."""
        code = self.code or "<NO CODE>"
        code_type = (
            f"type={self.code_type}"
            if isinstance(self.code_type, int)
            else self.code_type.name
        )
        return f"{code}[{code_type}]"
Attributes¤
code instance-attribute ¤
code: str | None

The code representing the unique identifier for the delivery area.

code_type instance-attribute ¤

Type of code used for identifying the delivery area itself.

This code could be extended in the future, in case an unknown code type is encountered, a plain integer value is used to represent it.

Functions¤
__str__ ¤
__str__() -> str

Return a human-readable string representation of this instance.

Source code in frequenz/client/assets/_delivery_area.py
def __str__(self) -> str:
    """Return a human-readable string representation of this instance."""
    code = self.code or "<NO CODE>"
    code_type = (
        f"type={self.code_type}"
        if isinstance(self.code_type, int)
        else self.code_type.name
    )
    return f"{code}[{code_type}]"

frequenz.client.assets.EnergyMarketCodeType ¤

Bases: Enum

The identification code types used in the energy market.

CodeType specifies the type of identification code used for uniquely identifying various entities such as delivery areas, market participants, and grid components within the energy market.

This enumeration aims to offer compatibility across different jurisdictional standards.

Understanding Code Types

Different regions or countries may have their own standards for uniquely identifying various entities within the energy market. For example, in Europe, the Energy Identification Code (EIC) is commonly used for this purpose.

Extensibility

New code types can be added to this enum to accommodate additional regional standards, enhancing the API's adaptability.

Validation Required

The chosen code type should correspond correctly with the code field in the relevant message objects, such as DeliveryArea or Counterparty. Failure to match the code type with the correct code could lead to processing errors.

Source code in frequenz/client/assets/_delivery_area.py
@enum.unique
class EnergyMarketCodeType(enum.Enum):
    """The identification code types used in the energy market.

    CodeType specifies the type of identification code used for uniquely
    identifying various entities such as delivery areas, market participants,
    and grid components within the energy market.

    This enumeration aims to
    offer compatibility across different jurisdictional standards.

    Note: Understanding Code Types
        Different regions or countries may have their own standards for uniquely
        identifying various entities within the energy market. For example, in
        Europe, the Energy Identification Code (EIC) is commonly used for this
        purpose.

    Note: Extensibility
        New code types can be added to this enum to accommodate additional regional
        standards, enhancing the API's adaptability.

    Danger: Validation Required
        The chosen code type should correspond correctly with the `code` field in
        the relevant message objects, such as `DeliveryArea` or `Counterparty`.
        Failure to match the code type with the correct code could lead to
        processing errors.
    """

    UNSPECIFIED = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_UNSPECIFIED
    """Unspecified type. This value is a placeholder and should not be used."""

    EUROPE_EIC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC
    """European Energy Identification Code Standard."""

    US_NERC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_US_NERC
    """North American Electric Reliability Corporation identifiers."""
Attributes¤
EUROPE_EIC class-attribute instance-attribute ¤
EUROPE_EIC = ENERGY_MARKET_CODE_TYPE_EUROPE_EIC

European Energy Identification Code Standard.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = ENERGY_MARKET_CODE_TYPE_UNSPECIFIED

Unspecified type. This value is a placeholder and should not be used.

US_NERC class-attribute instance-attribute ¤
US_NERC = ENERGY_MARKET_CODE_TYPE_US_NERC

North American Electric Reliability Corporation identifiers.

frequenz.client.assets.Location dataclass ¤

A location of a microgrid.

Source code in frequenz/client/assets/_location.py
@dataclass(frozen=True, kw_only=True)
class Location:
    """A location of a microgrid."""

    latitude: float | None
    """The latitude of the microgrid in degree."""

    longitude: float | None
    """The longitude of the microgrid in degree."""

    country_code: str | None
    """The country code of the microgrid in ISO 3166-1 Alpha 2 format."""

    def __str__(self) -> str:
        """Return the short string representation of this instance."""
        country = self.country_code or "<NO COUNTRY CODE>"
        lat = f"{self.latitude:.2f}" if self.latitude is not None else "?"
        lon = f"{self.longitude:.2f}" if self.longitude is not None else "?"
        coordinates = ""
        if self.latitude is not None or self.longitude is not None:
            coordinates = f":({lat}, {lon})"
        return f"{country}{coordinates}"
Attributes¤
country_code instance-attribute ¤
country_code: str | None

The country code of the microgrid in ISO 3166-1 Alpha 2 format.

latitude instance-attribute ¤
latitude: float | None

The latitude of the microgrid in degree.

longitude instance-attribute ¤
longitude: float | None

The longitude of the microgrid in degree.

Functions¤
__str__ ¤
__str__() -> str

Return the short string representation of this instance.

Source code in frequenz/client/assets/_location.py
def __str__(self) -> str:
    """Return the short string representation of this instance."""
    country = self.country_code or "<NO COUNTRY CODE>"
    lat = f"{self.latitude:.2f}" if self.latitude is not None else "?"
    lon = f"{self.longitude:.2f}" if self.longitude is not None else "?"
    coordinates = ""
    if self.latitude is not None or self.longitude is not None:
        coordinates = f":({lat}, {lon})"
    return f"{country}{coordinates}"

frequenz.client.assets.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 frequenz/client/assets/_microgrid.py
@dataclass(frozen=True, kw_only=True)
class Microgrid:
    """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 | None
    """Name of the microgrid."""

    delivery_area: DeliveryArea | None
    """The delivery area where the microgrid is located, as identified by a specific code."""

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

    status: MicrogridStatus | int
    """The current status of the microgrid."""

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

    @cached_property
    def is_active(self) -> bool:
        """Whether the microgrid is active."""
        if self.status is MicrogridStatus.UNSPECIFIED:
            # Because this is a cached property, the warning will only be logged once.
            _logger.warning(
                "Microgrid %s has an unspecified status. Assuming it is active.", self
            )
        return self.status in (MicrogridStatus.ACTIVE, MicrogridStatus.UNSPECIFIED)

    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_timestamp instance-attribute ¤
create_timestamp: datetime

The UTC timestamp indicating when the microgrid was initially created.

delivery_area instance-attribute ¤
delivery_area: DeliveryArea | None

The delivery area where the microgrid is located, as identified by a specific code.

enterprise_id instance-attribute ¤
enterprise_id: EnterpriseId

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

id instance-attribute ¤
id: MicrogridId

The unique identifier of the microgrid.

is_active cached property ¤
is_active: bool

Whether the microgrid is active.

location instance-attribute ¤
location: Location | None

Physical location of the microgrid, in geographical co-ordinates.

name instance-attribute ¤
name: str | None

Name of the microgrid.

status instance-attribute ¤
status: MicrogridStatus | int

The current status of the microgrid.

Functions¤
__str__ ¤
__str__() -> str

Return the ID of this microgrid as a string.

Source code in frequenz/client/assets/_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}"

frequenz.client.assets.MicrogridStatus ¤

Bases: Enum

The possible statuses for a microgrid.

Source code in frequenz/client/assets/_microgrid.py
@enum.unique
class MicrogridStatus(enum.Enum):
    """The possible statuses for a microgrid."""

    UNSPECIFIED = microgrid_pb2.MICROGRID_STATUS_UNSPECIFIED
    """The status is unspecified. This should not be used."""

    ACTIVE = microgrid_pb2.MICROGRID_STATUS_ACTIVE
    """The microgrid is active."""

    INACTIVE = microgrid_pb2.MICROGRID_STATUS_INACTIVE
    """The microgrid is inactive."""
Attributes¤
ACTIVE class-attribute instance-attribute ¤
ACTIVE = MICROGRID_STATUS_ACTIVE

The microgrid is active.

INACTIVE class-attribute instance-attribute ¤
INACTIVE = MICROGRID_STATUS_INACTIVE

The microgrid is inactive.

UNSPECIFIED class-attribute instance-attribute ¤
UNSPECIFIED = MICROGRID_STATUS_UNSPECIFIED

The status is unspecified. This should not be used.