v1alpha8

frequenz.client.common.metrics.proto.v1alpha8

Conversion of metrics enums from/to protobuf v1alpha8.

Functions

frequenz.client.common.metrics.proto.v1alpha8.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 src/frequenz/client/common/metrics/proto/v1alpha8/_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.v1alpha8.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 src/frequenz/client/common/metrics/proto/v1alpha8/_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.v1alpha8.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 src/frequenz/client/common/metrics/proto/v1alpha8/_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.v1alpha8.metric_connection_category_from_proto

metric_connection_category_from_proto(
    message: ValueType,
) -> MetricConnectionCategory | int

Convert a protobuf MetricConnectionCategory enum value to an enum member.

PARAMETER	DESCRIPTION
message	A protobuf MetricConnectionCategory enum value. TYPE: ValueType

RETURNS	DESCRIPTION
MetricConnectionCategory | int	The corresponding MetricConnectionCategory enum member, or the raw int if the protobuf value is not recognized.

Source code in src/frequenz/client/common/metrics/proto/v1alpha8/_metric_connection_category.py

def metric_connection_category_from_proto(
    message: metrics_pb2.MetricConnectionCategory.ValueType,
) -> MetricConnectionCategory | int:
    """Convert a protobuf MetricConnectionCategory enum value to an enum member.

    Args:
        message: A protobuf MetricConnectionCategory enum value.

    Returns:
        The corresponding MetricConnectionCategory enum member, or the raw `int` if
            the protobuf value is not recognized.
    """
    return enum_from_proto(message, MetricConnectionCategory)

frequenz.client.common.metrics.proto.v1alpha8.metric_connection_category_to_proto

metric_connection_category_to_proto(
    category: MetricConnectionCategory,
) -> ValueType

Convert a MetricConnectionCategory enum member to a protobuf enum value.

PARAMETER	DESCRIPTION
category	A MetricConnectionCategory enum member. TYPE: MetricConnectionCategory

RETURNS	DESCRIPTION
ValueType	The corresponding protobuf MetricConnectionCategory enum value.

Source code in src/frequenz/client/common/metrics/proto/v1alpha8/_metric_connection_category.py

def metric_connection_category_to_proto(
    category: MetricConnectionCategory,
) -> metrics_pb2.MetricConnectionCategory.ValueType:
    """Convert a MetricConnectionCategory enum member to a protobuf enum value.

    Args:
        category: A MetricConnectionCategory enum member.

    Returns:
        The corresponding protobuf MetricConnectionCategory enum value.
    """
    return metrics_pb2.MetricConnectionCategory.ValueType(category.value)

frequenz.client.common.metrics.proto.v1alpha8.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 src/frequenz/client/common/metrics/proto/v1alpha8/_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 = metric_connection_category_from_proto(message.category)

    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.v1alpha8.metric_from_proto

metric_from_proto(message: ValueType) -> Metric | int

Convert a protobuf Metric message to a Metric enum member.

PARAMETER	DESCRIPTION
message	A protobuf Metric message. TYPE: ValueType

RETURNS	DESCRIPTION
Metric | int	The corresponding Metric enum member.

Source code in src/frequenz/client/common/metrics/proto/v1alpha8/_metric.py

def metric_from_proto(message: metrics_pb2.Metric.ValueType) -> Metric | int:
    """Convert a protobuf Metric message to a Metric enum member.

    Args:
        message: A protobuf Metric message.

    Returns:
        The corresponding Metric enum member.
    """
    return enum_from_proto(message, Metric)

frequenz.client.common.metrics.proto.v1alpha8.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 src/frequenz/client/common/metrics/proto/v1alpha8/_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 = metric_from_proto(message.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,
    )

frequenz.client.common.metrics.proto.v1alpha8.metric_to_proto

metric_to_proto(metric: Metric) -> ValueType

Convert a Metric enum member to a protobuf Metric message.

PARAMETER	DESCRIPTION
metric	A Metric enum member. TYPE: Metric

RETURNS	DESCRIPTION
ValueType	The corresponding protobuf Metric message.

Source code in src/frequenz/client/common/metrics/proto/v1alpha8/_metric.py

def metric_to_proto(metric: Metric) -> metrics_pb2.Metric.ValueType:
    """Convert a Metric enum member to a protobuf Metric message.

    Args:
        metric: A Metric enum member.

    Returns:
        The corresponding protobuf Metric message.
    """
    return metrics_pb2.Metric.ValueType(metric.value)