Index

frequenz.client.assets

Assets API client.

Classes

frequenz.client.assets.AssetsApiClient

Bases: BaseApiClient[PlatformAssetsStub]

A client for the Assets API.

Source code in src/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,DOC503 (raises indirectly)
        self,
        microgrid_id: MicrogridId,
        *,
        raise_on_errors: bool = False,
    ) -> Microgrid:
        """
        Get the details of a microgrid.

        Args:
            microgrid_id: The ID of the microgrid to get the details of.
            raise_on_errors: If True, raise an
                [InvalidMicrogridError][frequenz.client.assets.exceptions.InvalidMicrogridError]
                when major validation issues are found instead of just
                logging them.

        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].
            InvalidMicrogridError: If `raise_on_errors` is True and major
                validation issues are found.
        """
        response = await call_stub_method(
            self,
            lambda: self.stub.GetMicrogrid(
                assets_pb2.GetMicrogridRequest(microgrid_id=int(microgrid_id)),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="GetMicrogrid",
        )

        if raise_on_errors:
            major_issues: list[str] = []
            minor_issues: list[str] = []
            microgrid = microgrid_from_proto_with_issues(
                response.microgrid,
                major_issues=major_issues,
                minor_issues=minor_issues,
            )
            if major_issues:
                raise InvalidMicrogridError(
                    microgrid=microgrid,
                    major_issues=major_issues,
                    minor_issues=minor_issues,
                    raw_message=response.microgrid,
                )
            return microgrid

        return microgrid_from_proto(response.microgrid)

    async def list_microgrid_electrical_components(
        self,
        microgrid_id: MicrogridId,
        *,
        raise_on_errors: bool = False,
    ) -> list[ElectricalComponent]:
        """
        Get the electrical components of a microgrid.

        Args:
            microgrid_id: The ID of the microgrid to get the electrical components of.
            raise_on_errors: If True, raise an
                `ExceptionGroup[InvalidElectricalComponentError]`
                when major validation issues are found in any component instead
                of just logging them.

        Returns:
            The electrical components of the microgrid.

        Raises:
            ExceptionGroup: If `raise_on_errors` is True and major validation
                issues are found. All exceptions in the group are
                [InvalidElectricalComponentError][frequenz.client.assets.exceptions.InvalidElectricalComponentError].
        """
        response = await call_stub_method(
            self,
            lambda: self.stub.ListMicrogridElectricalComponents(
                assets_pb2.ListMicrogridElectricalComponentsRequest(
                    microgrid_id=int(microgrid_id),
                ),
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="ListMicrogridElectricalComponents",
        )

        if raise_on_errors:
            components: list[ElectricalComponent] = []
            exceptions: list[InvalidElectricalComponentError] = []
            for component_pb in response.components:
                major_issues: list[str] = []
                minor_issues: list[str] = []
                component = electrical_component_from_proto_with_issues(
                    component_pb,
                    major_issues=major_issues,
                    minor_issues=minor_issues,
                )
                if major_issues:
                    exceptions.append(
                        InvalidElectricalComponentError(
                            component=component,
                            major_issues=major_issues,
                            minor_issues=minor_issues,
                            raw_message=component_pb,
                        )
                    )
                else:
                    components.append(component)
            if exceptions:
                raise ExceptionGroup(
                    f"{len(exceptions)} electrical component(s) failed validation",
                    exceptions,
                )
            return components

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

    async def list_microgrid_electrical_component_connections(
        self,
        microgrid_id: MicrogridId,
        source_component_ids: Iterable[ElectricalComponentId] = (),
        destination_component_ids: Iterable[ElectricalComponentId] = (),
        *,
        raise_on_errors: bool = False,
    ) -> list[ComponentConnection]:
        """
        Get the electrical component connections of a microgrid.

        Args:
            microgrid_id: The ID of the microgrid to get the electrical
                component connections of.
            source_component_ids: Only return connections that originate from
                these component IDs. If None or empty, no filtering is applied.
            destination_component_ids: Only return connections that terminate at
                these component IDs. If None or empty, no filtering is applied.
            raise_on_errors: If True, raise an
                `ExceptionGroup[InvalidConnectionError]`
                when major validation issues are found in any connection instead
                of just logging them.

        Returns:
            The electrical component connections of the microgrid.

        Raises:
            ExceptionGroup: If `raise_on_errors` is True and major validation
                issues are found. All exceptions in the group are
                [InvalidConnectionError][frequenz.client.assets.exceptions.InvalidConnectionError].
        """
        request = assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
            microgrid_id=int(microgrid_id),
            source_component_ids=(int(c) for c in source_component_ids),
            destination_component_ids=(int(c) for c in destination_component_ids),
        )

        response = await call_stub_method(
            self,
            lambda: self.stub.ListMicrogridElectricalComponentConnections(
                request,
                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
            ),
            method_name="ListMicrogridElectricalComponentConnections",
        )

        if raise_on_errors:
            valid_connections: list[ComponentConnection] = []
            exceptions: list[InvalidConnectionError] = []
            for conn_pb in filter(bool, response.connections):
                major_issues: list[str] = []
                connection = component_connection_from_proto_with_issues(
                    conn_pb, major_issues=major_issues
                )
                if major_issues:
                    exceptions.append(
                        InvalidConnectionError(
                            connection=connection,
                            major_issues=major_issues,
                            minor_issues=[],
                            raw_message=conn_pb,
                        )
                    )
                elif connection is not None:
                    valid_connections.append(connection)
            if exceptions:
                raise ExceptionGroup(
                    f"{len(exceptions)} connection(s) failed validation",
                    exceptions,
                )
            return valid_connections

        return [
            c
            for c in map(component_connection_from_proto, response.connections)
            if c is not None
        ]

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 src/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: MicrogridId,
    *,
    raise_on_errors: bool = False
) -> Microgrid

Get the details of a microgrid.

PARAMETER	DESCRIPTION
microgrid_id	The ID of the microgrid to get the details of. TYPE: MicrogridId
raise_on_errors	If True, raise an InvalidMicrogridError when major validation issues are found instead of just logging them. TYPE: bool DEFAULT: False

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.
InvalidMicrogridError	If raise_on_errors is True and major validation issues are found.

Source code in src/frequenz/client/assets/_client.py

async def get_microgrid(  # noqa: DOC502,DOC503 (raises indirectly)
    self,
    microgrid_id: MicrogridId,
    *,
    raise_on_errors: bool = False,
) -> Microgrid:
    """
    Get the details of a microgrid.

    Args:
        microgrid_id: The ID of the microgrid to get the details of.
        raise_on_errors: If True, raise an
            [InvalidMicrogridError][frequenz.client.assets.exceptions.InvalidMicrogridError]
            when major validation issues are found instead of just
            logging them.

    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].
        InvalidMicrogridError: If `raise_on_errors` is True and major
            validation issues are found.
    """
    response = await call_stub_method(
        self,
        lambda: self.stub.GetMicrogrid(
            assets_pb2.GetMicrogridRequest(microgrid_id=int(microgrid_id)),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="GetMicrogrid",
    )

    if raise_on_errors:
        major_issues: list[str] = []
        minor_issues: list[str] = []
        microgrid = microgrid_from_proto_with_issues(
            response.microgrid,
            major_issues=major_issues,
            minor_issues=minor_issues,
        )
        if major_issues:
            raise InvalidMicrogridError(
                microgrid=microgrid,
                major_issues=major_issues,
                minor_issues=minor_issues,
                raw_message=response.microgrid,
            )
        return microgrid

    return microgrid_from_proto(response.microgrid)

list_microgrid_electrical_component_connections async

list_microgrid_electrical_component_connections(
    microgrid_id: MicrogridId,
    source_component_ids: Iterable[
        ElectricalComponentId
    ] = (),
    destination_component_ids: Iterable[
        ElectricalComponentId
    ] = (),
    *,
    raise_on_errors: bool = False
) -> list[ComponentConnection]

Get the electrical component connections of a microgrid.

PARAMETER	DESCRIPTION
microgrid_id	The ID of the microgrid to get the electrical component connections of. TYPE: MicrogridId
source_component_ids	Only return connections that originate from these component IDs. If None or empty, no filtering is applied. TYPE: Iterable[ElectricalComponentId] DEFAULT: ()
destination_component_ids	Only return connections that terminate at these component IDs. If None or empty, no filtering is applied. TYPE: Iterable[ElectricalComponentId] DEFAULT: ()
raise_on_errors	If True, raise an ExceptionGroup[InvalidConnectionError] when major validation issues are found in any connection instead of just logging them. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[ComponentConnection]	The electrical component connections of the microgrid.

RAISES	DESCRIPTION
ExceptionGroup	If raise_on_errors is True and major validation issues are found. All exceptions in the group are InvalidConnectionError.

Source code in src/frequenz/client/assets/_client.py

async def list_microgrid_electrical_component_connections(
    self,
    microgrid_id: MicrogridId,
    source_component_ids: Iterable[ElectricalComponentId] = (),
    destination_component_ids: Iterable[ElectricalComponentId] = (),
    *,
    raise_on_errors: bool = False,
) -> list[ComponentConnection]:
    """
    Get the electrical component connections of a microgrid.

    Args:
        microgrid_id: The ID of the microgrid to get the electrical
            component connections of.
        source_component_ids: Only return connections that originate from
            these component IDs. If None or empty, no filtering is applied.
        destination_component_ids: Only return connections that terminate at
            these component IDs. If None or empty, no filtering is applied.
        raise_on_errors: If True, raise an
            `ExceptionGroup[InvalidConnectionError]`
            when major validation issues are found in any connection instead
            of just logging them.

    Returns:
        The electrical component connections of the microgrid.

    Raises:
        ExceptionGroup: If `raise_on_errors` is True and major validation
            issues are found. All exceptions in the group are
            [InvalidConnectionError][frequenz.client.assets.exceptions.InvalidConnectionError].
    """
    request = assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
        microgrid_id=int(microgrid_id),
        source_component_ids=(int(c) for c in source_component_ids),
        destination_component_ids=(int(c) for c in destination_component_ids),
    )

    response = await call_stub_method(
        self,
        lambda: self.stub.ListMicrogridElectricalComponentConnections(
            request,
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="ListMicrogridElectricalComponentConnections",
    )

    if raise_on_errors:
        valid_connections: list[ComponentConnection] = []
        exceptions: list[InvalidConnectionError] = []
        for conn_pb in filter(bool, response.connections):
            major_issues: list[str] = []
            connection = component_connection_from_proto_with_issues(
                conn_pb, major_issues=major_issues
            )
            if major_issues:
                exceptions.append(
                    InvalidConnectionError(
                        connection=connection,
                        major_issues=major_issues,
                        minor_issues=[],
                        raw_message=conn_pb,
                    )
                )
            elif connection is not None:
                valid_connections.append(connection)
        if exceptions:
            raise ExceptionGroup(
                f"{len(exceptions)} connection(s) failed validation",
                exceptions,
            )
        return valid_connections

    return [
        c
        for c in map(component_connection_from_proto, response.connections)
        if c is not None
    ]

list_microgrid_electrical_components async

list_microgrid_electrical_components(
    microgrid_id: MicrogridId,
    *,
    raise_on_errors: bool = False
) -> 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: MicrogridId
raise_on_errors	If True, raise an ExceptionGroup[InvalidElectricalComponentError] when major validation issues are found in any component instead of just logging them. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[ElectricalComponent]	The electrical components of the microgrid.

RAISES	DESCRIPTION
ExceptionGroup	If raise_on_errors is True and major validation issues are found. All exceptions in the group are InvalidElectricalComponentError.

Source code in src/frequenz/client/assets/_client.py

async def list_microgrid_electrical_components(
    self,
    microgrid_id: MicrogridId,
    *,
    raise_on_errors: bool = False,
) -> list[ElectricalComponent]:
    """
    Get the electrical components of a microgrid.

    Args:
        microgrid_id: The ID of the microgrid to get the electrical components of.
        raise_on_errors: If True, raise an
            `ExceptionGroup[InvalidElectricalComponentError]`
            when major validation issues are found in any component instead
            of just logging them.

    Returns:
        The electrical components of the microgrid.

    Raises:
        ExceptionGroup: If `raise_on_errors` is True and major validation
            issues are found. All exceptions in the group are
            [InvalidElectricalComponentError][frequenz.client.assets.exceptions.InvalidElectricalComponentError].
    """
    response = await call_stub_method(
        self,
        lambda: self.stub.ListMicrogridElectricalComponents(
            assets_pb2.ListMicrogridElectricalComponentsRequest(
                microgrid_id=int(microgrid_id),
            ),
            timeout=DEFAULT_GRPC_CALL_TIMEOUT,
        ),
        method_name="ListMicrogridElectricalComponents",
    )

    if raise_on_errors:
        components: list[ElectricalComponent] = []
        exceptions: list[InvalidElectricalComponentError] = []
        for component_pb in response.components:
            major_issues: list[str] = []
            minor_issues: list[str] = []
            component = electrical_component_from_proto_with_issues(
                component_pb,
                major_issues=major_issues,
                minor_issues=minor_issues,
            )
            if major_issues:
                exceptions.append(
                    InvalidElectricalComponentError(
                        component=component,
                        major_issues=major_issues,
                        minor_issues=minor_issues,
                        raw_message=component_pb,
                    )
                )
            else:
                components.append(component)
        if exceptions:
            raise ExceptionGroup(
                f"{len(exceptions)} electrical component(s) failed validation",
                exceptions,
            )
        return components

    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 src/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

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.

Functions

__str__

__str__() -> str

Return a human-readable string representation of this instance.

Source code in src/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 src/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.Lifetime dataclass

An active operational period of a microgrid asset.

Warning

The end timestamp indicates that the asset has been permanently removed from the system.

Source code in src/frequenz/client/assets/_lifetime.py

@dataclass(frozen=True, kw_only=True)
class Lifetime:
    """An active operational period of a microgrid asset.

    Warning:
        The [`end`][frequenz.client.assets.Lifetime.end] timestamp indicates that the
        asset has been permanently removed from the system.
    """

    start: 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`][frequenz.client.assets.Lifetime.end].
    """

    end: 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.
    """

    def __post_init__(self) -> None:
        """Validate this lifetime."""
        if self.start is not None and self.end is not None and self.start > self.end:
            raise ValueError(
                f"Start ({self.start}) must be before or equal to end ({self.end})"
            )

    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 is in the future
        if self.start is not None and self.start > timestamp:
            return False
        # Handle end time - active up to and including end time
        if self.end is not None:
            return self.end >= timestamp
        # self.end is None, and either self.start is None or self.start <= 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 class-attribute instance-attribute

end: 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 class-attribute instance-attribute

start: 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.

Functions

__post_init__

__post_init__() -> None

Validate this lifetime.

Source code in src/frequenz/client/assets/_lifetime.py

def __post_init__(self) -> None:
    """Validate this lifetime."""
    if self.start is not None and self.end is not None and self.start > self.end:
        raise ValueError(
            f"Start ({self.start}) must be before or equal to end ({self.end})"
        )

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/assets/_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 is in the future
    if self.start is not None and self.start > timestamp:
        return False
    # Handle end time - active up to and including end time
    if self.end is not None:
        return self.end >= timestamp
    # self.end is None, and either self.start is None or self.start <= 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/assets/_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.assets.Location dataclass

A location of a microgrid.

Source code in src/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 src/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 src/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 src/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 src/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.