proto

frequenz.client.common.metrics.proto

Metrics objects to proto conversion functions.

Functions

frequenz.client.common.metrics.proto.aggregated_metric_sample_from_proto

aggregated_metric_sample_from_proto(
    message: AggregatedMetricValue,
) -> AggregatedMetricValue

Convert a protobuf message to a AggregatedMetricValue object.

PARAMETER	DESCRIPTION
message	The protobuf message to convert. TYPE: AggregatedMetricValue

RETURNS	DESCRIPTION
AggregatedMetricValue	The resulting AggregatedMetricValue object.

Source code in frequenz/client/common/metrics/proto/_sample.py

def aggregated_metric_sample_from_proto(
    message: metrics_pb2.AggregatedMetricValue,
) -> AggregatedMetricValue:
    """Convert a protobuf message to a `AggregatedMetricValue` object.

    Args:
        message: The protobuf message to convert.

    Returns:
        The resulting `AggregatedMetricValue` object.
    """
    return AggregatedMetricValue(
        avg=message.avg_value,
        min=message.min_value if message.HasField("min_value") else None,
        max=message.max_value if message.HasField("max_value") else None,
        raw=message.raw_values,
    )

frequenz.client.common.metrics.proto.bounds_from_proto

bounds_from_proto(message: Bounds) -> Bounds

Create a Bounds object from a protobuf message.

PARAMETER	DESCRIPTION
message	The protobuf message to convert. TYPE: Bounds

RETURNS	DESCRIPTION
Bounds	The corresponding Bounds object.

RAISES	DESCRIPTION
ValueError	If the message is not valid.

Source code in frequenz/client/common/metrics/proto/_bounds.py

def bounds_from_proto(message: bounds_pb2.Bounds) -> Bounds:  # noqa: DOC502
    """Create a `Bounds` object from a protobuf message.

    Args:
        message: The protobuf message to convert.

    Returns:
        The corresponding `Bounds` object.

    Raises:
        ValueError: If the message is not valid.
    """
    return Bounds(
        lower=message.lower if message.HasField("lower") else None,
        upper=message.upper if message.HasField("upper") else None,
    )

frequenz.client.common.metrics.proto.bounds_from_proto_with_issues

bounds_from_proto_with_issues(
    message: Bounds,
    *,
    major_issues: list[str],
    minor_issues: list[str]
) -> Bounds | None

Create a Bounds object from a protobuf message, collecting issues.

PARAMETER	DESCRIPTION
message	The protobuf message to convert. TYPE: Bounds
major_issues	A list to append major issues to. TYPE: list[str]
minor_issues	A list to append minor issues to. TYPE: list[str]

RETURNS	DESCRIPTION
Bounds | None	The corresponding Bounds object.

Source code in frequenz/client/common/metrics/proto/_bounds.py

def bounds_from_proto_with_issues(
    message: bounds_pb2.Bounds,
    *,
    major_issues: list[str],
    minor_issues: list[str],  # pylint: disable=unused-argument
) -> Bounds | None:  # noqa: DOC502
    """Create a `Bounds` object from a protobuf message, collecting issues.

    Args:
        message: The protobuf message to convert.
        major_issues: A list to append major issues to.
        minor_issues: A list to append minor issues to.

    Returns:
        The corresponding `Bounds` object.
    """
    try:
        return bounds_from_proto(message)
    except ValueError as exc:
        major_issues.append(str(exc))
        return None

frequenz.client.common.metrics.proto.metric_connection_from_proto_with_issues

metric_connection_from_proto_with_issues(
    message: MetricConnection,
    *,
    major_issues: list[str],
    minor_issues: list[str]
) -> MetricConnection

Convert a protobuf message to a MetricConnection object.

PARAMETER	DESCRIPTION
message	The protobuf message to convert. TYPE: MetricConnection
major_issues	A list to append major issues to. TYPE: list[str]
minor_issues	A list to append minor issues to. TYPE: list[str]

RETURNS	DESCRIPTION
MetricConnection	The resulting MetricConnection object.

Source code in frequenz/client/common/metrics/proto/_sample.py

def metric_connection_from_proto_with_issues(
    message: metrics_pb2.MetricConnection,
    *,
    major_issues: list[str],
    minor_issues: list[str],
) -> MetricConnection:
    """Convert a protobuf message to a `MetricConnection` object.

    Args:
        message: The protobuf message to convert.
        major_issues: A list to append major issues to.
        minor_issues: A list to append minor issues to.

    Returns:
        The resulting `MetricConnection` object.
    """
    category = enum_from_proto(message.category, MetricConnectionCategory)

    match category:
        case MetricConnectionCategory.UNSPECIFIED:
            major_issues.append("unspecified category")
        case int():
            minor_issues.append(f"unrecognized category {category}")

    return MetricConnection(
        category=category,
        name=message.name or None,
    )

frequenz.client.common.metrics.proto.metric_sample_from_proto_with_issues

metric_sample_from_proto_with_issues(
    message: MetricSample,
    *,
    major_issues: list[str],
    minor_issues: list[str]
) -> MetricSample

Convert a protobuf message to a MetricSample object.

PARAMETER	DESCRIPTION
message	The protobuf message to convert. TYPE: MetricSample
major_issues	A list to append major issues to. TYPE: list[str]
minor_issues	A list to append minor issues to. TYPE: list[str]

RETURNS	DESCRIPTION
MetricSample	The resulting MetricSample object.

Source code in frequenz/client/common/metrics/proto/_sample.py

def metric_sample_from_proto_with_issues(
    message: metrics_pb2.MetricSample,
    *,
    major_issues: list[str],
    minor_issues: list[str],
) -> MetricSample:
    """Convert a protobuf message to a `MetricSample` object.

    Args:
        message: The protobuf message to convert.
        major_issues: A list to append major issues to.
        minor_issues: A list to append minor issues to.

    Returns:
        The resulting `MetricSample` object.
    """
    sample_time = datetime_from_proto(message.sample_time)

    metric = enum_from_proto(message.metric, Metric)

    value: float | AggregatedMetricValue | None = None
    if message.HasField("value"):
        match message.value.WhichOneof("metric_value_variant"):
            case "simple_metric":
                value = message.value.simple_metric.value
            case "aggregated_metric":
                value = aggregated_metric_sample_from_proto(
                    message.value.aggregated_metric
                )

    bounds = _metric_bounds_from_proto(
        metric, message.bounds, major_issues=major_issues, minor_issues=minor_issues
    )

    connection = None
    if message.HasField("connection"):
        connection = metric_connection_from_proto_with_issues(
            message.connection, major_issues=major_issues, minor_issues=minor_issues
        )

    return MetricSample(
        sample_time=sample_time,
        metric=metric,
        value=value,
        bounds=bounds,
        connection=connection,
    )