Index

frequenz.client.assets

Assets API client.

Classes

frequenz.client.assets.AssetsApiClient

Bases: BaseApiClient[PlatformAssetsServiceStub]

A client for the Assets API.

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

class AssetsApiClient(
    BaseApiClient[platformassets_pb2_grpc.PlatformAssetsServiceStub]
):  # 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,
            platformassets_pb2_grpc.PlatformAssetsServiceStub,
            connect=connect,
            channel_defaults=channel_defaults,
            auth_key=auth_key,
            sign_secret=sign_secret,
        )

    @property
    def stub(self) -> platformassets_pb2_grpc.PlatformAssetsServiceAsyncStub:
        """
        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 list_gridpools(  # noqa: DOC502 (raises indirectly)
        self,
        gridpool_ids: Iterable[int] = (),
    ) -> list[Gridpool]:
        """
        List gridpools within the current enterprise scope.

        Args:
            gridpool_ids: Only return gridpools whose IDs are included in this list.
                If empty, no filtering is applied.

        Returns:
            The matching gridpools.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        """
        request = platformassets_pb2.ListGridpoolsRequest()
        if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
            request.filter.gridpool_ids.extend(ids)

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

        return [gridpool_from_proto(gridpool) for gridpool in response.gridpools]

    async def list_gridpool_energy_schedules(  # noqa: DOC502 (raises indirectly)
        self,
        gridpool_id: int,
        schedule_ids: Iterable[int] = (),
        directions: Iterable[GridpoolEnergyScheduleDirection | int] = (),
        *,
        time_series_interval: Interval | None = None,
        effective_validity_period: Interval | None = None,
    ) -> list[GridpoolEnergySchedule]:
        """
        List energy schedules for a gridpool.

        Args:
            gridpool_id: The ID of the gridpool whose schedules should be listed.
            schedule_ids: Only return schedules whose IDs are included in this list.
                If empty, no schedule-ID filtering is applied.
            directions: Only return schedules with one of these directions. If empty,
                no direction filtering is applied.
            time_series_interval: Restrict returned time-series entries to delivery
                periods that overlap this interval.
            effective_validity_period: Only return schedules whose effective validity
                period overlaps this interval.

        Returns:
            The matching gridpool energy schedules.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        """
        request = platformassets_pb2.ListGridpoolEnergySchedulesRequest(
            gridpool_id=int(gridpool_id),
        )
        if ids := [int(schedule_id) for schedule_id in schedule_ids]:
            request.filter.schedule_ids.extend(ids)
        if direction_values := _proto_enum_values(
            directions,
            platformassets_pb2.GridpoolEnergyScheduleDirection.ValueType,
        ):
            request.filter.directions.extend(direction_values)
        if time_series_interval is not None:
            request.filter.time_series_interval.CopyFrom(
                interval_to_proto(time_series_interval)
            )
        if effective_validity_period is not None:
            request.filter.effective_validity_period.CopyFrom(
                interval_to_proto(effective_validity_period)
            )

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

        return [
            gridpool_energy_schedule_from_proto(schedule)
            for schedule in response.schedules
        ]

    async def list_market_topology_relations(  # noqa: DOC502 (raises indirectly)
        self,
        *,
        gridpool_ids: Iterable[int] = (),
        microgrid_ids: Iterable[MicrogridId] = (),
        market_location_id_values: Iterable[str] = (),
        delivery_areas: Iterable[DeliveryArea] = (),
        participation_types: Iterable[MarketParticipationType | int] = (),
    ) -> list[MarketTopologyRelation]:
        """
        List market-topology relations within the current enterprise scope.

        Args:
            gridpool_ids: Only return relations involving any of these gridpools.
            microgrid_ids: Only return relations involving any of these microgrids.
            market_location_id_values: Only return relations involving market
                locations whose ID values match any of these values.
            delivery_areas: Only return relations applying to any of these delivery
                areas.
            participation_types: Only return relations that include at least one
                participation with one of these types.

        Returns:
            The matching market-topology relations.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        """
        request = platformassets_pb2.ListMarketTopologyRelationsRequest()
        if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
            request.filter.gridpool_ids.extend(ids)
        if ids := [int(microgrid_id) for microgrid_id in microgrid_ids]:
            request.filter.microgrid_ids.extend(ids)
        if values := [
            market_location_id_value_to_proto(value)
            for value in market_location_id_values
        ]:
            request.filter.market_location_id_values.extend(values)
        if areas := [delivery_area_to_proto(area) for area in delivery_areas]:
            request.filter.delivery_areas.extend(areas)
        if types := _proto_enum_values(
            participation_types,
            platformassets_pb2.MarketParticipationType.ValueType,
        ):
            request.filter.participation_types.extend(types)

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

        return [
            market_topology_relation_from_proto(relation)
            for relation in response.relations
        ]

    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(
                platformassets_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_microgrids(  # noqa: DOC502,DOC503 (raises indirectly)
        self,
        microgrid_ids: Iterable[MicrogridId] = (),
        gridpool_ids: Iterable[int] = (),
        *,
        raise_on_errors: bool = False,
    ) -> list[Microgrid]:
        """
        List microgrids within the current enterprise scope.

        Args:
            microgrid_ids: Only return microgrids whose IDs are included in this list.
                If empty, no microgrid-ID filtering is applied.
            gridpool_ids: Only return microgrids that are part of a market-topology
                relation involving any of these gridpools.
            raise_on_errors: If True, raise an `ExceptionGroup[InvalidMicrogridError]`
                when major validation issues are found in any microgrid instead of
                just logging them.

        Returns:
            The matching microgrids.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
            ExceptionGroup: If `raise_on_errors` is True and major validation
                issues are found. All exceptions in the group are
                [InvalidMicrogridError][frequenz.client.assets.exceptions.InvalidMicrogridError].
        """
        request = platformassets_pb2.ListMicrogridsRequest()
        if ids := [int(microgrid_id) for microgrid_id in microgrid_ids]:
            request.filter.microgrid_ids.extend(ids)
        if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
            request.filter.gridpool_ids.extend(ids)

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

        if raise_on_errors:
            microgrids: list[Microgrid] = []
            exceptions: list[InvalidMicrogridError] = []
            for microgrid_pb in response.microgrids:
                major_issues: list[str] = []
                minor_issues: list[str] = []
                microgrid = microgrid_from_proto_with_issues(
                    microgrid_pb,
                    major_issues=major_issues,
                    minor_issues=minor_issues,
                )
                if major_issues:
                    exceptions.append(
                        InvalidMicrogridError(
                            microgrid=microgrid,
                            major_issues=major_issues,
                            minor_issues=minor_issues,
                            raw_message=microgrid_pb,
                        )
                    )
                else:
                    microgrids.append(microgrid)
            if exceptions:
                raise ExceptionGroup(
                    f"{len(exceptions)} microgrid(s) failed validation",
                    exceptions,
                )
            return microgrids

        return [microgrid_from_proto(microgrid) for microgrid in response.microgrids]

    async def list_microgrid_electrical_components(
        self,
        microgrid_id: MicrogridId,
        component_ids: Iterable[ElectricalComponentId] = (),
        categories: Iterable[ElectricalComponentCategory | int] = (),
        *,
        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.
            component_ids: Only return components whose IDs are included in this list.
                If empty, no component-ID filtering is applied.
            categories: Only return components whose categories are included in this
                list. If empty, no category filtering is applied.
            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].
        """
        request = platformassets_pb2.ListMicrogridElectricalComponentsRequest(
            microgrid_id=int(microgrid_id),
        )
        if ids := [int(component_id) for component_id in component_ids]:
            request.filter.component_ids.extend(ids)
        if category_values := _proto_enum_values(
            categories,
            electrical_components_pb2.ElectricalComponentCategory.ValueType,
        ):
            request.filter.categories.extend(category_values)

        response = await call_stub_method(
            self,
            lambda: self.stub.ListMicrogridElectricalComponents(
                request,
                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].
        """
        source_ids = [int(c) for c in source_component_ids]
        destination_ids = [int(c) for c in destination_component_ids]
        request = platformassets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
            microgrid_id=int(microgrid_id),
        )
        if source_ids or destination_ids:
            request.filter.source_component_ids.extend(source_ids)
            request.filter.destination_component_ids.extend(destination_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
        ]

    async def list_microgrid_sensors(  # noqa: DOC502 (raises indirectly)
        self,
        microgrid_id: MicrogridId,
        sensor_ids: Iterable[SensorId] = (),
    ) -> list[Sensor]:
        """
        List sensors in a microgrid.

        Args:
            microgrid_id: The ID of the microgrid whose sensors should be listed.
            sensor_ids: Only return sensors whose IDs are included in this list. If
                empty, no filtering is applied.

        Returns:
            The matching sensors.

        Raises:
            ApiClientError: If there are any errors communicating with the Assets API,
                most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        """
        request = platformassets_pb2.ListMicrogridSensorsRequest(
            microgrid_id=int(microgrid_id),
        )
        if ids := [int(sensor_id) for sensor_id in sensor_ids]:
            request.filter.sensor_ids.extend(ids)

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

        return [sensor_from_proto(sensor) for sensor in response.sensors]

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: PlatformAssetsServiceAsyncStub

The gRPC stub for the Assets API.

RETURNS	DESCRIPTION
PlatformAssetsServiceAsyncStub	The gRPC stub for the Assets API.

RAISES	DESCRIPTION
ClientNotConnected	If the client is not connected to the server.

Methods:

__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,
        platformassets_pb2_grpc.PlatformAssetsServiceStub,
        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(
            platformassets_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_gridpool_energy_schedules async

list_gridpool_energy_schedules(
    gridpool_id: int,
    schedule_ids: Iterable[int] = (),
    directions: Iterable[
        GridpoolEnergyScheduleDirection | int
    ] = (),
    *,
    time_series_interval: Interval | None = None,
    effective_validity_period: Interval | None = None
) -> list[GridpoolEnergySchedule]

List energy schedules for a gridpool.

PARAMETER	DESCRIPTION
gridpool_id	The ID of the gridpool whose schedules should be listed. TYPE: int
schedule_ids	Only return schedules whose IDs are included in this list. If empty, no schedule-ID filtering is applied. TYPE: Iterable[int] DEFAULT: ()
directions	Only return schedules with one of these directions. If empty, no direction filtering is applied. TYPE: Iterable[GridpoolEnergyScheduleDirection | int] DEFAULT: ()
time_series_interval	Restrict returned time-series entries to delivery periods that overlap this interval. TYPE: Interval | None DEFAULT: None
effective_validity_period	Only return schedules whose effective validity period overlaps this interval. TYPE: Interval | None DEFAULT: None

RETURNS	DESCRIPTION
list[GridpoolEnergySchedule]	The matching gridpool energy schedules.

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

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

async def list_gridpool_energy_schedules(  # noqa: DOC502 (raises indirectly)
    self,
    gridpool_id: int,
    schedule_ids: Iterable[int] = (),
    directions: Iterable[GridpoolEnergyScheduleDirection | int] = (),
    *,
    time_series_interval: Interval | None = None,
    effective_validity_period: Interval | None = None,
) -> list[GridpoolEnergySchedule]:
    """
    List energy schedules for a gridpool.

    Args:
        gridpool_id: The ID of the gridpool whose schedules should be listed.
        schedule_ids: Only return schedules whose IDs are included in this list.
            If empty, no schedule-ID filtering is applied.
        directions: Only return schedules with one of these directions. If empty,
            no direction filtering is applied.
        time_series_interval: Restrict returned time-series entries to delivery
            periods that overlap this interval.
        effective_validity_period: Only return schedules whose effective validity
            period overlaps this interval.

    Returns:
        The matching gridpool energy schedules.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
    """
    request = platformassets_pb2.ListGridpoolEnergySchedulesRequest(
        gridpool_id=int(gridpool_id),
    )
    if ids := [int(schedule_id) for schedule_id in schedule_ids]:
        request.filter.schedule_ids.extend(ids)
    if direction_values := _proto_enum_values(
        directions,
        platformassets_pb2.GridpoolEnergyScheduleDirection.ValueType,
    ):
        request.filter.directions.extend(direction_values)
    if time_series_interval is not None:
        request.filter.time_series_interval.CopyFrom(
            interval_to_proto(time_series_interval)
        )
    if effective_validity_period is not None:
        request.filter.effective_validity_period.CopyFrom(
            interval_to_proto(effective_validity_period)
        )

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

    return [
        gridpool_energy_schedule_from_proto(schedule)
        for schedule in response.schedules
    ]

list_gridpools async

list_gridpools(
    gridpool_ids: Iterable[int] = (),
) -> list[Gridpool]

List gridpools within the current enterprise scope.

PARAMETER	DESCRIPTION
gridpool_ids	Only return gridpools whose IDs are included in this list. If empty, no filtering is applied. TYPE: Iterable[int] DEFAULT: ()

RETURNS	DESCRIPTION
list[Gridpool]	The matching gridpools.

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

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

async def list_gridpools(  # noqa: DOC502 (raises indirectly)
    self,
    gridpool_ids: Iterable[int] = (),
) -> list[Gridpool]:
    """
    List gridpools within the current enterprise scope.

    Args:
        gridpool_ids: Only return gridpools whose IDs are included in this list.
            If empty, no filtering is applied.

    Returns:
        The matching gridpools.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
    """
    request = platformassets_pb2.ListGridpoolsRequest()
    if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
        request.filter.gridpool_ids.extend(ids)

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

    return [gridpool_from_proto(gridpool) for gridpool in response.gridpools]

list_market_topology_relations async

list_market_topology_relations(
    *,
    gridpool_ids: Iterable[int] = (),
    microgrid_ids: Iterable[MicrogridId] = (),
    market_location_id_values: Iterable[str] = (),
    delivery_areas: Iterable[DeliveryArea] = (),
    participation_types: Iterable[
        MarketParticipationType | int
    ] = ()
) -> list[MarketTopologyRelation]

List market-topology relations within the current enterprise scope.

PARAMETER	DESCRIPTION
gridpool_ids	Only return relations involving any of these gridpools. TYPE: Iterable[int] DEFAULT: ()
microgrid_ids	Only return relations involving any of these microgrids. TYPE: Iterable[MicrogridId] DEFAULT: ()
market_location_id_values	Only return relations involving market locations whose ID values match any of these values. TYPE: Iterable[str] DEFAULT: ()
delivery_areas	Only return relations applying to any of these delivery areas. TYPE: Iterable[DeliveryArea] DEFAULT: ()
participation_types	Only return relations that include at least one participation with one of these types. TYPE: Iterable[MarketParticipationType | int] DEFAULT: ()

RETURNS	DESCRIPTION
list[MarketTopologyRelation]	The matching market-topology relations.

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

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

async def list_market_topology_relations(  # noqa: DOC502 (raises indirectly)
    self,
    *,
    gridpool_ids: Iterable[int] = (),
    microgrid_ids: Iterable[MicrogridId] = (),
    market_location_id_values: Iterable[str] = (),
    delivery_areas: Iterable[DeliveryArea] = (),
    participation_types: Iterable[MarketParticipationType | int] = (),
) -> list[MarketTopologyRelation]:
    """
    List market-topology relations within the current enterprise scope.

    Args:
        gridpool_ids: Only return relations involving any of these gridpools.
        microgrid_ids: Only return relations involving any of these microgrids.
        market_location_id_values: Only return relations involving market
            locations whose ID values match any of these values.
        delivery_areas: Only return relations applying to any of these delivery
            areas.
        participation_types: Only return relations that include at least one
            participation with one of these types.

    Returns:
        The matching market-topology relations.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
    """
    request = platformassets_pb2.ListMarketTopologyRelationsRequest()
    if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
        request.filter.gridpool_ids.extend(ids)
    if ids := [int(microgrid_id) for microgrid_id in microgrid_ids]:
        request.filter.microgrid_ids.extend(ids)
    if values := [
        market_location_id_value_to_proto(value)
        for value in market_location_id_values
    ]:
        request.filter.market_location_id_values.extend(values)
    if areas := [delivery_area_to_proto(area) for area in delivery_areas]:
        request.filter.delivery_areas.extend(areas)
    if types := _proto_enum_values(
        participation_types,
        platformassets_pb2.MarketParticipationType.ValueType,
    ):
        request.filter.participation_types.extend(types)

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

    return [
        market_topology_relation_from_proto(relation)
        for relation in response.relations
    ]

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].
    """
    source_ids = [int(c) for c in source_component_ids]
    destination_ids = [int(c) for c in destination_component_ids]
    request = platformassets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
        microgrid_id=int(microgrid_id),
    )
    if source_ids or destination_ids:
        request.filter.source_component_ids.extend(source_ids)
        request.filter.destination_component_ids.extend(destination_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,
    component_ids: Iterable[ElectricalComponentId] = (),
    categories: Iterable[
        ElectricalComponentCategory | int
    ] = (),
    *,
    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
component_ids	Only return components whose IDs are included in this list. If empty, no component-ID filtering is applied. TYPE: Iterable[ElectricalComponentId] DEFAULT: ()
categories	Only return components whose categories are included in this list. If empty, no category filtering is applied. TYPE: Iterable[ElectricalComponentCategory | int] DEFAULT: ()
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,
    component_ids: Iterable[ElectricalComponentId] = (),
    categories: Iterable[ElectricalComponentCategory | int] = (),
    *,
    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.
        component_ids: Only return components whose IDs are included in this list.
            If empty, no component-ID filtering is applied.
        categories: Only return components whose categories are included in this
            list. If empty, no category filtering is applied.
        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].
    """
    request = platformassets_pb2.ListMicrogridElectricalComponentsRequest(
        microgrid_id=int(microgrid_id),
    )
    if ids := [int(component_id) for component_id in component_ids]:
        request.filter.component_ids.extend(ids)
    if category_values := _proto_enum_values(
        categories,
        electrical_components_pb2.ElectricalComponentCategory.ValueType,
    ):
        request.filter.categories.extend(category_values)

    response = await call_stub_method(
        self,
        lambda: self.stub.ListMicrogridElectricalComponents(
            request,
            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
    ]

list_microgrid_sensors async

list_microgrid_sensors(
    microgrid_id: MicrogridId,
    sensor_ids: Iterable[SensorId] = (),
) -> list[Sensor]

List sensors in a microgrid.

PARAMETER	DESCRIPTION
microgrid_id	The ID of the microgrid whose sensors should be listed. TYPE: MicrogridId
sensor_ids	Only return sensors whose IDs are included in this list. If empty, no filtering is applied. TYPE: Iterable[SensorId] DEFAULT: ()

RETURNS	DESCRIPTION
list[Sensor]	The matching sensors.

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

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

async def list_microgrid_sensors(  # noqa: DOC502 (raises indirectly)
    self,
    microgrid_id: MicrogridId,
    sensor_ids: Iterable[SensorId] = (),
) -> list[Sensor]:
    """
    List sensors in a microgrid.

    Args:
        microgrid_id: The ID of the microgrid whose sensors should be listed.
        sensor_ids: Only return sensors whose IDs are included in this list. If
            empty, no filtering is applied.

    Returns:
        The matching sensors.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
    """
    request = platformassets_pb2.ListMicrogridSensorsRequest(
        microgrid_id=int(microgrid_id),
    )
    if ids := [int(sensor_id) for sensor_id in sensor_ids]:
        request.filter.sensor_ids.extend(ids)

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

    return [sensor_from_proto(sensor) for sensor in response.sensors]

list_microgrids async

list_microgrids(
    microgrid_ids: Iterable[MicrogridId] = (),
    gridpool_ids: Iterable[int] = (),
    *,
    raise_on_errors: bool = False
) -> list[Microgrid]

List microgrids within the current enterprise scope.

PARAMETER	DESCRIPTION
microgrid_ids	Only return microgrids whose IDs are included in this list. If empty, no microgrid-ID filtering is applied. TYPE: Iterable[MicrogridId] DEFAULT: ()
gridpool_ids	Only return microgrids that are part of a market-topology relation involving any of these gridpools. TYPE: Iterable[int] DEFAULT: ()
raise_on_errors	If True, raise an ExceptionGroup[InvalidMicrogridError] when major validation issues are found in any microgrid instead of just logging them. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[Microgrid]	The matching microgrids.

RAISES	DESCRIPTION
ApiClientError	If there are any errors communicating with the Assets API, most likely a subclass of GrpcError.
ExceptionGroup	If raise_on_errors is True and major validation issues are found. All exceptions in the group are InvalidMicrogridError.

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

async def list_microgrids(  # noqa: DOC502,DOC503 (raises indirectly)
    self,
    microgrid_ids: Iterable[MicrogridId] = (),
    gridpool_ids: Iterable[int] = (),
    *,
    raise_on_errors: bool = False,
) -> list[Microgrid]:
    """
    List microgrids within the current enterprise scope.

    Args:
        microgrid_ids: Only return microgrids whose IDs are included in this list.
            If empty, no microgrid-ID filtering is applied.
        gridpool_ids: Only return microgrids that are part of a market-topology
            relation involving any of these gridpools.
        raise_on_errors: If True, raise an `ExceptionGroup[InvalidMicrogridError]`
            when major validation issues are found in any microgrid instead of
            just logging them.

    Returns:
        The matching microgrids.

    Raises:
        ApiClientError: If there are any errors communicating with the Assets API,
            most likely a subclass of [GrpcError][frequenz.client.base.exception.GrpcError].
        ExceptionGroup: If `raise_on_errors` is True and major validation
            issues are found. All exceptions in the group are
            [InvalidMicrogridError][frequenz.client.assets.exceptions.InvalidMicrogridError].
    """
    request = platformassets_pb2.ListMicrogridsRequest()
    if ids := [int(microgrid_id) for microgrid_id in microgrid_ids]:
        request.filter.microgrid_ids.extend(ids)
    if ids := [int(gridpool_id) for gridpool_id in gridpool_ids]:
        request.filter.gridpool_ids.extend(ids)

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

    if raise_on_errors:
        microgrids: list[Microgrid] = []
        exceptions: list[InvalidMicrogridError] = []
        for microgrid_pb in response.microgrids:
            major_issues: list[str] = []
            minor_issues: list[str] = []
            microgrid = microgrid_from_proto_with_issues(
                microgrid_pb,
                major_issues=major_issues,
                minor_issues=minor_issues,
            )
            if major_issues:
                exceptions.append(
                    InvalidMicrogridError(
                        microgrid=microgrid,
                        major_issues=major_issues,
                        minor_issues=minor_issues,
                        raw_message=microgrid_pb,
                    )
                )
            else:
                microgrids.append(microgrid)
        if exceptions:
            raise ExceptionGroup(
                f"{len(exceptions)} microgrid(s) failed validation",
                exceptions,
            )
        return microgrids

    return [microgrid_from_proto(microgrid) for microgrid in response.microgrids]

frequenz.client.assets.BalancingGroup dataclass

A market balancing group identified by its market code.

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

@dataclass(frozen=True, kw_only=True)
class BalancingGroup:
    """A market balancing group identified by its market code."""

    code: str | None
    """The balancing group code."""

    code_type: EnergyMarketCodeType | int
    """The type of market code used to identify the balancing group."""

Attributes

code instance-attribute

code: str | None

The balancing group code.

code_type instance-attribute

code_type: EnergyMarketCodeType | int

The type of market code used to identify the balancing group.

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.

Methods:

__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.DeliveryDuration

Bases: Enum

Delivery duration used by scheduled energy values.

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

@enum.unique
class DeliveryDuration(enum.Enum):
    """Delivery duration used by scheduled energy values."""

    UNSPECIFIED = delivery_duration_pb2.DELIVERY_DURATION_UNSPECIFIED
    """The delivery duration is unspecified."""

    MINUTES_5 = delivery_duration_pb2.DELIVERY_DURATION_5
    """A 5-minute delivery duration."""

    MINUTES_15 = delivery_duration_pb2.DELIVERY_DURATION_15
    """A 15-minute delivery duration."""

    MINUTES_30 = delivery_duration_pb2.DELIVERY_DURATION_30
    """A 30-minute delivery duration."""

    MINUTES_60 = delivery_duration_pb2.DELIVERY_DURATION_60
    """A 60-minute delivery duration."""

Attributes

MINUTES_15 class-attribute instance-attribute

MINUTES_15 = delivery_duration_pb2.DELIVERY_DURATION_15

A 15-minute delivery duration.

MINUTES_30 class-attribute instance-attribute

MINUTES_30 = delivery_duration_pb2.DELIVERY_DURATION_30

A 30-minute delivery duration.

MINUTES_5 class-attribute instance-attribute

MINUTES_5 = delivery_duration_pb2.DELIVERY_DURATION_5

A 5-minute delivery duration.

MINUTES_60 class-attribute instance-attribute

MINUTES_60 = delivery_duration_pb2.DELIVERY_DURATION_60

A 60-minute delivery duration.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    delivery_duration_pb2.DELIVERY_DURATION_UNSPECIFIED
)

The delivery duration is unspecified.

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 = (
    delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC
)

European Energy Identification Code Standard.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    delivery_area_pb2.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 = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_US_NERC

North American Electric Reliability Corporation identifiers.

frequenz.client.assets.Gridpool dataclass

A virtual balancing-group structure used for market interactions.

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

@dataclass(frozen=True, kw_only=True)
class Gridpool:
    """A virtual balancing-group structure used for market interactions."""

    id: int
    """The unique identifier of the gridpool."""

    name: str | None
    """The human-readable gridpool name."""

Attributes

id instance-attribute

id: int

The unique identifier of the gridpool.

name instance-attribute

name: str | None

The human-readable gridpool name.

frequenz.client.assets.GridpoolEnergySchedule dataclass

A static energy schedule associated with a gridpool.

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

@dataclass(frozen=True, kw_only=True)
class GridpoolEnergySchedule:  # pylint: disable=too-many-instance-attributes
    """A static energy schedule associated with a gridpool."""

    gridpool_id: int
    """The unique identifier of the gridpool this schedule belongs to."""

    schedule_id: int
    """The unique identifier of the energy schedule."""

    name: str | None
    """The human-readable schedule name."""

    counterparty_balancing_group: BalancingGroup | None
    """The third-party balancing group involved in the scheduled exchange."""

    counterparty_delivery_area: DeliveryArea | None
    """Delivery area of the counterparty side of the scheduled exchange."""

    frequenz_delivery_area: DeliveryArea | None
    """Delivery area of the Frequenz side of the scheduled exchange."""

    direction: GridpoolEnergyScheduleDirection | int
    """Direction of the scheduled exchange from the Frequenz perspective."""

    validity_period: Interval | None
    """Validity interval of the schedule configuration."""

    cancel_time: datetime | None
    """Timestamp at which the schedule was cancelled."""

    delivery_duration: DeliveryDuration | int
    """Delivery duration used by all time-series entries in this schedule."""

    time_series: list[GridpoolEnergyScheduleTimeSeriesEntry]
    """Scheduled active-power values."""

Attributes

cancel_time instance-attribute

cancel_time: datetime | None

Timestamp at which the schedule was cancelled.

counterparty_balancing_group instance-attribute

counterparty_balancing_group: BalancingGroup | None

The third-party balancing group involved in the scheduled exchange.

counterparty_delivery_area instance-attribute

counterparty_delivery_area: DeliveryArea | None

Delivery area of the counterparty side of the scheduled exchange.

delivery_duration instance-attribute

delivery_duration: DeliveryDuration | int

Delivery duration used by all time-series entries in this schedule.

direction instance-attribute

direction: GridpoolEnergyScheduleDirection | int

Direction of the scheduled exchange from the Frequenz perspective.

frequenz_delivery_area instance-attribute

frequenz_delivery_area: DeliveryArea | None

Delivery area of the Frequenz side of the scheduled exchange.

gridpool_id instance-attribute

gridpool_id: int

The unique identifier of the gridpool this schedule belongs to.

name instance-attribute

name: str | None

The human-readable schedule name.

schedule_id instance-attribute

schedule_id: int

The unique identifier of the energy schedule.

time_series instance-attribute

time_series: list[GridpoolEnergyScheduleTimeSeriesEntry]

Scheduled active-power values.

validity_period instance-attribute

validity_period: Interval | None

Validity interval of the schedule configuration.

frequenz.client.assets.GridpoolEnergyScheduleDirection

Bases: Enum

Direction of a scheduled energy exchange.

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

@enum.unique
class GridpoolEnergyScheduleDirection(enum.Enum):
    """Direction of a scheduled energy exchange."""

    UNSPECIFIED = platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_UNSPECIFIED
    """The direction is unspecified."""

    IMPORT = platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_IMPORT
    """Energy is imported into the Frequenz balancing group."""

    EXPORT = platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_EXPORT
    """Energy is exported from the Frequenz balancing group."""

Attributes

EXPORT class-attribute instance-attribute

EXPORT = (
    platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_EXPORT
)

Energy is exported from the Frequenz balancing group.

IMPORT class-attribute instance-attribute

IMPORT = (
    platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_IMPORT
)

Energy is imported into the Frequenz balancing group.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    platformassets_pb2.GRIDPOOL_ENERGY_SCHEDULE_DIRECTION_UNSPECIFIED
)

The direction is unspecified.

frequenz.client.assets.GridpoolEnergyScheduleTimeSeriesEntry dataclass

A scheduled active-power value for one delivery period.

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

@dataclass(frozen=True, kw_only=True)
class GridpoolEnergyScheduleTimeSeriesEntry:
    """A scheduled active-power value for one delivery period."""

    start_time: datetime | None
    """The inclusive start timestamp of the scheduled delivery value."""

    active_power_w: float
    """Scheduled active power in watts."""

Attributes

active_power_w instance-attribute

active_power_w: float

Scheduled active power in watts.

start_time instance-attribute

start_time: datetime | None

The inclusive start timestamp of the scheduled delivery value.

frequenz.client.assets.Interval dataclass

A half-open time interval: [start, end).

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

@dataclass(frozen=True, kw_only=True)
class Interval:
    """A half-open time interval: `[start, end)`."""

    start: datetime | None = None
    """The inclusive start of the interval."""

    end: datetime | None = None
    """The exclusive end of the interval."""

    def __post_init__(self) -> None:
        """Validate this interval."""
        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})"
            )

Attributes

end class-attribute instance-attribute

end: datetime | None = None

The exclusive end of the interval.

start class-attribute instance-attribute

start: datetime | None = None

The inclusive start of the interval.

Methods:

__post_init__

__post_init__() -> None

Validate this interval.

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

def __post_init__(self) -> None:
    """Validate this interval."""
    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})"
        )

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.

Methods:

__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.

Methods:

__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.MarketLocationId dataclass

A market-standard identifier for a market location.

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

@dataclass(frozen=True, kw_only=True)
class MarketLocationId:
    """A market-standard identifier for a market location."""

    value: str | None
    """The official market location identifier value."""

    type: MarketLocationIdType | int
    """The type of official market identifier."""

Attributes

type instance-attribute

type: MarketLocationIdType | int

The type of official market identifier.

value instance-attribute

value: str | None

The official market location identifier value.

frequenz.client.assets.MarketLocationIdType

Bases: Enum

External market identifier types used for market locations.

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

@enum.unique
class MarketLocationIdType(enum.Enum):
    """External market identifier types used for market locations."""

    UNSPECIFIED = market_location_pb2.MARKET_LOCATION_ID_TYPE_UNSPECIFIED
    """The market location ID type is unspecified."""

    MALO_ID = market_location_pb2.MARKET_LOCATION_ID_TYPE_MALO_ID
    """Germany Marktlokations-ID."""

    ZAEHLPUNKT = market_location_pb2.MARKET_LOCATION_ID_TYPE_ZAEHLPUNKT
    """Austria Zaehlpunktbezeichnung."""

    MPAN = market_location_pb2.MARKET_LOCATION_ID_TYPE_MPAN
    """United Kingdom Meter Point Administration Number."""

    POD = market_location_pb2.MARKET_LOCATION_ID_TYPE_POD
    """Italy Point of Delivery."""

    CUPS = market_location_pb2.MARKET_LOCATION_ID_TYPE_CUPS
    """Spain Codigo Universal de Punto de Suministro."""

    PRM = market_location_pb2.MARKET_LOCATION_ID_TYPE_PRM
    """France Point de Reference et Mesure."""

    EAN = market_location_pb2.MARKET_LOCATION_ID_TYPE_EAN
    """European Article Number."""

    GSRN = market_location_pb2.MARKET_LOCATION_ID_TYPE_GSRN
    """GS1 Global Service Relation Number."""

    ESI_ID = market_location_pb2.MARKET_LOCATION_ID_TYPE_ESI_ID
    """United States Electric Service Identifier."""

    NMI = market_location_pb2.MARKET_LOCATION_ID_TYPE_NMI
    """Australia National Metering Identifier."""

    ICP = market_location_pb2.MARKET_LOCATION_ID_TYPE_ICP
    """New Zealand Installation Control Point."""

    SPN = market_location_pb2.MARKET_LOCATION_ID_TYPE_SPN
    """Japan Supply Point Number."""

    OTHER = market_location_pb2.MARKET_LOCATION_ID_TYPE_OTHER
    """Generic identifier for markets not modeled explicitly."""

Attributes

CUPS class-attribute instance-attribute

CUPS = market_location_pb2.MARKET_LOCATION_ID_TYPE_CUPS

Spain Codigo Universal de Punto de Suministro.

EAN class-attribute instance-attribute

EAN = market_location_pb2.MARKET_LOCATION_ID_TYPE_EAN

European Article Number.

ESI_ID class-attribute instance-attribute

ESI_ID = market_location_pb2.MARKET_LOCATION_ID_TYPE_ESI_ID

United States Electric Service Identifier.

GSRN class-attribute instance-attribute

GSRN = market_location_pb2.MARKET_LOCATION_ID_TYPE_GSRN

GS1 Global Service Relation Number.

ICP class-attribute instance-attribute

ICP = market_location_pb2.MARKET_LOCATION_ID_TYPE_ICP

New Zealand Installation Control Point.

MALO_ID class-attribute instance-attribute

MALO_ID = (
    market_location_pb2.MARKET_LOCATION_ID_TYPE_MALO_ID
)

Germany Marktlokations-ID.

MPAN class-attribute instance-attribute

MPAN = market_location_pb2.MARKET_LOCATION_ID_TYPE_MPAN

United Kingdom Meter Point Administration Number.

NMI class-attribute instance-attribute

NMI = market_location_pb2.MARKET_LOCATION_ID_TYPE_NMI

Australia National Metering Identifier.

OTHER class-attribute instance-attribute

OTHER = market_location_pb2.MARKET_LOCATION_ID_TYPE_OTHER

Generic identifier for markets not modeled explicitly.

POD class-attribute instance-attribute

POD = market_location_pb2.MARKET_LOCATION_ID_TYPE_POD

Italy Point of Delivery.

PRM class-attribute instance-attribute

PRM = market_location_pb2.MARKET_LOCATION_ID_TYPE_PRM

France Point de Reference et Mesure.

SPN class-attribute instance-attribute

SPN = market_location_pb2.MARKET_LOCATION_ID_TYPE_SPN

Japan Supply Point Number.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    market_location_pb2.MARKET_LOCATION_ID_TYPE_UNSPECIFIED
)

The market location ID type is unspecified.

ZAEHLPUNKT class-attribute instance-attribute

ZAEHLPUNKT = (
    market_location_pb2.MARKET_LOCATION_ID_TYPE_ZAEHLPUNKT
)

Austria Zaehlpunktbezeichnung.

frequenz.client.assets.MarketLocationRef dataclass

A reference to a market-facing metering point in a specific market area.

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

@dataclass(frozen=True, kw_only=True)
class MarketLocationRef:
    """A reference to a market-facing metering point in a specific market area."""

    market_area: int
    """The market area in which this market location is registered."""

    market_location_id: MarketLocationId | None
    """The official market location identifier."""

Attributes

market_area instance-attribute

market_area: int

The market area in which this market location is registered.

market_location_id instance-attribute

market_location_id: MarketLocationId | None

The official market location identifier.

frequenz.client.assets.MarketParticipation dataclass

A relation's participation in a specific market use case.

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

@dataclass(frozen=True, kw_only=True)
class MarketParticipation:
    """A relation's participation in a specific market use case."""

    type: MarketParticipationType | int
    """The use case for which this relation participates."""

    validity_period: Interval | None
    """Configured validity interval for this participation."""

Attributes

type instance-attribute

type: MarketParticipationType | int

The use case for which this relation participates.

validity_period instance-attribute

validity_period: Interval | None

Configured validity interval for this participation.

frequenz.client.assets.MarketParticipationType

Bases: Enum

Market-related use cases for topology relations.

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

@enum.unique
class MarketParticipationType(enum.Enum):
    """Market-related use cases for topology relations."""

    UNSPECIFIED = platformassets_pb2.MARKET_PARTICIPATION_TYPE_UNSPECIFIED
    """The market participation type is unspecified."""

    ENERGY_TRADING = platformassets_pb2.MARKET_PARTICIPATION_TYPE_ENERGY_TRADING
    """Energy trading, supply, balancing, or settlement participation."""

    FLEX_MARKETS = platformassets_pb2.MARKET_PARTICIPATION_TYPE_FLEX_MARKETS
    """Flex-market or ancillary-service participation."""

Attributes

ENERGY_TRADING class-attribute instance-attribute

ENERGY_TRADING = (
    platformassets_pb2.MARKET_PARTICIPATION_TYPE_ENERGY_TRADING
)

Energy trading, supply, balancing, or settlement participation.

FLEX_MARKETS class-attribute instance-attribute

FLEX_MARKETS = (
    platformassets_pb2.MARKET_PARTICIPATION_TYPE_FLEX_MARKETS
)

Flex-market or ancillary-service participation.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    platformassets_pb2.MARKET_PARTICIPATION_TYPE_UNSPECIFIED
)

The market participation type is unspecified.

frequenz.client.assets.MarketTopologyRelation dataclass

A relation between a gridpool, microgrid, and market location.

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

@dataclass(frozen=True, kw_only=True)
class MarketTopologyRelation:
    """A relation between a gridpool, microgrid, and market location."""

    microgrid_id: MicrogridId | None
    """The microgrid associated with this relation."""

    market_location_ref: MarketLocationRef | None
    """The market location associated with this relation."""

    gridpool_id: int | None
    """The gridpool associated with this relation."""

    delivery_area: DeliveryArea | None
    """Delivery area in which this relation applies."""

    participations: list[MarketParticipation]
    """Use-case-specific participations for this relation."""

Attributes

delivery_area instance-attribute

delivery_area: DeliveryArea | None

Delivery area in which this relation applies.

gridpool_id instance-attribute

gridpool_id: int | None

The gridpool associated with this relation.

market_location_ref instance-attribute

market_location_ref: MarketLocationRef | None

The market location associated with this relation.

microgrid_id instance-attribute

microgrid_id: MicrogridId | None

The microgrid associated with this relation.

participations instance-attribute

participations: list[MarketParticipation]

Use-case-specific participations for this relation.

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.

Methods:

__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_pb2.MICROGRID_STATUS_ACTIVE

The microgrid is active.

INACTIVE class-attribute instance-attribute

INACTIVE = microgrid_pb2.MICROGRID_STATUS_INACTIVE

The microgrid is inactive.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = microgrid_pb2.MICROGRID_STATUS_UNSPECIFIED

The status is unspecified. This should not be used.

frequenz.client.assets.Sensor dataclass

A sensor that measures a physical metric in a microgrid environment.

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

@dataclass(frozen=True, kw_only=True)
class Sensor:
    """A sensor that measures a physical metric in a microgrid environment."""

    id: SensorId
    """The unique identifier of the sensor."""

    microgrid_id: MicrogridId
    """The unique identifier of the parent microgrid."""

    name: str | None
    """The human-readable sensor name."""

    model: str | None
    """The sensor model name."""

    operational_lifetime: Lifetime | None
    """The operational lifetime of the sensor."""

Attributes

id instance-attribute

id: SensorId

The unique identifier of the sensor.

microgrid_id instance-attribute

microgrid_id: MicrogridId

The unique identifier of the parent microgrid.

model instance-attribute

model: str | None

The sensor model name.

name instance-attribute

name: str | None

The human-readable sensor name.

operational_lifetime instance-attribute

operational_lifetime: Lifetime | None

The operational lifetime of the sensor.