Index

frequenz.client.marketmetering

Market Metering API client for Python.

This package provides a Python client for the Frequenz Market Metering API, which allows streaming historical and real-time metering samples from Market Locations.

Example

from datetime import datetime, timezone
from frequenz.client.marketmetering import MarketMeteringApiClient
from frequenz.client.marketmetering.types import (
    EnergyFlowDirection,
    MarketArea,
    MarketLocationId,
    MarketLocationIdType,
    MarketLocationRef,
    MetricType,
)

client = MarketMeteringApiClient(
    server_url="grpc://marketmetering.example.com",
    auth_key="your-api-key",
    sign_secret="your-sign-secret",
)

market_location = MarketLocationRef(
    market_area=MarketArea.EU_DE,
    market_location_id=MarketLocationId(
        value="DE01234567890",
        type=MarketLocationIdType.MALO_ID,
    ),
)

async for series in client.stream_samples(
    market_locations=[market_location],
    directions=[EnergyFlowDirection.IMPORT],
    metric_types=[MetricType.ACTIVE_ENERGY],
    start_time=datetime(2025, 1, 1, tzinfo=timezone.utc),
):
    for sample in series.samples:
        print(f"{sample.sample_time}: {sample.value}")

Classes

frequenz.client.marketmetering.ActivationFilter

Bases: Enum

Filter for Market Location activation status.

Source code in src/frequenz/client/marketmetering/types.py

class ActivationFilter(Enum):
    """Filter for Market Location activation status."""

    UNSPECIFIED = pb.ACTIVATION_FILTER_UNSPECIFIED
    """Unspecified filter (defaults to ONLY_ACTIVE)."""

    ONLY_ACTIVE = pb.ACTIVATION_FILTER_ONLY_ACTIVE
    """Return only active Market Locations."""

    ONLY_INACTIVE = pb.ACTIVATION_FILTER_ONLY_INACTIVE
    """Return only inactive Market Locations."""

    ALL = pb.ACTIVATION_FILTER_ALL
    """Return all Market Locations regardless of activation status."""

Attributes

ALL class-attribute instance-attribute

ALL = pb.ACTIVATION_FILTER_ALL

Return all Market Locations regardless of activation status.

ONLY_ACTIVE class-attribute instance-attribute

ONLY_ACTIVE = pb.ACTIVATION_FILTER_ONLY_ACTIVE

Return only active Market Locations.

ONLY_INACTIVE class-attribute instance-attribute

ONLY_INACTIVE = pb.ACTIVATION_FILTER_ONLY_INACTIVE

Return only inactive Market Locations.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.ACTIVATION_FILTER_UNSPECIFIED

Unspecified filter (defaults to ONLY_ACTIVE).

frequenz.client.marketmetering.DataQuality

Bases: Enum

Classification of data quality for a metering sample.

Source code in src/frequenz/client/marketmetering/types.py

class DataQuality(Enum):
    """Classification of data quality for a metering sample."""

    UNSPECIFIED = pb.DATA_QUALITY_UNSPECIFIED
    """Unspecified quality."""

    MEASURED = pb.DATA_QUALITY_MEASURED
    """Raw value directly received from the metering system."""

    ESTIMATED = pb.DATA_QUALITY_ESTIMATED
    """Value inferred or interpolated due to missing or invalid readings."""

    CORRECTED = pb.DATA_QUALITY_CORRECTED
    """Value that was initially delivered but later amended."""

    MISSING = pb.DATA_QUALITY_MISSING
    """No valid value is available for this interval."""

    def to_protobuf(self) -> int:
        """Convert to protobuf message.

        Returns:
            The protobuf representation (integer enum value).
        """
        return self.value

Attributes

CORRECTED class-attribute instance-attribute

CORRECTED = pb.DATA_QUALITY_CORRECTED

Value that was initially delivered but later amended.

ESTIMATED class-attribute instance-attribute

ESTIMATED = pb.DATA_QUALITY_ESTIMATED

Value inferred or interpolated due to missing or invalid readings.

MEASURED class-attribute instance-attribute

MEASURED = pb.DATA_QUALITY_MEASURED

Raw value directly received from the metering system.

MISSING class-attribute instance-attribute

MISSING = pb.DATA_QUALITY_MISSING

No valid value is available for this interval.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.DATA_QUALITY_UNSPECIFIED

Unspecified quality.

Methods:

to_protobuf

to_protobuf() -> int

Convert to protobuf message.

RETURNS	DESCRIPTION
int	The protobuf representation (integer enum value).

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> int:
    """Convert to protobuf message.

    Returns:
        The protobuf representation (integer enum value).
    """
    return self.value

frequenz.client.marketmetering.DownsamplingMethod

Bases: Enum

Defines how multiple native samples are combined when downsampling.

Source code in src/frequenz/client/marketmetering/types.py

class DownsamplingMethod(Enum):
    """Defines how multiple native samples are combined when downsampling."""

    UNSPECIFIED = pb.DOWNSAMPLING_METHOD_UNSPECIFIED
    """Defaults to MEAN."""

    MEAN = pb.DOWNSAMPLING_METHOD_MEAN
    """Arithmetic mean of all samples within the interval."""

Attributes

MEAN class-attribute instance-attribute

MEAN = pb.DOWNSAMPLING_METHOD_MEAN

Arithmetic mean of all samples within the interval.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.DOWNSAMPLING_METHOD_UNSPECIFIED

Defaults to MEAN.

frequenz.client.marketmetering.EnergyFlowDirection

Bases: Enum

Direction of energy flow for metering samples.

Source code in src/frequenz/client/marketmetering/types.py

class EnergyFlowDirection(Enum):
    """Direction of energy flow for metering samples."""

    UNSPECIFIED = pb.ENERGY_FLOW_DIRECTION_UNSPECIFIED
    """Unspecified direction."""

    IMPORT = pb.ENERGY_FLOW_DIRECTION_IMPORT
    """Energy flowing from grid to customer (consumption)."""

    EXPORT = pb.ENERGY_FLOW_DIRECTION_EXPORT
    """Energy flowing from customer to grid (generation / feed-in)."""

Attributes

EXPORT class-attribute instance-attribute

EXPORT = pb.ENERGY_FLOW_DIRECTION_EXPORT

Energy flowing from customer to grid (generation / feed-in).

IMPORT class-attribute instance-attribute

IMPORT = pb.ENERGY_FLOW_DIRECTION_IMPORT

Energy flowing from grid to customer (consumption).

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.ENERGY_FLOW_DIRECTION_UNSPECIFIED

Unspecified direction.

frequenz.client.marketmetering.MarketArea

Bases: Enum

Market area enum representing the jurisdiction.

Market areas are organized by geographical region: - EU_: Europe - NA_: North America - AP_: Asia-Pacific - OC_: Oceania - EURASIA_*: Eurasia

Source code in src/frequenz/client/marketmetering/types.py

class MarketArea(Enum):
    """Market area enum representing the jurisdiction.

    Market areas are organized by geographical region:
    - EU_*: Europe
    - NA_*: North America
    - AP_*: Asia-Pacific
    - OC_*: Oceania
    - EURASIA_*: Eurasia
    """

    UNSPECIFIED = market_area_pb.MARKET_AREA_UNSPECIFIED
    """Unspecified market area."""

    # Europe
    EU_DE = market_area_pb.MARKET_AREA_EU_DE
    """Germany (Marktlokation / MaLo)."""

    EU_UK = market_area_pb.MARKET_AREA_EU_UK
    """United Kingdom (MPAN)."""

    EU_IT = market_area_pb.MARKET_AREA_EU_IT
    """Italy (POD)."""

    EU_FR = market_area_pb.MARKET_AREA_EU_FR
    """France."""

    EU_ES = market_area_pb.MARKET_AREA_EU_ES
    """Spain (CUPS)."""

    EU_NL = market_area_pb.MARKET_AREA_EU_NL
    """Netherlands (EAN)."""

    EU_BE = market_area_pb.MARKET_AREA_EU_BE
    """Belgium (EAN)."""

    EU_CH = market_area_pb.MARKET_AREA_EU_CH
    """Switzerland."""

    EU_AT = market_area_pb.MARKET_AREA_EU_AT
    """Austria."""

    EU_NORDICS = market_area_pb.MARKET_AREA_EU_NORDICS
    """Nordic countries (Denmark, Finland, Norway, Sweden)."""

    # North America
    NA_US_ERCOT = market_area_pb.MARKET_AREA_NA_US_ERCOT
    """US - ERCOT region (ESI ID)."""

    NA_US_PJM = market_area_pb.MARKET_AREA_NA_US_PJM
    """US - PJM Interconnection."""

    NA_US_ISONE = market_area_pb.MARKET_AREA_NA_US_ISONE
    """US - ISO New England."""

    NA_US_CAISO = market_area_pb.MARKET_AREA_NA_US_CAISO
    """US - California ISO."""

    # Asia-Pacific
    AP_JP = market_area_pb.MARKET_AREA_AP_JP
    """Japan."""

    AP_CN = market_area_pb.MARKET_AREA_AP_CN
    """China."""

    AP_IN = market_area_pb.MARKET_AREA_AP_IN
    """India."""

    AP_SG = market_area_pb.MARKET_AREA_AP_SG
    """Singapore."""

    # Oceania
    OC_AU = market_area_pb.MARKET_AREA_OC_AU
    """Australia (NMI)."""

    OC_NZ = market_area_pb.MARKET_AREA_OC_NZ
    """New Zealand (ICP)."""

    # Eurasia
    EURASIA_RU = market_area_pb.MARKET_AREA_EURASIA_RU
    """Russia."""

    OTHER = market_area_pb.MARKET_AREA_OTHER
    """Other or not yet modelled areas."""

Attributes

AP_CN class-attribute instance-attribute

AP_CN = market_area_pb.MARKET_AREA_AP_CN

China.

AP_IN class-attribute instance-attribute

AP_IN = market_area_pb.MARKET_AREA_AP_IN

India.

AP_JP class-attribute instance-attribute

AP_JP = market_area_pb.MARKET_AREA_AP_JP

Japan.

AP_SG class-attribute instance-attribute

AP_SG = market_area_pb.MARKET_AREA_AP_SG

Singapore.

EURASIA_RU class-attribute instance-attribute

EURASIA_RU = market_area_pb.MARKET_AREA_EURASIA_RU

Russia.

EU_AT class-attribute instance-attribute

EU_AT = market_area_pb.MARKET_AREA_EU_AT

Austria.

EU_BE class-attribute instance-attribute

EU_BE = market_area_pb.MARKET_AREA_EU_BE

Belgium (EAN).

EU_CH class-attribute instance-attribute

EU_CH = market_area_pb.MARKET_AREA_EU_CH

Switzerland.

EU_DE class-attribute instance-attribute

EU_DE = market_area_pb.MARKET_AREA_EU_DE

Germany (Marktlokation / MaLo).

EU_ES class-attribute instance-attribute

EU_ES = market_area_pb.MARKET_AREA_EU_ES

Spain (CUPS).

EU_FR class-attribute instance-attribute

EU_FR = market_area_pb.MARKET_AREA_EU_FR

France.

EU_IT class-attribute instance-attribute

EU_IT = market_area_pb.MARKET_AREA_EU_IT

Italy (POD).

EU_NL class-attribute instance-attribute

EU_NL = market_area_pb.MARKET_AREA_EU_NL

Netherlands (EAN).

EU_NORDICS class-attribute instance-attribute

EU_NORDICS = market_area_pb.MARKET_AREA_EU_NORDICS

Nordic countries (Denmark, Finland, Norway, Sweden).

EU_UK class-attribute instance-attribute

EU_UK = market_area_pb.MARKET_AREA_EU_UK

United Kingdom (MPAN).

NA_US_CAISO class-attribute instance-attribute

NA_US_CAISO = market_area_pb.MARKET_AREA_NA_US_CAISO

US - California ISO.

NA_US_ERCOT class-attribute instance-attribute

NA_US_ERCOT = market_area_pb.MARKET_AREA_NA_US_ERCOT

US - ERCOT region (ESI ID).

NA_US_ISONE class-attribute instance-attribute

NA_US_ISONE = market_area_pb.MARKET_AREA_NA_US_ISONE

US - ISO New England.

NA_US_PJM class-attribute instance-attribute

NA_US_PJM = market_area_pb.MARKET_AREA_NA_US_PJM

US - PJM Interconnection.

OC_AU class-attribute instance-attribute

OC_AU = market_area_pb.MARKET_AREA_OC_AU

Australia (NMI).

OC_NZ class-attribute instance-attribute

OC_NZ = market_area_pb.MARKET_AREA_OC_NZ

New Zealand (ICP).

OTHER class-attribute instance-attribute

OTHER = market_area_pb.MARKET_AREA_OTHER

Other or not yet modelled areas.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = market_area_pb.MARKET_AREA_UNSPECIFIED

Unspecified market area.

frequenz.client.marketmetering.MarketLocation dataclass

A Market Location with its configuration.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocation:
    """A Market Location with its configuration."""

    display_name: str
    """Human-readable name of the Market Location."""

    supported_directions: list[EnergyFlowDirection]
    """Supported energy flow directions."""

    time_resolution: TimeResolution
    """Time resolution of the metering data."""

    payload: dict[str, Any]
    """Additional arbitrary metadata."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationMetadata) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocation instance.
        """
        return cls(
            display_name=pb_obj.display_name,
            supported_directions=[
                EnergyFlowDirection(d) for d in pb_obj.supported_directions
            ],
            time_resolution=TimeResolution(pb_obj.time_resolution),
            payload=dict(pb_obj.payload.items()),
        )

    def to_protobuf(self) -> pb.MarketLocationMetadata:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        pb_struct = struct_pb2.Struct()
        pb_struct.update(self.payload)

        return pb.MarketLocationMetadata(
            display_name=self.display_name,
            supported_directions=[d.value for d in self.supported_directions],
            time_resolution=self.time_resolution.value,
            payload=pb_struct,
        )

Attributes

display_name instance-attribute

display_name: str

Human-readable name of the Market Location.

payload instance-attribute

payload: dict[str, Any]

Additional arbitrary metadata.

supported_directions instance-attribute

supported_directions: list[EnergyFlowDirection]

Supported energy flow directions.

time_resolution instance-attribute

time_resolution: TimeResolution

Time resolution of the metering data.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationMetadata) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationMetadata

RETURNS	DESCRIPTION
Self	A new MarketLocation instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationMetadata) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocation instance.
    """
    return cls(
        display_name=pb_obj.display_name,
        supported_directions=[
            EnergyFlowDirection(d) for d in pb_obj.supported_directions
        ],
        time_resolution=TimeResolution(pb_obj.time_resolution),
        payload=dict(pb_obj.payload.items()),
    )

to_protobuf

to_protobuf() -> MarketLocationMetadata

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationMetadata	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pb.MarketLocationMetadata:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    pb_struct = struct_pb2.Struct()
    pb_struct.update(self.payload)

    return pb.MarketLocationMetadata(
        display_name=self.display_name,
        supported_directions=[d.value for d in self.supported_directions],
        time_resolution=self.time_resolution.value,
        payload=pb_struct,
    )

frequenz.client.marketmetering.MarketLocationChangedField

Bases: Enum

Fields that can be tracked for changes in Market Location history.

Source code in src/frequenz/client/marketmetering/types.py

class MarketLocationChangedField(Enum):
    """Fields that can be tracked for changes in Market Location history."""

    UNSPECIFIED = pb.MARKET_LOCATION_CHANGED_FIELD_UNSPECIFIED
    """Unspecified field."""

    SUPPORTED_DIRECTIONS = pb.MARKET_LOCATION_CHANGED_FIELD_SUPPORTED_DIRECTIONS
    """Supported energy flow directions."""

    TIME_RESOLUTION = pb.MARKET_LOCATION_CHANGED_FIELD_TIME_RESOLUTION
    """Time resolution of the metering data."""

    IS_ACTIVE = pb.MARKET_LOCATION_CHANGED_FIELD_IS_ACTIVE
    """Activity status of the Market Location."""

Attributes

IS_ACTIVE class-attribute instance-attribute

IS_ACTIVE = pb.MARKET_LOCATION_CHANGED_FIELD_IS_ACTIVE

Activity status of the Market Location.

SUPPORTED_DIRECTIONS class-attribute instance-attribute

SUPPORTED_DIRECTIONS = (
    pb.MARKET_LOCATION_CHANGED_FIELD_SUPPORTED_DIRECTIONS
)

Supported energy flow directions.

TIME_RESOLUTION class-attribute instance-attribute

TIME_RESOLUTION = (
    pb.MARKET_LOCATION_CHANGED_FIELD_TIME_RESOLUTION
)

Time resolution of the metering data.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.MARKET_LOCATION_CHANGED_FIELD_UNSPECIFIED

Unspecified field.

frequenz.client.marketmetering.MarketLocationDetail dataclass

A Market Location with server-managed metadata.

This includes the core MarketLocation configuration plus server-assigned fields such as revision, activation status, and timestamps.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationDetail:
    """A Market Location with server-managed metadata.

    This includes the core `MarketLocation` configuration plus
    server-assigned fields such as revision, activation status,
    and timestamps.
    """

    market_location_ref: MarketLocationRef
    """Reference to this Market Location."""

    market_location: MarketLocation
    """The core Market Location configuration."""

    revision: int
    """Server-managed revision number (monotonically increasing)."""

    is_active: bool
    """Whether the Market Location is currently active."""

    create_time: datetime
    """Timestamp when the Market Location was created."""

    update_time: datetime
    """Timestamp of the last update to this Market Location."""

    last_deactivated_time: datetime | None
    """Timestamp when the Market Location was last deactivated, if ever."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationDetail) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationDetail instance.
        """
        last_deactivated = None
        if pb_obj.HasField("last_deactivated_time"):
            last_deactivated = _timestamp_to_datetime(pb_obj.last_deactivated_time)

        return cls(
            market_location_ref=MarketLocationRef.from_protobuf(
                pb_obj.market_location_ref
            ),
            market_location=MarketLocation.from_protobuf(pb_obj.market_location),
            revision=pb_obj.revision,
            is_active=pb_obj.is_active,
            create_time=_timestamp_to_datetime(pb_obj.create_time),
            update_time=_timestamp_to_datetime(pb_obj.update_time),
            last_deactivated_time=last_deactivated,
        )

Attributes

create_time instance-attribute

create_time: datetime

Timestamp when the Market Location was created.

is_active instance-attribute

is_active: bool

Whether the Market Location is currently active.

last_deactivated_time instance-attribute

last_deactivated_time: datetime | None

Timestamp when the Market Location was last deactivated, if ever.

market_location instance-attribute

market_location: MarketLocation

The core Market Location configuration.

market_location_ref instance-attribute

market_location_ref: MarketLocationRef

Reference to this Market Location.

revision instance-attribute

revision: int

Server-managed revision number (monotonically increasing).

update_time instance-attribute

update_time: datetime

Timestamp of the last update to this Market Location.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationDetail) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationDetail

RETURNS	DESCRIPTION
Self	A new MarketLocationDetail instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationDetail) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationDetail instance.
    """
    last_deactivated = None
    if pb_obj.HasField("last_deactivated_time"):
        last_deactivated = _timestamp_to_datetime(pb_obj.last_deactivated_time)

    return cls(
        market_location_ref=MarketLocationRef.from_protobuf(
            pb_obj.market_location_ref
        ),
        market_location=MarketLocation.from_protobuf(pb_obj.market_location),
        revision=pb_obj.revision,
        is_active=pb_obj.is_active,
        create_time=_timestamp_to_datetime(pb_obj.create_time),
        update_time=_timestamp_to_datetime(pb_obj.update_time),
        last_deactivated_time=last_deactivated,
    )

frequenz.client.marketmetering.MarketLocationEntry dataclass

A Market Location entry as returned by list operations.

Wraps a MarketLocationDetail together with the owning enterprise ID.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationEntry:
    """A Market Location entry as returned by list operations.

    Wraps a `MarketLocationDetail` together with the owning enterprise ID.
    """

    enterprise_id: int
    """Enterprise ID owning this Market Location."""

    market_location_detail: MarketLocationDetail
    """Full Market Location details including metadata."""

    @property
    def market_location_ref(self) -> MarketLocationRef:
        """Reference to this Market Location."""
        return self.market_location_detail.market_location_ref

    @property
    def market_location(self) -> MarketLocation:
        """The core Market Location configuration (convenience accessor)."""
        return self.market_location_detail.market_location

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationDetail) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationEntry instance.
        """
        return cls(
            enterprise_id=pb_obj.market_location_ref.enterprise_id,
            market_location_detail=MarketLocationDetail.from_protobuf(pb_obj),
        )

Attributes

enterprise_id instance-attribute

enterprise_id: int

Enterprise ID owning this Market Location.

market_location property

market_location: MarketLocation

The core Market Location configuration (convenience accessor).

market_location_detail instance-attribute

market_location_detail: MarketLocationDetail

Full Market Location details including metadata.

market_location_ref property

market_location_ref: MarketLocationRef

Reference to this Market Location.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationDetail) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationDetail

RETURNS	DESCRIPTION
Self	A new MarketLocationEntry instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationDetail) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationEntry instance.
    """
    return cls(
        enterprise_id=pb_obj.market_location_ref.enterprise_id,
        market_location_detail=MarketLocationDetail.from_protobuf(pb_obj),
    )

frequenz.client.marketmetering.MarketLocationId dataclass

Market-standard identifier describing a Market Location.

A Market Location is a jurisdiction-specific point of metering used for regulatory processes such as settlement, billing, supplier switching, etc.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationId:
    """Market-standard identifier describing a Market Location.

    A Market Location is a jurisdiction-specific point of metering used for
    regulatory processes such as settlement, billing, supplier switching, etc.
    """

    value: str
    """Opaque identifier in its original market format."""

    type: MarketLocationIdType
    """Type of official market identifier."""

    @classmethod
    def from_protobuf(cls, pb_obj: grid_pb.MarketLocationId) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationId instance.
        """
        return cls(
            value=pb_obj.id.value,
            type=MarketLocationIdType(pb_obj.type),
        )

    def to_protobuf(self) -> grid_pb.MarketLocationId:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        return grid_pb.MarketLocationId(
            id=grid_pb.MarketLocationIdValue(value=self.value),
            type=self.type.value,
        )

    def to_id_value_protobuf(self) -> grid_pb.MarketLocationIdValue:
        """Convert to a MarketLocationIdValue protobuf message.

        Returns:
            The protobuf representation containing only the value.
        """
        return grid_pb.MarketLocationIdValue(value=self.value)

Attributes

type instance-attribute

type: MarketLocationIdType

Type of official market identifier.

value instance-attribute

value: str

Opaque identifier in its original market format.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationId) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationId

RETURNS	DESCRIPTION
Self	A new MarketLocationId instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: grid_pb.MarketLocationId) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationId instance.
    """
    return cls(
        value=pb_obj.id.value,
        type=MarketLocationIdType(pb_obj.type),
    )

to_id_value_protobuf

to_id_value_protobuf() -> MarketLocationIdValue

Convert to a MarketLocationIdValue protobuf message.

RETURNS	DESCRIPTION
MarketLocationIdValue	The protobuf representation containing only the value.

Source code in src/frequenz/client/marketmetering/types.py

def to_id_value_protobuf(self) -> grid_pb.MarketLocationIdValue:
    """Convert to a MarketLocationIdValue protobuf message.

    Returns:
        The protobuf representation containing only the value.
    """
    return grid_pb.MarketLocationIdValue(value=self.value)

to_protobuf

to_protobuf() -> MarketLocationId

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationId	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> grid_pb.MarketLocationId:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    return grid_pb.MarketLocationId(
        id=grid_pb.MarketLocationIdValue(value=self.value),
        type=self.type.value,
    )

frequenz.client.marketmetering.MarketLocationIdType

Bases: Enum

Type of external market identifier.

Source code in src/frequenz/client/marketmetering/types.py

class MarketLocationIdType(Enum):
    """Type of external market identifier."""

    UNSPECIFIED = grid_pb.MARKET_LOCATION_ID_TYPE_UNSPECIFIED
    """Unspecified identifier type."""

    MALO_ID = grid_pb.MARKET_LOCATION_ID_TYPE_MALO_ID
    """Germany – Marktlokations-ID (MaLo-ID)."""

    ZAEHLPUNKT = grid_pb.MARKET_LOCATION_ID_TYPE_ZAEHLPUNKT
    """Austria – Zählpunktbezeichnung."""

    MPAN = grid_pb.MARKET_LOCATION_ID_TYPE_MPAN
    """United Kingdom – Meter Point Administration Number."""

    POD = grid_pb.MARKET_LOCATION_ID_TYPE_POD
    """Italy – Punto di Prelievo (Point of Delivery)."""

    CUPS = grid_pb.MARKET_LOCATION_ID_TYPE_CUPS
    """Spain – Código Unificado de Punto de Suministro."""

    PRM = grid_pb.MARKET_LOCATION_ID_TYPE_PRM
    """France – Point de Référence et Mesure (PRM)."""

    EAN = grid_pb.MARKET_LOCATION_ID_TYPE_EAN
    """European Article Number (used in Netherlands, Belgium, etc.)."""

    GSRN = grid_pb.MARKET_LOCATION_ID_TYPE_GSRN
    """Nordic countries – GS1 Global Service Relation Number."""

    ESI_ID = grid_pb.MARKET_LOCATION_ID_TYPE_ESI_ID
    """United States – Electric Service Identifier (ESI ID)."""

    NMI = grid_pb.MARKET_LOCATION_ID_TYPE_NMI
    """Australia – National Metering Identifier."""

    ICP = grid_pb.MARKET_LOCATION_ID_TYPE_ICP
    """New Zealand – Installation Control Point."""

    SPN = grid_pb.MARKET_LOCATION_ID_TYPE_SPN
    """Japan – Supply Point Number."""

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

Attributes

CUPS class-attribute instance-attribute

CUPS = grid_pb.MARKET_LOCATION_ID_TYPE_CUPS

Spain – Código Unificado de Punto de Suministro.

EAN class-attribute instance-attribute

EAN = grid_pb.MARKET_LOCATION_ID_TYPE_EAN

European Article Number (used in Netherlands, Belgium, etc.).

ESI_ID class-attribute instance-attribute

ESI_ID = grid_pb.MARKET_LOCATION_ID_TYPE_ESI_ID

United States – Electric Service Identifier (ESI ID).

GSRN class-attribute instance-attribute

GSRN = grid_pb.MARKET_LOCATION_ID_TYPE_GSRN

Nordic countries – GS1 Global Service Relation Number.

ICP class-attribute instance-attribute

ICP = grid_pb.MARKET_LOCATION_ID_TYPE_ICP

New Zealand – Installation Control Point.

MALO_ID class-attribute instance-attribute

MALO_ID = grid_pb.MARKET_LOCATION_ID_TYPE_MALO_ID

Germany – Marktlokations-ID (MaLo-ID).

MPAN class-attribute instance-attribute

MPAN = grid_pb.MARKET_LOCATION_ID_TYPE_MPAN

United Kingdom – Meter Point Administration Number.

NMI class-attribute instance-attribute

NMI = grid_pb.MARKET_LOCATION_ID_TYPE_NMI

Australia – National Metering Identifier.

OTHER class-attribute instance-attribute

OTHER = grid_pb.MARKET_LOCATION_ID_TYPE_OTHER

Generic meter identifier for markets not modeled explicitly.

POD class-attribute instance-attribute

POD = grid_pb.MARKET_LOCATION_ID_TYPE_POD

Italy – Punto di Prelievo (Point of Delivery).

PRM class-attribute instance-attribute

PRM = grid_pb.MARKET_LOCATION_ID_TYPE_PRM

France – Point de Référence et Mesure (PRM).

SPN class-attribute instance-attribute

SPN = grid_pb.MARKET_LOCATION_ID_TYPE_SPN

Japan – Supply Point Number.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = grid_pb.MARKET_LOCATION_ID_TYPE_UNSPECIFIED

Unspecified identifier type.

ZAEHLPUNKT class-attribute instance-attribute

ZAEHLPUNKT = grid_pb.MARKET_LOCATION_ID_TYPE_ZAEHLPUNKT

Austria – Zählpunktbezeichnung.

frequenz.client.marketmetering.MarketLocationOperationError dataclass

Structured error for a rejected Market Location operation.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationOperationError:
    """Structured error for a rejected Market Location operation."""

    code: MarketLocationOperationErrorCode
    """Machine-readable classification of the failure."""

    message: str
    """Human-readable diagnostic message for logging and debugging."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationOperationError) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationOperationError instance.
        """
        return cls(
            code=MarketLocationOperationErrorCode(pb_obj.code),
            message=pb_obj.message,
        )

Attributes

code instance-attribute

code: MarketLocationOperationErrorCode

Machine-readable classification of the failure.

message instance-attribute

message: str

Human-readable diagnostic message for logging and debugging.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationOperationError) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationOperationError

RETURNS	DESCRIPTION
Self	A new MarketLocationOperationError instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationOperationError) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationOperationError instance.
    """
    return cls(
        code=MarketLocationOperationErrorCode(pb_obj.code),
        message=pb_obj.message,
    )

frequenz.client.marketmetering.MarketLocationOperationErrorCode

Bases: Enum

Error codes for Market Location activate/deactivate operations.

Source code in src/frequenz/client/marketmetering/types.py

class MarketLocationOperationErrorCode(Enum):
    """Error codes for Market Location activate/deactivate operations."""

    UNSPECIFIED = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_UNSPECIFIED
    """Unspecified error."""

    NOT_FOUND = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_NOT_FOUND
    """The Market Location was not found."""

    ALREADY_IN_TARGET_STATE = (
        pb.MARKET_LOCATION_OPERATION_ERROR_CODE_ALREADY_IN_TARGET_STATE
    )
    """The Market Location is already in the requested state."""

    ENTERPRISE_MISMATCH = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_ENTERPRISE_MISMATCH
    """The enterprise ID does not match the Market Location's owner."""

    PERMISSION_DENIED = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_PERMISSION_DENIED
    """The caller does not have permission for this operation."""

    OPERATION_REJECTED = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_OPERATION_REJECTED
    """The operation was rejected by the server."""

    UNKNOWN_ERROR = pb.MARKET_LOCATION_OPERATION_ERROR_CODE_UNKNOWN_ERROR
    """Unknown error."""

Attributes

ALREADY_IN_TARGET_STATE class-attribute instance-attribute

ALREADY_IN_TARGET_STATE = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_ALREADY_IN_TARGET_STATE
)

The Market Location is already in the requested state.

ENTERPRISE_MISMATCH class-attribute instance-attribute

ENTERPRISE_MISMATCH = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_ENTERPRISE_MISMATCH
)

The enterprise ID does not match the Market Location's owner.

NOT_FOUND class-attribute instance-attribute

NOT_FOUND = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_NOT_FOUND
)

The Market Location was not found.

OPERATION_REJECTED class-attribute instance-attribute

OPERATION_REJECTED = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_OPERATION_REJECTED
)

The operation was rejected by the server.

PERMISSION_DENIED class-attribute instance-attribute

PERMISSION_DENIED = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_PERMISSION_DENIED
)

The caller does not have permission for this operation.

UNKNOWN_ERROR class-attribute instance-attribute

UNKNOWN_ERROR = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_UNKNOWN_ERROR
)

Unknown error.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = (
    pb.MARKET_LOCATION_OPERATION_ERROR_CODE_UNSPECIFIED
)

Unspecified error.

frequenz.client.marketmetering.MarketLocationOperationResult dataclass

Result of an activate or deactivate operation on a Market Location.

If error is None the operation succeeded; otherwise it was rejected.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationOperationResult:
    """Result of an activate or deactivate operation on a Market Location.

    If ``error`` is ``None`` the operation succeeded; otherwise it was rejected.
    """

    market_location_ref: MarketLocationRef
    """Reference to the Market Location."""

    update_time: datetime | None
    """Timestamp of the operation. Populated only on success."""

    revision: int
    """Server-managed revision after the operation. Populated only on success."""

    error: MarketLocationOperationError | None
    """Error details if the operation was rejected, ``None`` on success."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationOperationResult) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationOperationResult instance.
        """
        update_time = None
        if pb_obj.HasField("update_time"):
            update_time = _timestamp_to_datetime(pb_obj.update_time)

        error = None
        if pb_obj.HasField("error"):
            error = MarketLocationOperationError.from_protobuf(pb_obj.error)

        return cls(
            market_location_ref=MarketLocationRef.from_protobuf(
                pb_obj.market_location_ref
            ),
            update_time=update_time,
            revision=pb_obj.revision,
            error=error,
        )

Attributes

error instance-attribute

error: MarketLocationOperationError | None

Error details if the operation was rejected, None on success.

market_location_ref instance-attribute

market_location_ref: MarketLocationRef

Reference to the Market Location.

revision instance-attribute

revision: int

Server-managed revision after the operation. Populated only on success.

update_time instance-attribute

update_time: datetime | None

Timestamp of the operation. Populated only on success.

Methods:

from_protobuf classmethod

from_protobuf(
    pb_obj: MarketLocationOperationResult,
) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationOperationResult

RETURNS	DESCRIPTION
Self	A new MarketLocationOperationResult instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationOperationResult) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationOperationResult instance.
    """
    update_time = None
    if pb_obj.HasField("update_time"):
        update_time = _timestamp_to_datetime(pb_obj.update_time)

    error = None
    if pb_obj.HasField("error"):
        error = MarketLocationOperationError.from_protobuf(pb_obj.error)

    return cls(
        market_location_ref=MarketLocationRef.from_protobuf(
            pb_obj.market_location_ref
        ),
        update_time=update_time,
        revision=pb_obj.revision,
        error=error,
    )

frequenz.client.marketmetering.MarketLocationRef dataclass

Reference to a Market Location within a market area.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationRef:
    """Reference to a Market Location within a market area."""

    market_area: MarketArea
    """Regulatory jurisdiction in which this Market Location is registered."""

    market_location_id: MarketLocationId
    """Market-wide identifier (MaLo, MPAN, ESI-ID, NMI, ...)."""

    enterprise_id: int = field(default=0, kw_only=True)
    """Owning enterprise ID, when returned by the service."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationRef) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationRef instance.
        """
        return cls(
            enterprise_id=pb_obj.enterprise_id,
            market_area=MarketArea(pb_obj.market_location.market_area),
            market_location_id=MarketLocationId.from_protobuf(
                pb_obj.market_location.market_location_id
            ),
        )

    def to_protobuf(self) -> grid_pb.MarketLocationRef:
        """Convert to protobuf message.

        Returns:
            The enterprise-less protobuf selector used in requests.
        """
        return grid_pb.MarketLocationRef(
            market_area=self.market_area.value,
            market_location_id=self.market_location_id.to_protobuf(),
        )

Attributes

enterprise_id class-attribute instance-attribute

enterprise_id: int = field(default=0, kw_only=True)

Owning enterprise ID, when returned by the service.

market_area instance-attribute

market_area: MarketArea

Regulatory jurisdiction in which this Market Location is registered.

market_location_id instance-attribute

market_location_id: MarketLocationId

Market-wide identifier (MaLo, MPAN, ESI-ID, NMI, ...).

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationRef) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationRef

RETURNS	DESCRIPTION
Self	A new MarketLocationRef instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationRef) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationRef instance.
    """
    return cls(
        enterprise_id=pb_obj.enterprise_id,
        market_area=MarketArea(pb_obj.market_location.market_area),
        market_location_id=MarketLocationId.from_protobuf(
            pb_obj.market_location.market_location_id
        ),
    )

to_protobuf

to_protobuf() -> MarketLocationRef

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationRef	The enterprise-less protobuf selector used in requests.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> grid_pb.MarketLocationRef:
    """Convert to protobuf message.

    Returns:
        The enterprise-less protobuf selector used in requests.
    """
    return grid_pb.MarketLocationRef(
        market_area=self.market_area.value,
        market_location_id=self.market_location_id.to_protobuf(),
    )

frequenz.client.marketmetering.MarketLocationSample dataclass

A metering sample with metadata.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationSample:
    """A metering sample with metadata."""

    sample_time: datetime
    """Timestamp of the sample (UTC)."""

    value: float | None
    """Numeric value, if available."""

    quality: DataQuality
    """Quality classification of the sample."""

    revision: int | None
    """Revision number for native samples."""

    update_time: datetime | None
    """Timestamp when the sample was last updated."""

    resampling_method: ResamplingMethod
    """How this sample was produced."""

    @classmethod
    def from_protobuf(
        cls, pb_obj: pb.MarketLocationSample | pb.MarketLocationSampleDetail
    ) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message (simple or detailed).

        Returns:
            A new MarketLocationSample instance.
        """
        # Handle both MarketLocationSample and MarketLocationSampleDetail
        update_time = None
        if isinstance(pb_obj, pb.MarketLocationSampleDetail) and pb_obj.HasField(
            "sample_update_time"
        ):
            update_time = _timestamp_to_datetime(pb_obj.sample_update_time)

        return cls(
            sample_time=_timestamp_to_datetime(pb_obj.sample_time),
            value=pb_obj.value if pb_obj.HasField("value") else None,
            quality=DataQuality(pb_obj.quality),
            # revision is an int32, which doesn't support HasField in proto3 unless optional.
            # Assuming it's a standard field, 0 is the default.
            # If the API treats 0 as a valid revision, we can just use it.
            # If 0 means "not set", we'd check for 0.
            # Here we assume it's always present or defaults to 0.
            revision=pb_obj.revision,
            update_time=update_time,
            resampling_method=(
                ResamplingMethod(pb_obj.resampling_method)
                if isinstance(pb_obj, pb.MarketLocationSampleDetail)
                # Fallback to UNSPECIFIED if it's a simple MarketLocationSample
                # which does not have this field.
                else ResamplingMethod.UNSPECIFIED
            ),
        )

    def to_protobuf(self) -> pb.MarketLocationSample:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        sample = pb.MarketLocationSample(
            sample_time=_datetime_to_timestamp(self.sample_time),
            value=self.value,
            quality=self.quality.value,
        )
        if self.revision is not None:
            sample.revision = self.revision
        return sample

Attributes

quality instance-attribute

quality: DataQuality

Quality classification of the sample.

resampling_method instance-attribute

resampling_method: ResamplingMethod

How this sample was produced.

revision instance-attribute

revision: int | None

Revision number for native samples.

sample_time instance-attribute

sample_time: datetime

Timestamp of the sample (UTC).

update_time instance-attribute

update_time: datetime | None

Timestamp when the sample was last updated.

value instance-attribute

value: float | None

Numeric value, if available.

Methods:

from_protobuf classmethod

from_protobuf(
    pb_obj: (
        MarketLocationSample | MarketLocationSampleDetail
    ),
) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message (simple or detailed). TYPE: MarketLocationSample | MarketLocationSampleDetail

RETURNS	DESCRIPTION
Self	A new MarketLocationSample instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(
    cls, pb_obj: pb.MarketLocationSample | pb.MarketLocationSampleDetail
) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message (simple or detailed).

    Returns:
        A new MarketLocationSample instance.
    """
    # Handle both MarketLocationSample and MarketLocationSampleDetail
    update_time = None
    if isinstance(pb_obj, pb.MarketLocationSampleDetail) and pb_obj.HasField(
        "sample_update_time"
    ):
        update_time = _timestamp_to_datetime(pb_obj.sample_update_time)

    return cls(
        sample_time=_timestamp_to_datetime(pb_obj.sample_time),
        value=pb_obj.value if pb_obj.HasField("value") else None,
        quality=DataQuality(pb_obj.quality),
        # revision is an int32, which doesn't support HasField in proto3 unless optional.
        # Assuming it's a standard field, 0 is the default.
        # If the API treats 0 as a valid revision, we can just use it.
        # If 0 means "not set", we'd check for 0.
        # Here we assume it's always present or defaults to 0.
        revision=pb_obj.revision,
        update_time=update_time,
        resampling_method=(
            ResamplingMethod(pb_obj.resampling_method)
            if isinstance(pb_obj, pb.MarketLocationSampleDetail)
            # Fallback to UNSPECIFIED if it's a simple MarketLocationSample
            # which does not have this field.
            else ResamplingMethod.UNSPECIFIED
        ),
    )

to_protobuf

to_protobuf() -> MarketLocationSample

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationSample	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pb.MarketLocationSample:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    sample = pb.MarketLocationSample(
        sample_time=_datetime_to_timestamp(self.sample_time),
        value=self.value,
        quality=self.quality.value,
    )
    if self.revision is not None:
        sample.revision = self.revision
    return sample

frequenz.client.marketmetering.MarketLocationSeries dataclass

A time series for a single logical Market Location.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationSeries:
    """A time series for a single logical Market Location."""

    market_location_ref: MarketLocationRef
    """Reference to the Market Location."""

    direction: EnergyFlowDirection
    """Energy-flow direction of this series."""

    metric_type: MetricType
    """Metric type represented by this series."""

    metric_unit: MetricUnit
    """Physical unit in which the metric is expressed."""

    resolution: TimeResolution
    """Market Location's current contractual resolution."""

    samples: list[MarketLocationSample]
    """Ordered samples representing this time series."""

    @classmethod
    def from_protobuf(cls, pb_obj: pb.MarketLocationSeries) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new MarketLocationSeries instance.
        """
        return cls(
            market_location_ref=MarketLocationRef.from_protobuf(
                pb_obj.market_location_ref
            ),
            direction=EnergyFlowDirection(pb_obj.direction),
            metric_type=MetricType(pb_obj.metric_type),
            metric_unit=MetricUnit(pb_obj.metric_unit),
            resolution=TimeResolution(pb_obj.resolution),
            samples=[MarketLocationSample.from_protobuf(s) for s in pb_obj.samples],
        )

Attributes

direction instance-attribute

direction: EnergyFlowDirection

Energy-flow direction of this series.

market_location_ref instance-attribute

market_location_ref: MarketLocationRef

Reference to the Market Location.

metric_type instance-attribute

metric_type: MetricType

Metric type represented by this series.

metric_unit instance-attribute

metric_unit: MetricUnit

Physical unit in which the metric is expressed.

resolution instance-attribute

resolution: TimeResolution

Market Location's current contractual resolution.

samples instance-attribute

samples: list[MarketLocationSample]

Ordered samples representing this time series.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: MarketLocationSeries) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: MarketLocationSeries

RETURNS	DESCRIPTION
Self	A new MarketLocationSeries instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.MarketLocationSeries) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new MarketLocationSeries instance.
    """
    return cls(
        market_location_ref=MarketLocationRef.from_protobuf(
            pb_obj.market_location_ref
        ),
        direction=EnergyFlowDirection(pb_obj.direction),
        metric_type=MetricType(pb_obj.metric_type),
        metric_unit=MetricUnit(pb_obj.metric_unit),
        resolution=TimeResolution(pb_obj.resolution),
        samples=[MarketLocationSample.from_protobuf(s) for s in pb_obj.samples],
    )

frequenz.client.marketmetering.MarketLocationUpdate dataclass

Fields to update in a Market Location.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationUpdate:
    """Fields to update in a Market Location."""

    display_name: str | None = None
    """New display name."""

    supported_directions: list[EnergyFlowDirection] | None = None
    """New supported directions."""

    time_resolution: TimeResolution | None = None
    """New time resolution."""

    payload: dict[str, Any] | None = None
    """New payload (replaces existing payload)."""

    def to_protobuf(
        self,
    ) -> tuple[
        pb.UpdateMarketLocationRequest.MarketLocationUpdate, field_mask_pb2.FieldMask
    ]:
        """Convert to protobuf message and field mask.

        Returns:
            A tuple containing the update message and the field mask.
        """
        update_pb = pb.UpdateMarketLocationRequest.MarketLocationUpdate()
        paths = []

        if self.display_name is not None:
            update_pb.display_name = self.display_name
            paths.append("display_name")

        if self.supported_directions is not None:
            update_pb.supported_directions.extend(
                d.value for d in self.supported_directions
            )
            paths.append("supported_directions")

        if self.time_resolution is not None:
            update_pb.time_resolution = self.time_resolution.value
            paths.append("time_resolution")

        if self.payload is not None:
            pb_struct = struct_pb2.Struct()
            pb_struct.update(self.payload)
            update_pb.payload.CopyFrom(pb_struct)
            paths.append("payload")

        return update_pb, field_mask_pb2.FieldMask(paths=paths)

Attributes

display_name class-attribute instance-attribute

display_name: str | None = None

New display name.

payload class-attribute instance-attribute

payload: dict[str, Any] | None = None

New payload (replaces existing payload).

supported_directions class-attribute instance-attribute

supported_directions: list[EnergyFlowDirection] | None = (
    None
)

New supported directions.

time_resolution class-attribute instance-attribute

time_resolution: TimeResolution | None = None

New time resolution.

Methods:

to_protobuf

to_protobuf() -> tuple[MarketLocationUpdate, FieldMask]

Convert to protobuf message and field mask.

RETURNS	DESCRIPTION
tuple[MarketLocationUpdate, FieldMask]	A tuple containing the update message and the field mask.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(
    self,
) -> tuple[
    pb.UpdateMarketLocationRequest.MarketLocationUpdate, field_mask_pb2.FieldMask
]:
    """Convert to protobuf message and field mask.

    Returns:
        A tuple containing the update message and the field mask.
    """
    update_pb = pb.UpdateMarketLocationRequest.MarketLocationUpdate()
    paths = []

    if self.display_name is not None:
        update_pb.display_name = self.display_name
        paths.append("display_name")

    if self.supported_directions is not None:
        update_pb.supported_directions.extend(
            d.value for d in self.supported_directions
        )
        paths.append("supported_directions")

    if self.time_resolution is not None:
        update_pb.time_resolution = self.time_resolution.value
        paths.append("time_resolution")

    if self.payload is not None:
        pb_struct = struct_pb2.Struct()
        pb_struct.update(self.payload)
        update_pb.payload.CopyFrom(pb_struct)
        paths.append("payload")

    return update_pb, field_mask_pb2.FieldMask(paths=paths)

frequenz.client.marketmetering.MarketLocationsFilter dataclass

Filter criteria for listing Market Locations.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class MarketLocationsFilter:
    """Filter criteria for listing Market Locations."""

    market_location_id_filters: Iterable[MarketLocationId] = field(default_factory=list)
    """Filter by specific Market Location IDs."""

    activation_filter: ActivationFilter = ActivationFilter.ONLY_ACTIVE
    """Filter by activation status (defaults to only active locations)."""

    def to_protobuf(self) -> pb.MarketLocationsFilter:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        return pb.MarketLocationsFilter(
            market_location_id_values=[
                ml_id.to_id_value_protobuf()
                for ml_id in self.market_location_id_filters
            ],
            activation_filter=self.activation_filter.value,
        )

Attributes

activation_filter class-attribute instance-attribute

activation_filter: ActivationFilter = (
    ActivationFilter.ONLY_ACTIVE
)

Filter by activation status (defaults to only active locations).

market_location_id_filters class-attribute instance-attribute

market_location_id_filters: Iterable[MarketLocationId] = (
    field(default_factory=list)
)

Filter by specific Market Location IDs.

Methods:

to_protobuf

to_protobuf() -> MarketLocationsFilter

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationsFilter	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pb.MarketLocationsFilter:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    return pb.MarketLocationsFilter(
        market_location_id_values=[
            ml_id.to_id_value_protobuf()
            for ml_id in self.market_location_id_filters
        ],
        activation_filter=self.activation_filter.value,
    )

frequenz.client.marketmetering.MarketMeteringApiClient

Bases: BaseApiClient[MarketMeteringServiceStub]

Market Metering API client.

This client provides access to the Market Metering Service, allowing you to stream historical and real-time metering samples from Market Locations.

Example

from datetime import datetime, timezone
from frequenz.client.marketmetering import MarketMeteringApiClient
from frequenz.client.marketmetering.types import (
    EnergyFlowDirection,
    MarketArea,
    MarketLocationId,
    MarketLocationIdType,
    MarketLocationRef,
    MetricType,
)

client = MarketMeteringApiClient(
    server_url="grpc://marketmetering.example.com",
    auth_key="your-api-key",
    sign_secret="your-sign-secret",
)

market_location = MarketLocationRef(
    market_area=MarketArea.EU_DE,
    market_location_id=MarketLocationId(
        value="DE01234567890",
        type=MarketLocationIdType.MALO_ID,
    ),
)

async for series in client.stream_samples(
    market_locations=[market_location],
    directions=[EnergyFlowDirection.IMPORT],
    metric_types=[MetricType.ACTIVE_ENERGY],
    start_time=datetime(2025, 1, 1, tzinfo=timezone.utc),
):
    for sample in series.samples:
        print(f"{sample.sample_time}: {sample.value} {series.metric_unit.name}")

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

class MarketMeteringApiClient(
    BaseApiClient[marketmetering_pb2_grpc.MarketMeteringServiceStub]
):
    """Market Metering API client.

    This client provides access to the Market Metering Service, allowing you to
    stream historical and real-time metering samples from Market Locations.

    Example:
        ```python
        from datetime import datetime, timezone
        from frequenz.client.marketmetering import MarketMeteringApiClient
        from frequenz.client.marketmetering.types import (
            EnergyFlowDirection,
            MarketArea,
            MarketLocationId,
            MarketLocationIdType,
            MarketLocationRef,
            MetricType,
        )

        client = MarketMeteringApiClient(
            server_url="grpc://marketmetering.example.com",
            auth_key="your-api-key",
            sign_secret="your-sign-secret",
        )

        market_location = MarketLocationRef(
            market_area=MarketArea.EU_DE,
            market_location_id=MarketLocationId(
                value="DE01234567890",
                type=MarketLocationIdType.MALO_ID,
            ),
        )

        async for series in client.stream_samples(
            market_locations=[market_location],
            directions=[EnergyFlowDirection.IMPORT],
            metric_types=[MetricType.ACTIVE_ENERGY],
            start_time=datetime(2025, 1, 1, tzinfo=timezone.utc),
        ):
            for sample in series.samples:
                print(f"{sample.sample_time}: {sample.value} {series.metric_unit.name}")
        ```
    """

    # pylint: disable=too-many-arguments
    def __init__(
        self,
        *,
        server_url: str,
        auth_key: str,
        sign_secret: str | None = None,
        connect: bool = True,
        call_timeout: timedelta = timedelta(seconds=60),
        stream_timeout: timedelta = timedelta(minutes=5),
    ) -> None:
        """Initialize the client.

        Args:
            server_url: The URL of the server to connect to.
            auth_key: API key to use for authentication.
            sign_secret: Optional secret for signing requests.
            connect: Whether to connect to the service immediately.
            call_timeout: Timeout for gRPC calls, default is 60 seconds.
            stream_timeout: Timeout for gRPC streams, default is 5 minutes.
        """
        super().__init__(
            server_url,
            marketmetering_pb2_grpc.MarketMeteringServiceStub,
            connect=connect,
            channel_defaults=ChannelOptions(
                port=DEFAULT_PORT,
                ssl=SslOptions(enabled=True),
            ),
            auth_key=auth_key,
            sign_secret=sign_secret,
        )

        self._streams: dict[
            tuple[
                tuple[MarketLocationRef, ...],
                tuple[EnergyFlowDirection, ...],
                tuple[MetricType, ...],
            ],
            GrpcStreamBroadcaster[
                pb.ReceiveMarketLocationSamplesStreamResponse, MarketLocationSeries
            ],
        ] = {}

        self._call_timeout_seconds = call_timeout.total_seconds()
        self._stream_timeout_seconds = stream_timeout.total_seconds()

    @property
    def call_timeout(self) -> timedelta:
        """Get the call timeout."""
        return timedelta(seconds=self._call_timeout_seconds)

    @property
    def stream_timeout(self) -> timedelta:
        """Get the stream timeout."""
        return timedelta(seconds=self._stream_timeout_seconds)

    @property
    def stub(self) -> marketmetering_pb2_grpc.MarketMeteringServiceStub:
        """The stub for the service."""
        if self._channel is None or self._stub is None:
            raise ClientNotConnected(server_url=self.server_url, operation="stub")
        return self._stub

    def _metadata(self, method: str) -> tuple[tuple[str, str | bytes], ...] | None:
        """Build request metadata for RPCs not covered by client-base interceptors."""
        if self._auth_key is None:
            return None

        metadata: list[tuple[str, str | bytes]] = [("key", self._auth_key)]
        if self._sign_secret is None:
            return tuple(metadata)

        ts = str(int(time.time())).encode()
        nonce = urlsafe_b64encode(secrets.token_bytes(16))

        digest = hmac.new(self._sign_secret.encode(), digestmod="sha256")
        digest.update(self._auth_key.encode())
        digest.update(ts)
        digest.update(nonce)
        digest.update(method.encode())

        metadata.extend(
            [
                ("ts", ts),
                ("nonce", nonce),
                ("sig", urlsafe_b64encode(digest.digest()).rstrip(b"=")),
            ]
        )
        return tuple(metadata)

    async def create_market_location(
        self,
        *,
        market_location_ref: MarketLocationRef,
        market_location: MarketLocation,
    ) -> MarketLocationDetail:
        """Create a new Market Location.

        Args:
            market_location_ref: The reference ID for the new location.
            market_location: The configuration of the new location.

        Returns:
            The created Market Location with server-assigned metadata.
        """
        request = pb.CreateMarketLocationRequest(
            market_location=market_location_ref.to_protobuf(),
            market_location_metadata=market_location.to_protobuf(),
        )
        response = await self.stub.CreateMarketLocation(  # type: ignore[misc]
            request,
            timeout=self._call_timeout_seconds,
        )
        return MarketLocationDetail.from_protobuf(response.market_location_detail)

    async def update_market_location(
        self,
        *,
        market_location_ref: MarketLocationRef,
        update: MarketLocationUpdate,
        expected_revision: int,
    ) -> MarketLocationDetail:
        """Update an existing Market Location.

        Args:
            market_location_ref: The reference ID of the location to update.
            update: The fields to update.
            expected_revision: The revision the caller expects to be the
                latest. This prevents lost updates when multiple callers
                modify the same Market Location concurrently. Pass the
                revision from the most recent read of the location.

        Returns:
            The updated Market Location with server-assigned metadata.
        """
        update_pb, update_mask_pb = update.to_protobuf()
        request = pb.UpdateMarketLocationRequest(
            market_location=market_location_ref.to_protobuf(),
            expected_revision=expected_revision,
            update_fields=update_pb,
            update_mask=update_mask_pb,
        )
        response = await self.stub.UpdateMarketLocation(  # type: ignore[misc]
            request,
            timeout=self._call_timeout_seconds,
        )
        return MarketLocationDetail.from_protobuf(response.market_location_detail)

    async def activate_market_locations(
        self,
        *,
        market_location_refs: list[MarketLocationRef],
    ) -> list[MarketLocationOperationResult]:
        """Activate one or more Market Locations.

        Args:
            market_location_refs: References to the locations to activate.

        Returns:
            A list of operation results, one per requested location.
        """
        request = pb.ActivateMarketLocationRequest(
            market_locations=[ref.to_protobuf() for ref in market_location_refs],
        )
        response = await self.stub.ActivateMarketLocation(  # type: ignore[misc]
            request,
            timeout=self._call_timeout_seconds,
        )
        return [
            MarketLocationOperationResult.from_protobuf(r) for r in response.results
        ]

    async def deactivate_market_locations(
        self,
        *,
        market_location_refs: list[MarketLocationRef],
    ) -> list[MarketLocationOperationResult]:
        """Deactivate one or more Market Locations.

        Args:
            market_location_refs: References to the locations to deactivate.

        Returns:
            A list of operation results, one per requested location.
        """
        request = pb.DeactivateMarketLocationRequest(
            market_locations=[ref.to_protobuf() for ref in market_location_refs],
        )
        response = await self.stub.DeactivateMarketLocation(  # type: ignore[misc]
            request,
            timeout=self._call_timeout_seconds,
        )
        return [
            MarketLocationOperationResult.from_protobuf(r) for r in response.results
        ]

    async def list_market_locations(
        self,
        *,
        filters: MarketLocationsFilter | None = None,
        revision_selection: RevisionSelection | None = None,
        pagination_params: PaginationParams | None = None,
    ) -> tuple[list[MarketLocationEntry], PaginationParams | None]:
        """List Market Locations.

        Args:
            filters: Optional filters for the query.
            revision_selection: Optional revision selection criteria.
            pagination_params: Optional pagination parameters.

        Returns:
            A tuple containing a list of Market Location entries and optional
            pagination parameters for the next page.
        """
        request = pb.ListMarketLocationsRequest(
            filter=filters.to_protobuf() if filters else None,
            revision_selection=(
                revision_selection.to_protobuf() if revision_selection else None
            ),
            pagination_params=(
                pagination_params.to_protobuf() if pagination_params else None
            ),
        )
        response = await self.stub.ListMarketLocations(  # type: ignore[misc]
            request,
            timeout=self._call_timeout_seconds,
        )

        market_locations = [
            MarketLocationEntry.from_protobuf(ml) for ml in response.market_locations
        ]

        # An empty `next_page_token` signals the end of the result set
        # (AIP-158). Returning a PaginationParams with an empty token would
        # make the server reject the follow-up request as an invalid page
        # token, so treat "empty token" the same as "no pagination info".
        next_page_params = None
        if (
            response.HasField("pagination_info")
            and response.pagination_info.next_page_token
        ):
            next_page_params = PaginationParams(
                page_token=response.pagination_info.next_page_token
            )

        return market_locations, next_page_params

    async def upsert_samples(
        self,
        samples_stream: AsyncIterator[tuple[MarketLocationRef, MarketLocationSeries]],
    ) -> AsyncIterator[UpsertResult]:
        """Upsert a stream of metering samples.

        Args:
            samples_stream: An async iterator yielding (MarketLocationRef, MarketLocationSeries)
                tuples. Each series should contain exactly one sample.

        Yields:
            UpsertResult objects indicating success or failure for each sample.
        """

        async def request_generator() -> (
            AsyncIterator[pb.UpsertMarketLocationSamplesStreamRequest]
        ):
            async for ml_ref, series in samples_stream:
                for sample in series.samples:
                    yield pb.UpsertMarketLocationSamplesStreamRequest(
                        market_location=ml_ref.to_protobuf(),
                        direction=series.direction.value,
                        metric_type=series.metric_type.value,
                        metric_unit=series.metric_unit.value,
                        sample=sample.to_protobuf(),
                    )

        response_stream = cast(
            AsyncIterator[pb.UpsertMarketLocationSamplesStreamResponse],
            self.stub.UpsertMarketLocationSamplesStream(
                request_generator(),  # type: ignore[arg-type]
                metadata=self._metadata("UpsertMarketLocationSamplesStream"),
                timeout=self._stream_timeout_seconds,
            ),
        )

        async for response in response_stream:
            yield UpsertResult.from_protobuf(response)

    # pylint: disable=too-many-arguments
    async def stream_samples(
        self,
        *,
        market_locations: list[MarketLocationRef],
        directions: list[EnergyFlowDirection],
        metric_types: list[MetricType],
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        resampling: ResamplingOptions | None = None,
        revision_strategy: RevisionStrategy | None = None,
    ) -> AsyncIterator[MarketLocationSeries]:
        """Stream metering samples for Market Locations.

        Streams historical and/or real-time metering samples for one or more
        Market Locations. The stream produces one `MarketLocationSeries` per
        unique combination of Market Location, direction, and metric type.

        Args:
            market_locations: List of Market Location references to stream.
            directions: Energy-flow directions requested (IMPORT and/or EXPORT).
            metric_types: Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER).
            start_time: Optional start time for historical data.
                If omitted, stream starts from real-time data.
            end_time: Optional end time. If omitted, stream continues in real-time.
            resampling: Optional resampling options for aggregation.
            revision_strategy: Optional revision strategy for the stream filter.

        Yields:
            MarketLocationSeries objects containing samples for each combination
            of Market Location, direction, and metric type.

        Example:
            ```python
            async for series in client.stream_samples(
                market_locations=[market_location],
                directions=[EnergyFlowDirection.IMPORT],
                metric_types=[MetricType.ACTIVE_ENERGY],
            ):
                print(f"Location: {series.market_location_ref.market_location_id.value}")
                for sample in series.samples:
                    print(f"  {sample.sample_time}: {sample.value}")
            ```
        """
        # Build the request
        request = pb.ReceiveMarketLocationSamplesStreamRequest(
            market_locations=[ml.to_protobuf() for ml in market_locations],
            directions=[d.value for d in directions],
            metric_types=[mt.value for mt in metric_types],
        )

        # Add stream filter with time interval if specified
        stream_filter = pb.MarketLocationSamplesStreamFilter()

        if start_time or end_time:
            time_filter = pb.TimeFilter()
            interval = PBInterval()
            if start_time:
                interval.start_time.CopyFrom(_datetime_to_timestamp(start_time))
            if end_time:
                interval.end_time.CopyFrom(_datetime_to_timestamp(end_time))
            time_filter.interval.CopyFrom(interval)
            stream_filter.time_filter.CopyFrom(time_filter)

        if resampling:
            stream_filter.resampling_options.CopyFrom(resampling.to_protobuf())

        if revision_strategy:
            stream_filter.revision_strategy = revision_strategy.value

        request.stream_filter.CopyFrom(stream_filter)

        # Make the streaming call
        response_stream = cast(
            AsyncIterator[pb.ReceiveMarketLocationSamplesStreamResponse],
            self.stub.ReceiveMarketLocationSamplesStream(
                request,
                timeout=self._stream_timeout_seconds,
            ),
        )

        async for response in response_stream:
            for series_pb in response.series:
                yield MarketLocationSeries.from_protobuf(series_pb)

    # pylint: disable=too-many-arguments
    def stream(
        self,
        *,
        market_locations: list[MarketLocationRef],
        directions: list[EnergyFlowDirection],
        metric_types: list[MetricType],
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        resampling: ResamplingOptions | None = None,
        revision_strategy: RevisionStrategy | None = None,
    ) -> channels.Receiver[MarketLocationSeries]:
        """Get a receiver for streaming metering samples.

        This method returns a channel receiver that can be used to receive
        Market Location sample series. The stream is managed internally and
        supports multiple receivers.

        Args:
            market_locations: List of Market Location references to stream.
            directions: Energy-flow directions requested (IMPORT and/or EXPORT).
            metric_types: Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER).
            start_time: Optional start time for historical data.
            end_time: Optional end time. If omitted, stream continues in real-time.
            resampling: Optional resampling options for aggregation.
            revision_strategy: Optional revision strategy for the stream filter.

        Returns:
            A channel receiver for MarketLocationSeries objects.

        Example:
            ```python
            receiver = client.stream(
                market_locations=[market_location],
                directions=[EnergyFlowDirection.IMPORT],
                metric_types=[MetricType.ACTIVE_ENERGY],
            )
            async for series in receiver:
                print(f"Received series: {series}")
            ```
        """
        return self._get_stream(
            market_locations=market_locations,
            directions=directions,
            metric_types=metric_types,
            start_time=start_time,
            end_time=end_time,
            resampling=resampling,
            revision_strategy=revision_strategy,
        ).new_receiver()

    # pylint: disable=too-many-arguments
    def _get_stream(
        self,
        *,
        market_locations: list[MarketLocationRef],
        directions: list[EnergyFlowDirection],
        metric_types: list[MetricType],
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        resampling: ResamplingOptions | None = None,
        revision_strategy: RevisionStrategy | None = None,
    ) -> GrpcStreamBroadcaster[
        pb.ReceiveMarketLocationSamplesStreamResponse, MarketLocationSeries
    ]:
        """Get or create a streaming broadcaster for the given parameters."""
        # Create a key for caching the stream
        key = (
            tuple(market_locations),
            tuple(directions),
            tuple(metric_types),
        )

        broadcaster = self._streams.get(key)
        if broadcaster is not None and not broadcaster.is_running:
            del self._streams[key]
            broadcaster = None

        if broadcaster is None:
            # Build the request
            request = pb.ReceiveMarketLocationSamplesStreamRequest(
                market_locations=[ml.to_protobuf() for ml in market_locations],
                directions=[d.value for d in directions],
                metric_types=[mt.value for mt in metric_types],
            )

            # Add stream filter
            stream_filter = pb.MarketLocationSamplesStreamFilter()
            if start_time or end_time:
                time_filter = pb.TimeFilter()
                interval = PBInterval()
                if start_time:
                    interval.start_time.CopyFrom(_datetime_to_timestamp(start_time))
                if end_time:
                    interval.end_time.CopyFrom(_datetime_to_timestamp(end_time))
                time_filter.interval.CopyFrom(interval)
                stream_filter.time_filter.CopyFrom(time_filter)

            if resampling:
                stream_filter.resampling_options.CopyFrom(resampling.to_protobuf())

            if revision_strategy:
                stream_filter.revision_strategy = revision_strategy.value

            request.stream_filter.CopyFrom(stream_filter)

            def transform(
                response: pb.ReceiveMarketLocationSamplesStreamResponse,
            ) -> MarketLocationSeries:
                # Return the first series from the response
                # In practice, responses may have multiple series
                if response.series:
                    return MarketLocationSeries.from_protobuf(response.series[0])
                raise ValueError("Empty response received")

            broadcaster = GrpcStreamBroadcaster(
                stream_name="ReceiveMarketLocationSamplesStream",
                stream_method=lambda: cast(
                    AsyncIterator[pb.ReceiveMarketLocationSamplesStreamResponse],
                    self.stub.ReceiveMarketLocationSamplesStream(
                        request,
                        timeout=self._stream_timeout_seconds,
                    ),
                ),
                transform=transform,
                retry_strategy=LinearBackoff(interval=1, limit=None),
            )
            self._streams[key] = broadcaster

        return broadcaster

Attributes

call_timeout property

call_timeout: timedelta

Get the call timeout.

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.

stream_timeout property

stream_timeout: timedelta

Get the stream timeout.

stub property

stub: MarketMeteringServiceStub

The stub for the service.

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,
    sign_secret: str | None = None,
    connect: bool = True,
    call_timeout: timedelta = timedelta(seconds=60),
    stream_timeout: timedelta = timedelta(minutes=5)
) -> None

Initialize the client.

PARAMETER	DESCRIPTION
server_url	The URL of the server to connect to. TYPE: str
auth_key	API key to use for authentication. TYPE: str
sign_secret	Optional secret for signing requests. TYPE: str | None DEFAULT: None
connect	Whether to connect to the service immediately. TYPE: bool DEFAULT: True
call_timeout	Timeout for gRPC calls, default is 60 seconds. TYPE: timedelta DEFAULT: timedelta(seconds=60)
stream_timeout	Timeout for gRPC streams, default is 5 minutes. TYPE: timedelta DEFAULT: timedelta(minutes=5)

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

def __init__(
    self,
    *,
    server_url: str,
    auth_key: str,
    sign_secret: str | None = None,
    connect: bool = True,
    call_timeout: timedelta = timedelta(seconds=60),
    stream_timeout: timedelta = timedelta(minutes=5),
) -> None:
    """Initialize the client.

    Args:
        server_url: The URL of the server to connect to.
        auth_key: API key to use for authentication.
        sign_secret: Optional secret for signing requests.
        connect: Whether to connect to the service immediately.
        call_timeout: Timeout for gRPC calls, default is 60 seconds.
        stream_timeout: Timeout for gRPC streams, default is 5 minutes.
    """
    super().__init__(
        server_url,
        marketmetering_pb2_grpc.MarketMeteringServiceStub,
        connect=connect,
        channel_defaults=ChannelOptions(
            port=DEFAULT_PORT,
            ssl=SslOptions(enabled=True),
        ),
        auth_key=auth_key,
        sign_secret=sign_secret,
    )

    self._streams: dict[
        tuple[
            tuple[MarketLocationRef, ...],
            tuple[EnergyFlowDirection, ...],
            tuple[MetricType, ...],
        ],
        GrpcStreamBroadcaster[
            pb.ReceiveMarketLocationSamplesStreamResponse, MarketLocationSeries
        ],
    ] = {}

    self._call_timeout_seconds = call_timeout.total_seconds()
    self._stream_timeout_seconds = stream_timeout.total_seconds()

activate_market_locations async

activate_market_locations(
    *, market_location_refs: list[MarketLocationRef]
) -> list[MarketLocationOperationResult]

Activate one or more Market Locations.

PARAMETER	DESCRIPTION
market_location_refs	References to the locations to activate. TYPE: list[MarketLocationRef]

RETURNS	DESCRIPTION
list[MarketLocationOperationResult]	A list of operation results, one per requested location.

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

async def activate_market_locations(
    self,
    *,
    market_location_refs: list[MarketLocationRef],
) -> list[MarketLocationOperationResult]:
    """Activate one or more Market Locations.

    Args:
        market_location_refs: References to the locations to activate.

    Returns:
        A list of operation results, one per requested location.
    """
    request = pb.ActivateMarketLocationRequest(
        market_locations=[ref.to_protobuf() for ref in market_location_refs],
    )
    response = await self.stub.ActivateMarketLocation(  # type: ignore[misc]
        request,
        timeout=self._call_timeout_seconds,
    )
    return [
        MarketLocationOperationResult.from_protobuf(r) for r in response.results
    ]

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)

create_market_location async

create_market_location(
    *,
    market_location_ref: MarketLocationRef,
    market_location: MarketLocation
) -> MarketLocationDetail

Create a new Market Location.

PARAMETER	DESCRIPTION
market_location_ref	The reference ID for the new location. TYPE: MarketLocationRef
market_location	The configuration of the new location. TYPE: MarketLocation

RETURNS	DESCRIPTION
MarketLocationDetail	The created Market Location with server-assigned metadata.

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

async def create_market_location(
    self,
    *,
    market_location_ref: MarketLocationRef,
    market_location: MarketLocation,
) -> MarketLocationDetail:
    """Create a new Market Location.

    Args:
        market_location_ref: The reference ID for the new location.
        market_location: The configuration of the new location.

    Returns:
        The created Market Location with server-assigned metadata.
    """
    request = pb.CreateMarketLocationRequest(
        market_location=market_location_ref.to_protobuf(),
        market_location_metadata=market_location.to_protobuf(),
    )
    response = await self.stub.CreateMarketLocation(  # type: ignore[misc]
        request,
        timeout=self._call_timeout_seconds,
    )
    return MarketLocationDetail.from_protobuf(response.market_location_detail)

deactivate_market_locations async

deactivate_market_locations(
    *, market_location_refs: list[MarketLocationRef]
) -> list[MarketLocationOperationResult]

Deactivate one or more Market Locations.

PARAMETER	DESCRIPTION
market_location_refs	References to the locations to deactivate. TYPE: list[MarketLocationRef]

RETURNS	DESCRIPTION
list[MarketLocationOperationResult]	A list of operation results, one per requested location.

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

async def deactivate_market_locations(
    self,
    *,
    market_location_refs: list[MarketLocationRef],
) -> list[MarketLocationOperationResult]:
    """Deactivate one or more Market Locations.

    Args:
        market_location_refs: References to the locations to deactivate.

    Returns:
        A list of operation results, one per requested location.
    """
    request = pb.DeactivateMarketLocationRequest(
        market_locations=[ref.to_protobuf() for ref in market_location_refs],
    )
    response = await self.stub.DeactivateMarketLocation(  # type: ignore[misc]
        request,
        timeout=self._call_timeout_seconds,
    )
    return [
        MarketLocationOperationResult.from_protobuf(r) for r in response.results
    ]

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)

list_market_locations async

list_market_locations(
    *,
    filters: MarketLocationsFilter | None = None,
    revision_selection: RevisionSelection | None = None,
    pagination_params: PaginationParams | None = None
) -> tuple[
    list[MarketLocationEntry], PaginationParams | None
]

List Market Locations.

PARAMETER	DESCRIPTION
filters	Optional filters for the query. TYPE: MarketLocationsFilter | None DEFAULT: None
revision_selection	Optional revision selection criteria. TYPE: RevisionSelection | None DEFAULT: None
pagination_params	Optional pagination parameters. TYPE: PaginationParams | None DEFAULT: None

RETURNS	DESCRIPTION
list[MarketLocationEntry]	A tuple containing a list of Market Location entries and optional
PaginationParams | None	pagination parameters for the next page.

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

async def list_market_locations(
    self,
    *,
    filters: MarketLocationsFilter | None = None,
    revision_selection: RevisionSelection | None = None,
    pagination_params: PaginationParams | None = None,
) -> tuple[list[MarketLocationEntry], PaginationParams | None]:
    """List Market Locations.

    Args:
        filters: Optional filters for the query.
        revision_selection: Optional revision selection criteria.
        pagination_params: Optional pagination parameters.

    Returns:
        A tuple containing a list of Market Location entries and optional
        pagination parameters for the next page.
    """
    request = pb.ListMarketLocationsRequest(
        filter=filters.to_protobuf() if filters else None,
        revision_selection=(
            revision_selection.to_protobuf() if revision_selection else None
        ),
        pagination_params=(
            pagination_params.to_protobuf() if pagination_params else None
        ),
    )
    response = await self.stub.ListMarketLocations(  # type: ignore[misc]
        request,
        timeout=self._call_timeout_seconds,
    )

    market_locations = [
        MarketLocationEntry.from_protobuf(ml) for ml in response.market_locations
    ]

    # An empty `next_page_token` signals the end of the result set
    # (AIP-158). Returning a PaginationParams with an empty token would
    # make the server reject the follow-up request as an invalid page
    # token, so treat "empty token" the same as "no pagination info".
    next_page_params = None
    if (
        response.HasField("pagination_info")
        and response.pagination_info.next_page_token
    ):
        next_page_params = PaginationParams(
            page_token=response.pagination_info.next_page_token
        )

    return market_locations, next_page_params

stream

stream(
    *,
    market_locations: list[MarketLocationRef],
    directions: list[EnergyFlowDirection],
    metric_types: list[MetricType],
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    resampling: ResamplingOptions | None = None,
    revision_strategy: RevisionStrategy | None = None
) -> Receiver[MarketLocationSeries]

Get a receiver for streaming metering samples.

This method returns a channel receiver that can be used to receive Market Location sample series. The stream is managed internally and supports multiple receivers.

PARAMETER	DESCRIPTION
market_locations	List of Market Location references to stream. TYPE: list[MarketLocationRef]
directions	Energy-flow directions requested (IMPORT and/or EXPORT). TYPE: list[EnergyFlowDirection]
metric_types	Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER). TYPE: list[MetricType]
start_time	Optional start time for historical data. TYPE: datetime | None DEFAULT: None
end_time	Optional end time. If omitted, stream continues in real-time. TYPE: datetime | None DEFAULT: None
resampling	Optional resampling options for aggregation. TYPE: ResamplingOptions | None DEFAULT: None
revision_strategy	Optional revision strategy for the stream filter. TYPE: RevisionStrategy | None DEFAULT: None

RETURNS	DESCRIPTION
Receiver[MarketLocationSeries]	A channel receiver for MarketLocationSeries objects.

Example

receiver = client.stream(
    market_locations=[market_location],
    directions=[EnergyFlowDirection.IMPORT],
    metric_types=[MetricType.ACTIVE_ENERGY],
)
async for series in receiver:
    print(f"Received series: {series}")

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

def stream(
    self,
    *,
    market_locations: list[MarketLocationRef],
    directions: list[EnergyFlowDirection],
    metric_types: list[MetricType],
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    resampling: ResamplingOptions | None = None,
    revision_strategy: RevisionStrategy | None = None,
) -> channels.Receiver[MarketLocationSeries]:
    """Get a receiver for streaming metering samples.

    This method returns a channel receiver that can be used to receive
    Market Location sample series. The stream is managed internally and
    supports multiple receivers.

    Args:
        market_locations: List of Market Location references to stream.
        directions: Energy-flow directions requested (IMPORT and/or EXPORT).
        metric_types: Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER).
        start_time: Optional start time for historical data.
        end_time: Optional end time. If omitted, stream continues in real-time.
        resampling: Optional resampling options for aggregation.
        revision_strategy: Optional revision strategy for the stream filter.

    Returns:
        A channel receiver for MarketLocationSeries objects.

    Example:
        ```python
        receiver = client.stream(
            market_locations=[market_location],
            directions=[EnergyFlowDirection.IMPORT],
            metric_types=[MetricType.ACTIVE_ENERGY],
        )
        async for series in receiver:
            print(f"Received series: {series}")
        ```
    """
    return self._get_stream(
        market_locations=market_locations,
        directions=directions,
        metric_types=metric_types,
        start_time=start_time,
        end_time=end_time,
        resampling=resampling,
        revision_strategy=revision_strategy,
    ).new_receiver()

stream_samples async

stream_samples(
    *,
    market_locations: list[MarketLocationRef],
    directions: list[EnergyFlowDirection],
    metric_types: list[MetricType],
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    resampling: ResamplingOptions | None = None,
    revision_strategy: RevisionStrategy | None = None
) -> AsyncIterator[MarketLocationSeries]

Stream metering samples for Market Locations.

Streams historical and/or real-time metering samples for one or more Market Locations. The stream produces one MarketLocationSeries per unique combination of Market Location, direction, and metric type.

PARAMETER	DESCRIPTION
market_locations	List of Market Location references to stream. TYPE: list[MarketLocationRef]
directions	Energy-flow directions requested (IMPORT and/or EXPORT). TYPE: list[EnergyFlowDirection]
metric_types	Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER). TYPE: list[MetricType]
start_time	Optional start time for historical data. If omitted, stream starts from real-time data. TYPE: datetime | None DEFAULT: None
end_time	Optional end time. If omitted, stream continues in real-time. TYPE: datetime | None DEFAULT: None
resampling	Optional resampling options for aggregation. TYPE: ResamplingOptions | None DEFAULT: None
revision_strategy	Optional revision strategy for the stream filter. TYPE: RevisionStrategy | None DEFAULT: None

YIELDS	DESCRIPTION
AsyncIterator[MarketLocationSeries]	MarketLocationSeries objects containing samples for each combination
AsyncIterator[MarketLocationSeries]	of Market Location, direction, and metric type.

Example

async for series in client.stream_samples(
    market_locations=[market_location],
    directions=[EnergyFlowDirection.IMPORT],
    metric_types=[MetricType.ACTIVE_ENERGY],
):
    print(f"Location: {series.market_location_ref.market_location_id.value}")
    for sample in series.samples:
        print(f"  {sample.sample_time}: {sample.value}")

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

async def stream_samples(
    self,
    *,
    market_locations: list[MarketLocationRef],
    directions: list[EnergyFlowDirection],
    metric_types: list[MetricType],
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    resampling: ResamplingOptions | None = None,
    revision_strategy: RevisionStrategy | None = None,
) -> AsyncIterator[MarketLocationSeries]:
    """Stream metering samples for Market Locations.

    Streams historical and/or real-time metering samples for one or more
    Market Locations. The stream produces one `MarketLocationSeries` per
    unique combination of Market Location, direction, and metric type.

    Args:
        market_locations: List of Market Location references to stream.
        directions: Energy-flow directions requested (IMPORT and/or EXPORT).
        metric_types: Metric types to request (e.g., ACTIVE_ENERGY, ACTIVE_POWER).
        start_time: Optional start time for historical data.
            If omitted, stream starts from real-time data.
        end_time: Optional end time. If omitted, stream continues in real-time.
        resampling: Optional resampling options for aggregation.
        revision_strategy: Optional revision strategy for the stream filter.

    Yields:
        MarketLocationSeries objects containing samples for each combination
        of Market Location, direction, and metric type.

    Example:
        ```python
        async for series in client.stream_samples(
            market_locations=[market_location],
            directions=[EnergyFlowDirection.IMPORT],
            metric_types=[MetricType.ACTIVE_ENERGY],
        ):
            print(f"Location: {series.market_location_ref.market_location_id.value}")
            for sample in series.samples:
                print(f"  {sample.sample_time}: {sample.value}")
        ```
    """
    # Build the request
    request = pb.ReceiveMarketLocationSamplesStreamRequest(
        market_locations=[ml.to_protobuf() for ml in market_locations],
        directions=[d.value for d in directions],
        metric_types=[mt.value for mt in metric_types],
    )

    # Add stream filter with time interval if specified
    stream_filter = pb.MarketLocationSamplesStreamFilter()

    if start_time or end_time:
        time_filter = pb.TimeFilter()
        interval = PBInterval()
        if start_time:
            interval.start_time.CopyFrom(_datetime_to_timestamp(start_time))
        if end_time:
            interval.end_time.CopyFrom(_datetime_to_timestamp(end_time))
        time_filter.interval.CopyFrom(interval)
        stream_filter.time_filter.CopyFrom(time_filter)

    if resampling:
        stream_filter.resampling_options.CopyFrom(resampling.to_protobuf())

    if revision_strategy:
        stream_filter.revision_strategy = revision_strategy.value

    request.stream_filter.CopyFrom(stream_filter)

    # Make the streaming call
    response_stream = cast(
        AsyncIterator[pb.ReceiveMarketLocationSamplesStreamResponse],
        self.stub.ReceiveMarketLocationSamplesStream(
            request,
            timeout=self._stream_timeout_seconds,
        ),
    )

    async for response in response_stream:
        for series_pb in response.series:
            yield MarketLocationSeries.from_protobuf(series_pb)

update_market_location async

update_market_location(
    *,
    market_location_ref: MarketLocationRef,
    update: MarketLocationUpdate,
    expected_revision: int
) -> MarketLocationDetail

Update an existing Market Location.

PARAMETER	DESCRIPTION
market_location_ref	The reference ID of the location to update. TYPE: MarketLocationRef
update	The fields to update. TYPE: MarketLocationUpdate
expected_revision	The revision the caller expects to be the latest. This prevents lost updates when multiple callers modify the same Market Location concurrently. Pass the revision from the most recent read of the location. TYPE: int

RETURNS	DESCRIPTION
MarketLocationDetail	The updated Market Location with server-assigned metadata.

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

async def update_market_location(
    self,
    *,
    market_location_ref: MarketLocationRef,
    update: MarketLocationUpdate,
    expected_revision: int,
) -> MarketLocationDetail:
    """Update an existing Market Location.

    Args:
        market_location_ref: The reference ID of the location to update.
        update: The fields to update.
        expected_revision: The revision the caller expects to be the
            latest. This prevents lost updates when multiple callers
            modify the same Market Location concurrently. Pass the
            revision from the most recent read of the location.

    Returns:
        The updated Market Location with server-assigned metadata.
    """
    update_pb, update_mask_pb = update.to_protobuf()
    request = pb.UpdateMarketLocationRequest(
        market_location=market_location_ref.to_protobuf(),
        expected_revision=expected_revision,
        update_fields=update_pb,
        update_mask=update_mask_pb,
    )
    response = await self.stub.UpdateMarketLocation(  # type: ignore[misc]
        request,
        timeout=self._call_timeout_seconds,
    )
    return MarketLocationDetail.from_protobuf(response.market_location_detail)

upsert_samples async

upsert_samples(
    samples_stream: AsyncIterator[
        tuple[MarketLocationRef, MarketLocationSeries]
    ],
) -> AsyncIterator[UpsertResult]

Upsert a stream of metering samples.

PARAMETER	DESCRIPTION
samples_stream	An async iterator yielding (MarketLocationRef, MarketLocationSeries) tuples. Each series should contain exactly one sample. TYPE: AsyncIterator[tuple[MarketLocationRef, MarketLocationSeries]]

YIELDS	DESCRIPTION
AsyncIterator[UpsertResult]	UpsertResult objects indicating success or failure for each sample.

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

async def upsert_samples(
    self,
    samples_stream: AsyncIterator[tuple[MarketLocationRef, MarketLocationSeries]],
) -> AsyncIterator[UpsertResult]:
    """Upsert a stream of metering samples.

    Args:
        samples_stream: An async iterator yielding (MarketLocationRef, MarketLocationSeries)
            tuples. Each series should contain exactly one sample.

    Yields:
        UpsertResult objects indicating success or failure for each sample.
    """

    async def request_generator() -> (
        AsyncIterator[pb.UpsertMarketLocationSamplesStreamRequest]
    ):
        async for ml_ref, series in samples_stream:
            for sample in series.samples:
                yield pb.UpsertMarketLocationSamplesStreamRequest(
                    market_location=ml_ref.to_protobuf(),
                    direction=series.direction.value,
                    metric_type=series.metric_type.value,
                    metric_unit=series.metric_unit.value,
                    sample=sample.to_protobuf(),
                )

    response_stream = cast(
        AsyncIterator[pb.UpsertMarketLocationSamplesStreamResponse],
        self.stub.UpsertMarketLocationSamplesStream(
            request_generator(),  # type: ignore[arg-type]
            metadata=self._metadata("UpsertMarketLocationSamplesStream"),
            timeout=self._stream_timeout_seconds,
        ),
    )

    async for response in response_stream:
        yield UpsertResult.from_protobuf(response)

frequenz.client.marketmetering.MetricType

Bases: Enum

Fundamental physical property being measured.

Source code in src/frequenz/client/marketmetering/types.py

class MetricType(Enum):
    """Fundamental physical property being measured."""

    UNSPECIFIED = pb.METRIC_TYPE_UNSPECIFIED
    """Unspecified metric type."""

    ACTIVE_ENERGY = pb.METRIC_TYPE_ACTIVE_ENERGY
    """Active energy (e.g., kWh, Wh, MWh)."""

    ACTIVE_POWER = pb.METRIC_TYPE_ACTIVE_POWER
    """Active power (e.g., kW, MW)."""

    REACTIVE_ENERGY = pb.METRIC_TYPE_REACTIVE_ENERGY
    """Reactive energy (e.g., kVArh)."""

    REACTIVE_POWER = pb.METRIC_TYPE_REACTIVE_POWER
    """Reactive power (e.g., kVAr, MVAr)."""

Attributes

ACTIVE_ENERGY class-attribute instance-attribute

ACTIVE_ENERGY = pb.METRIC_TYPE_ACTIVE_ENERGY

Active energy (e.g., kWh, Wh, MWh).

ACTIVE_POWER class-attribute instance-attribute

ACTIVE_POWER = pb.METRIC_TYPE_ACTIVE_POWER

Active power (e.g., kW, MW).

REACTIVE_ENERGY class-attribute instance-attribute

REACTIVE_ENERGY = pb.METRIC_TYPE_REACTIVE_ENERGY

Reactive energy (e.g., kVArh).

REACTIVE_POWER class-attribute instance-attribute

REACTIVE_POWER = pb.METRIC_TYPE_REACTIVE_POWER

Reactive power (e.g., kVAr, MVAr).

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.METRIC_TYPE_UNSPECIFIED

Unspecified metric type.

frequenz.client.marketmetering.MetricUnit

Bases: Enum

Unit of measurement for a sample value.

Source code in src/frequenz/client/marketmetering/types.py

class MetricUnit(Enum):
    """Unit of measurement for a sample value."""

    UNSPECIFIED = pb.METRIC_UNIT_UNSPECIFIED
    """Unspecified unit."""

    # Energy units
    WH = pb.METRIC_UNIT_WH
    """Watt-hour (Wh)."""

    KWH = pb.METRIC_UNIT_KWH
    """Kilowatt-hour (kWh)."""

    MWH = pb.METRIC_UNIT_MWH
    """Megawatt-hour (MWh)."""

    # Power units
    W = pb.METRIC_UNIT_W
    """Watt (W)."""

    KW = pb.METRIC_UNIT_KW
    """Kilowatt (kW)."""

    MW = pb.METRIC_UNIT_MW
    """Megawatt (MW)."""

    # Reactive energy units
    VARH = pb.METRIC_UNIT_VARH
    """Volt-ampere-reactive-hour (VArh)."""

    KVARH = pb.METRIC_UNIT_KVARH
    """Kilovolt-ampere-reactive-hour (kVArh)."""

    MVARH = pb.METRIC_UNIT_MVARH
    """Megavolt-ampere-reactive-hour (MVArh)."""

    # Reactive power units
    VAR = pb.METRIC_UNIT_VAR
    """Volt-ampere-reactive (VAr)."""

    KVAR = pb.METRIC_UNIT_KVAR
    """Kilovolt-ampere-reactive (kVAr)."""

    MVAR = pb.METRIC_UNIT_MVAR
    """Megavolt-ampere-reactive (MVAr)."""

Attributes

KVAR class-attribute instance-attribute

KVAR = pb.METRIC_UNIT_KVAR

Kilovolt-ampere-reactive (kVAr).

KVARH class-attribute instance-attribute

KVARH = pb.METRIC_UNIT_KVARH

Kilovolt-ampere-reactive-hour (kVArh).

KW class-attribute instance-attribute

KW = pb.METRIC_UNIT_KW

Kilowatt (kW).

KWH class-attribute instance-attribute

KWH = pb.METRIC_UNIT_KWH

Kilowatt-hour (kWh).

MVAR class-attribute instance-attribute

MVAR = pb.METRIC_UNIT_MVAR

Megavolt-ampere-reactive (MVAr).

MVARH class-attribute instance-attribute

MVARH = pb.METRIC_UNIT_MVARH

Megavolt-ampere-reactive-hour (MVArh).

MW class-attribute instance-attribute

MW = pb.METRIC_UNIT_MW

Megawatt (MW).

MWH class-attribute instance-attribute

MWH = pb.METRIC_UNIT_MWH

Megawatt-hour (MWh).

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.METRIC_UNIT_UNSPECIFIED

Unspecified unit.

VAR class-attribute instance-attribute

VAR = pb.METRIC_UNIT_VAR

Volt-ampere-reactive (VAr).

VARH class-attribute instance-attribute

VARH = pb.METRIC_UNIT_VARH

Volt-ampere-reactive-hour (VArh).

W class-attribute instance-attribute

W = pb.METRIC_UNIT_W

Watt (W).

WH class-attribute instance-attribute

WH = pb.METRIC_UNIT_WH

Watt-hour (Wh).

frequenz.client.marketmetering.PaginationParams dataclass

Parameters for pagination.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class PaginationParams:
    """Parameters for pagination."""

    page_size: int | None = None
    """Maximum number of results to return."""

    page_token: str | None = None
    """Token to retrieve the next page of results."""

    def to_protobuf(self) -> pagination_params_pb.PaginationParams:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        # Note: We need to check if values are None, because protobuf fields
        # usually default to 0/empty if not set, but explicit checking is safer.
        params = pagination_params_pb.PaginationParams()
        if self.page_size is not None:
            params.page_size = self.page_size
        if self.page_token is not None:
            params.page_token = self.page_token
        return params

Attributes

page_size class-attribute instance-attribute

page_size: int | None = None

Maximum number of results to return.

page_token class-attribute instance-attribute

page_token: str | None = None

Token to retrieve the next page of results.

Methods:

to_protobuf

to_protobuf() -> PaginationParams

Convert to protobuf message.

RETURNS	DESCRIPTION
PaginationParams	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pagination_params_pb.PaginationParams:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    # Note: We need to check if values are None, because protobuf fields
    # usually default to 0/empty if not set, but explicit checking is safer.
    params = pagination_params_pb.PaginationParams()
    if self.page_size is not None:
        params.page_size = self.page_size
    if self.page_token is not None:
        params.page_token = self.page_token
    return params

frequenz.client.marketmetering.ResamplingMethod

Bases: Enum

Indicates how a returned sample was derived.

Source code in src/frequenz/client/marketmetering/types.py

class ResamplingMethod(Enum):
    """Indicates how a returned sample was derived."""

    UNSPECIFIED = pb.RESAMPLING_METHOD_UNSPECIFIED
    """Unspecified method."""

    NATIVE = pb.RESAMPLING_METHOD_NATIVE
    """Sample is stored at this exact resolution."""

    DOWNSAMPLED = pb.RESAMPLING_METHOD_DOWNSAMPLED
    """Sample is computed from finer-granularity data."""

    UPSAMPLED = pb.RESAMPLING_METHOD_UPSAMPLED
    """Sample is computed from coarser-granularity data."""

Attributes

DOWNSAMPLED class-attribute instance-attribute

DOWNSAMPLED = pb.RESAMPLING_METHOD_DOWNSAMPLED

Sample is computed from finer-granularity data.

NATIVE class-attribute instance-attribute

NATIVE = pb.RESAMPLING_METHOD_NATIVE

Sample is stored at this exact resolution.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.RESAMPLING_METHOD_UNSPECIFIED

Unspecified method.

UPSAMPLED class-attribute instance-attribute

UPSAMPLED = pb.RESAMPLING_METHOD_UPSAMPLED

Sample is computed from coarser-granularity data.

frequenz.client.marketmetering.ResamplingOptions dataclass

Resampling options for Market Location time-series data.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class ResamplingOptions:
    """Resampling options for Market Location time-series data."""

    resolution: TimeResolution | None = None
    """Optional resampling resolution for the output time series."""

    downsampling_method: DownsamplingMethod = DownsamplingMethod.MEAN
    """Aggregation method applied when downsampling."""

    def to_protobuf(self) -> pb.ResamplingOptions:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        return pb.ResamplingOptions(
            resolution=self.resolution.value if self.resolution else 0,  # type: ignore[arg-type]
            downsampling_method=self.downsampling_method.value,
        )

Attributes

downsampling_method class-attribute instance-attribute

downsampling_method: DownsamplingMethod = (
    DownsamplingMethod.MEAN
)

Aggregation method applied when downsampling.

resolution class-attribute instance-attribute

resolution: TimeResolution | None = None

Optional resampling resolution for the output time series.

Methods:

to_protobuf

to_protobuf() -> ResamplingOptions

Convert to protobuf message.

RETURNS	DESCRIPTION
ResamplingOptions	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pb.ResamplingOptions:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    return pb.ResamplingOptions(
        resolution=self.resolution.value if self.resolution else 0,  # type: ignore[arg-type]
        downsampling_method=self.downsampling_method.value,
    )

frequenz.client.marketmetering.RevisionSelection dataclass

Selection criteria for revisions of Market Location data.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class RevisionSelection:
    """Selection criteria for revisions of Market Location data."""

    revision_strategy: RevisionStrategy = RevisionStrategy.LATEST_ONLY
    """Strategy for selecting revisions."""

    changed_fields: Iterable[MarketLocationChangedField] = field(default_factory=list)
    """Filter revisions by changed fields (only for REVISION_STRATEGY_ALL)."""

    def to_protobuf(self) -> pb.MarketLocationRevisionSelection:
        """Convert to protobuf message.

        Returns:
            The protobuf representation.
        """
        return pb.MarketLocationRevisionSelection(
            revision_strategy=self.revision_strategy.value,
            changed_fields=[f.value for f in self.changed_fields],
        )

Attributes

changed_fields class-attribute instance-attribute

changed_fields: Iterable[MarketLocationChangedField] = (
    field(default_factory=list)
)

Filter revisions by changed fields (only for REVISION_STRATEGY_ALL).

revision_strategy class-attribute instance-attribute

revision_strategy: RevisionStrategy = (
    RevisionStrategy.LATEST_ONLY
)

Strategy for selecting revisions.

Methods:

to_protobuf

to_protobuf() -> MarketLocationRevisionSelection

Convert to protobuf message.

RETURNS	DESCRIPTION
MarketLocationRevisionSelection	The protobuf representation.

Source code in src/frequenz/client/marketmetering/types.py

def to_protobuf(self) -> pb.MarketLocationRevisionSelection:
    """Convert to protobuf message.

    Returns:
        The protobuf representation.
    """
    return pb.MarketLocationRevisionSelection(
        revision_strategy=self.revision_strategy.value,
        changed_fields=[f.value for f in self.changed_fields],
    )

frequenz.client.marketmetering.RevisionStrategy

Bases: Enum

Strategy for selecting revisions of Market Location data.

Source code in src/frequenz/client/marketmetering/types.py

class RevisionStrategy(Enum):
    """Strategy for selecting revisions of Market Location data."""

    UNSPECIFIED = pb.REVISION_STRATEGY_UNSPECIFIED
    """Unspecified strategy."""

    LATEST_ONLY = pb.REVISION_STRATEGY_LATEST_ONLY
    """Return only the latest revision of the data."""

    ALL = pb.REVISION_STRATEGY_ALL
    """Return all revisions of the data."""

Attributes

ALL class-attribute instance-attribute

ALL = pb.REVISION_STRATEGY_ALL

Return all revisions of the data.

LATEST_ONLY class-attribute instance-attribute

LATEST_ONLY = pb.REVISION_STRATEGY_LATEST_ONLY

Return only the latest revision of the data.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.REVISION_STRATEGY_UNSPECIFIED

Unspecified strategy.

frequenz.client.marketmetering.SampleUpsertError dataclass

Structured error for a rejected sample upsert.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class SampleUpsertError:
    """Structured error for a rejected sample upsert."""

    error_code: SampleUpsertErrorCode
    """Machine-readable classification of the failure."""

    error_message: str
    """Human-readable diagnostic message for logging and debugging."""

    existing_sample: MarketLocationSample | None
    """The stored sample that prevented this upsert, if available.

    Populated only for REVISION_CONFLICT and REVISION_TOO_OLD errors.
    """

    @classmethod
    def from_protobuf(cls, pb_obj: pb.SampleUpsertError) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new SampleUpsertError instance.
        """
        existing_sample = None
        if pb_obj.HasField("existing_sample"):
            existing_sample = MarketLocationSample.from_protobuf(pb_obj.existing_sample)
        return cls(
            error_code=SampleUpsertErrorCode(pb_obj.error_code),
            error_message=pb_obj.error_message,
            existing_sample=existing_sample,
        )

Attributes

error_code instance-attribute

error_code: SampleUpsertErrorCode

Machine-readable classification of the failure.

error_message instance-attribute

error_message: str

Human-readable diagnostic message for logging and debugging.

existing_sample instance-attribute

existing_sample: MarketLocationSample | None

The stored sample that prevented this upsert, if available.

Populated only for REVISION_CONFLICT and REVISION_TOO_OLD errors.

Methods:

from_protobuf classmethod

from_protobuf(pb_obj: SampleUpsertError) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: SampleUpsertError

RETURNS	DESCRIPTION
Self	A new SampleUpsertError instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(cls, pb_obj: pb.SampleUpsertError) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new SampleUpsertError instance.
    """
    existing_sample = None
    if pb_obj.HasField("existing_sample"):
        existing_sample = MarketLocationSample.from_protobuf(pb_obj.existing_sample)
    return cls(
        error_code=SampleUpsertErrorCode(pb_obj.error_code),
        error_message=pb_obj.error_message,
        existing_sample=existing_sample,
    )

frequenz.client.marketmetering.SampleUpsertErrorCode

Bases: Enum

Error codes for sample upsert operations.

Source code in src/frequenz/client/marketmetering/types.py

class SampleUpsertErrorCode(Enum):
    """Error codes for sample upsert operations."""

    UNSPECIFIED = pb.SAMPLE_UPSERT_ERROR_CODE_UNSPECIFIED
    """Unspecified error."""

    REVISION_CONFLICT = pb.SAMPLE_UPSERT_ERROR_CODE_REVISION_CONFLICT
    """The sample revision conflicts with an existing sample."""

    REVISION_TOO_OLD = pb.SAMPLE_UPSERT_ERROR_CODE_REVISION_TOO_OLD
    """The sample revision is older than the current revision."""

    UNSUPPORTED_DIRECTION = pb.SAMPLE_UPSERT_ERROR_CODE_UNSUPPORTED_DIRECTION
    """The direction is not supported by the Market Location."""

    UNSUPPORTED_METRIC_TYPE = pb.SAMPLE_UPSERT_ERROR_CODE_UNSUPPORTED_METRIC_TYPE
    """The metric type is not supported by the Market Location."""

    UNIT_MISMATCH = pb.SAMPLE_UPSERT_ERROR_CODE_UNIT_MISMATCH
    """The metric unit does not match the Market Location's unit."""

    MARKET_LOCATION_NOT_FOUND = pb.SAMPLE_UPSERT_ERROR_CODE_MARKET_LOCATION_NOT_FOUND
    """The Market Location does not exist."""

    MARKET_LOCATION_INACTIVE = pb.SAMPLE_UPSERT_ERROR_CODE_MARKET_LOCATION_INACTIVE
    """The Market Location is inactive."""

    INVALID_TIMESTAMP = pb.SAMPLE_UPSERT_ERROR_CODE_INVALID_TIMESTAMP
    """The sample timestamp is invalid (e.g., misalignment)."""

    STORAGE_FAILURE = pb.SAMPLE_UPSERT_ERROR_CODE_STORAGE_FAILURE
    """Internal storage error."""

    UNKNOWN_ERROR = pb.SAMPLE_UPSERT_ERROR_CODE_UNKNOWN_ERROR
    """Unknown error."""

Attributes

INVALID_TIMESTAMP class-attribute instance-attribute

INVALID_TIMESTAMP = (
    pb.SAMPLE_UPSERT_ERROR_CODE_INVALID_TIMESTAMP
)

The sample timestamp is invalid (e.g., misalignment).

MARKET_LOCATION_INACTIVE class-attribute instance-attribute

MARKET_LOCATION_INACTIVE = (
    pb.SAMPLE_UPSERT_ERROR_CODE_MARKET_LOCATION_INACTIVE
)

The Market Location is inactive.

MARKET_LOCATION_NOT_FOUND class-attribute instance-attribute

MARKET_LOCATION_NOT_FOUND = (
    pb.SAMPLE_UPSERT_ERROR_CODE_MARKET_LOCATION_NOT_FOUND
)

The Market Location does not exist.

REVISION_CONFLICT class-attribute instance-attribute

REVISION_CONFLICT = (
    pb.SAMPLE_UPSERT_ERROR_CODE_REVISION_CONFLICT
)

The sample revision conflicts with an existing sample.

REVISION_TOO_OLD class-attribute instance-attribute

REVISION_TOO_OLD = (
    pb.SAMPLE_UPSERT_ERROR_CODE_REVISION_TOO_OLD
)

The sample revision is older than the current revision.

STORAGE_FAILURE class-attribute instance-attribute

STORAGE_FAILURE = (
    pb.SAMPLE_UPSERT_ERROR_CODE_STORAGE_FAILURE
)

Internal storage error.

UNIT_MISMATCH class-attribute instance-attribute

UNIT_MISMATCH = pb.SAMPLE_UPSERT_ERROR_CODE_UNIT_MISMATCH

The metric unit does not match the Market Location's unit.

UNKNOWN_ERROR class-attribute instance-attribute

UNKNOWN_ERROR = pb.SAMPLE_UPSERT_ERROR_CODE_UNKNOWN_ERROR

Unknown error.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.SAMPLE_UPSERT_ERROR_CODE_UNSPECIFIED

Unspecified error.

UNSUPPORTED_DIRECTION class-attribute instance-attribute

UNSUPPORTED_DIRECTION = (
    pb.SAMPLE_UPSERT_ERROR_CODE_UNSUPPORTED_DIRECTION
)

The direction is not supported by the Market Location.

UNSUPPORTED_METRIC_TYPE class-attribute instance-attribute

UNSUPPORTED_METRIC_TYPE = (
    pb.SAMPLE_UPSERT_ERROR_CODE_UNSUPPORTED_METRIC_TYPE
)

The metric type is not supported by the Market Location.

frequenz.client.marketmetering.TimeResolution

Bases: Enum

Time resolution for metering data.

Source code in src/frequenz/client/marketmetering/types.py

class TimeResolution(Enum):
    """Time resolution for metering data."""

    UNSPECIFIED = pb.TIME_RESOLUTION_UNSPECIFIED
    """Unspecified resolution."""

    MIN_1 = pb.TIME_RESOLUTION_1_MIN
    """1-minute interval."""

    MIN_2 = pb.TIME_RESOLUTION_2_MIN
    """2-minute interval."""

    MIN_5 = pb.TIME_RESOLUTION_5_MIN
    """5-minute interval."""

    MIN_10 = pb.TIME_RESOLUTION_10_MIN
    """10-minute interval."""

    MIN_15 = pb.TIME_RESOLUTION_15_MIN
    """15-minute interval."""

    MIN_30 = pb.TIME_RESOLUTION_30_MIN
    """30-minute interval."""

    MIN_60 = pb.TIME_RESOLUTION_60_MIN
    """60-minute (1 hour) interval."""

    DAY_1 = pb.TIME_RESOLUTION_1_DAY
    """Daily resolution."""

    def to_timedelta(self) -> timedelta | None:
        """Convert to a timedelta.

        Returns:
            The timedelta representation, or None if unspecified.
        """
        mapping = {
            TimeResolution.MIN_1: timedelta(minutes=1),
            TimeResolution.MIN_2: timedelta(minutes=2),
            TimeResolution.MIN_5: timedelta(minutes=5),
            TimeResolution.MIN_10: timedelta(minutes=10),
            TimeResolution.MIN_15: timedelta(minutes=15),
            TimeResolution.MIN_30: timedelta(minutes=30),
            TimeResolution.MIN_60: timedelta(hours=1),
            TimeResolution.DAY_1: timedelta(days=1),
        }
        return mapping.get(self)

Attributes

DAY_1 class-attribute instance-attribute

DAY_1 = pb.TIME_RESOLUTION_1_DAY

Daily resolution.

MIN_1 class-attribute instance-attribute

MIN_1 = pb.TIME_RESOLUTION_1_MIN

1-minute interval.

MIN_10 class-attribute instance-attribute

MIN_10 = pb.TIME_RESOLUTION_10_MIN

10-minute interval.

MIN_15 class-attribute instance-attribute

MIN_15 = pb.TIME_RESOLUTION_15_MIN

15-minute interval.

MIN_2 class-attribute instance-attribute

MIN_2 = pb.TIME_RESOLUTION_2_MIN

2-minute interval.

MIN_30 class-attribute instance-attribute

MIN_30 = pb.TIME_RESOLUTION_30_MIN

30-minute interval.

MIN_5 class-attribute instance-attribute

MIN_5 = pb.TIME_RESOLUTION_5_MIN

5-minute interval.

MIN_60 class-attribute instance-attribute

MIN_60 = pb.TIME_RESOLUTION_60_MIN

60-minute (1 hour) interval.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = pb.TIME_RESOLUTION_UNSPECIFIED

Unspecified resolution.

Methods:

to_timedelta

to_timedelta() -> timedelta | None

Convert to a timedelta.

RETURNS	DESCRIPTION
timedelta | None	The timedelta representation, or None if unspecified.

Source code in src/frequenz/client/marketmetering/types.py

def to_timedelta(self) -> timedelta | None:
    """Convert to a timedelta.

    Returns:
        The timedelta representation, or None if unspecified.
    """
    mapping = {
        TimeResolution.MIN_1: timedelta(minutes=1),
        TimeResolution.MIN_2: timedelta(minutes=2),
        TimeResolution.MIN_5: timedelta(minutes=5),
        TimeResolution.MIN_10: timedelta(minutes=10),
        TimeResolution.MIN_15: timedelta(minutes=15),
        TimeResolution.MIN_30: timedelta(minutes=30),
        TimeResolution.MIN_60: timedelta(hours=1),
        TimeResolution.DAY_1: timedelta(days=1),
    }
    return mapping.get(self)

frequenz.client.marketmetering.UpsertResult dataclass

Result of a sample upsert operation.

If error is None the upsert was accepted; otherwise it was rejected.

Source code in src/frequenz/client/marketmetering/types.py

@dataclass(frozen=True)
class UpsertResult:
    """Result of a sample upsert operation.

    If ``error`` is ``None`` the upsert was accepted; otherwise it was rejected.
    """

    market_location_ref: MarketLocationRef
    """Reference to the Market Location."""

    direction: EnergyFlowDirection
    """Energy-flow direction of the sample (echoed from request)."""

    metric_type: MetricType
    """Metric type of the sample (echoed from request)."""

    metric_unit: MetricUnit
    """Metric unit of the sample (echoed from request)."""

    sample: MarketLocationSample
    """The sample that was upserted (echoed from request)."""

    ingest_time: datetime | None
    """Server-side timestamp when the sample was ingested. Populated only on success."""

    error: SampleUpsertError | None
    """Error details if the upsert was rejected, ``None`` on success."""

    @classmethod
    def from_protobuf(
        cls, pb_obj: pb.UpsertMarketLocationSamplesStreamResponse
    ) -> Self:
        """Create from protobuf message.

        Args:
            pb_obj: The protobuf message.

        Returns:
            A new UpsertResult instance.
        """
        ingest_time = None
        if pb_obj.HasField("ingest_time"):
            ingest_time = _timestamp_to_datetime(pb_obj.ingest_time)

        error = None
        if pb_obj.HasField("error"):
            error = SampleUpsertError.from_protobuf(pb_obj.error)

        return cls(
            market_location_ref=MarketLocationRef.from_protobuf(
                pb_obj.market_location_ref
            ),
            direction=EnergyFlowDirection(pb_obj.direction),
            metric_type=MetricType(pb_obj.metric_type),
            metric_unit=MetricUnit(pb_obj.metric_unit),
            sample=MarketLocationSample.from_protobuf(pb_obj.sample),
            ingest_time=ingest_time,
            error=error,
        )

Attributes

direction instance-attribute

direction: EnergyFlowDirection

Energy-flow direction of the sample (echoed from request).

error instance-attribute

error: SampleUpsertError | None

Error details if the upsert was rejected, None on success.

ingest_time instance-attribute

ingest_time: datetime | None

Server-side timestamp when the sample was ingested. Populated only on success.

market_location_ref instance-attribute

market_location_ref: MarketLocationRef

Reference to the Market Location.

metric_type instance-attribute

metric_type: MetricType

Metric type of the sample (echoed from request).

metric_unit instance-attribute

metric_unit: MetricUnit

Metric unit of the sample (echoed from request).

sample instance-attribute

sample: MarketLocationSample

The sample that was upserted (echoed from request).

Methods:

from_protobuf classmethod

from_protobuf(
    pb_obj: UpsertMarketLocationSamplesStreamResponse,
) -> Self

Create from protobuf message.

PARAMETER	DESCRIPTION
pb_obj	The protobuf message. TYPE: UpsertMarketLocationSamplesStreamResponse

RETURNS	DESCRIPTION
Self	A new UpsertResult instance.

Source code in src/frequenz/client/marketmetering/types.py

@classmethod
def from_protobuf(
    cls, pb_obj: pb.UpsertMarketLocationSamplesStreamResponse
) -> Self:
    """Create from protobuf message.

    Args:
        pb_obj: The protobuf message.

    Returns:
        A new UpsertResult instance.
    """
    ingest_time = None
    if pb_obj.HasField("ingest_time"):
        ingest_time = _timestamp_to_datetime(pb_obj.ingest_time)

    error = None
    if pb_obj.HasField("error"):
        error = SampleUpsertError.from_protobuf(pb_obj.error)

    return cls(
        market_location_ref=MarketLocationRef.from_protobuf(
            pb_obj.market_location_ref
        ),
        direction=EnergyFlowDirection(pb_obj.direction),
        metric_type=MetricType(pb_obj.metric_type),
        metric_unit=MetricUnit(pb_obj.metric_unit),
        sample=MarketLocationSample.from_protobuf(pb_obj.sample),
        ingest_time=ingest_time,
        error=error,
    )