proto

frequenz.client.common.proto

General utilities for converting common types to/from protobuf types.

Functions:

frequenz.client.common.proto.datetime_from_proto

datetime_from_proto(
    ts: Timestamp, tz: timezone = utc
) -> datetime

Convert a protobuf Timestamp to a datetime.

PARAMETER	DESCRIPTION
ts	The Timestamp object to convert. TYPE: Timestamp
tz	The timezone to use for the datetime. TYPE: timezone DEFAULT: utc

RETURNS	DESCRIPTION
datetime	The Timestamp converted to a datetime.

Source code in src/frequenz/client/common/proto/_timestamp.py

def datetime_from_proto(
    ts: timestamp_pb2.Timestamp, tz: timezone = timezone.utc
) -> datetime:
    """Convert a protobuf Timestamp to a datetime.

    Args:
        ts: The Timestamp object to convert.
        tz: The timezone to use for the datetime.

    Returns:
        The Timestamp converted to a datetime.
    """
    # Add microseconds and add nanoseconds converted to microseconds
    microseconds = int(ts.nanos / 1000)
    return datetime.fromtimestamp(ts.seconds + microseconds * 1e-6, tz=tz)

frequenz.client.common.proto.datetime_to_proto

datetime_to_proto(dt: datetime) -> Timestamp

datetime_to_proto(dt: None) -> None

datetime_to_proto(dt: datetime | None) -> Timestamp | None

Convert a datetime to a protobuf Timestamp.

PARAMETER	DESCRIPTION
dt	The datetime object to convert. TYPE: datetime | None

RETURNS	DESCRIPTION
Timestamp | None	The datetime converted to a Timestamp, or None if dt is None.

Source code in src/frequenz/client/common/proto/_timestamp.py

def datetime_to_proto(dt: datetime | None) -> timestamp_pb2.Timestamp | None:
    """Convert a datetime to a protobuf Timestamp.

    Args:
        dt: The datetime object to convert.

    Returns:
        The datetime converted to a Timestamp, or `None` if `dt` is `None`.
    """
    if dt is None:
        return None

    ts = timestamp_pb2.Timestamp()
    ts.FromDatetime(dt)
    return ts

frequenz.client.common.proto.enum_from_proto

enum_from_proto(
    value: int,
    enum_type: type[EnumT],
    *,
    allow_invalid: Literal[False]
) -> EnumT

enum_from_proto(
    value: int,
    enum_type: type[EnumT],
    *,
    allow_invalid: Literal[True] = True
) -> EnumT | int

enum_from_proto(
    value: int,
    enum_type: type[EnumT],
    *,
    allow_invalid: bool = True
) -> EnumT | int

Convert a protobuf int enum value to a python enum.

Example

from frequenz.core.enum import Enum, unique

from proto import proto_pb2  # Just an example. pylint: disable=import-error

@unique
class SomeEnum(Enum):
    # These values should match the protobuf enum values.
    SOME_VALUE = 1

enum_value = enum_from_proto(proto_pb2.SomeEnum.SOME_ENUM_SOME_VALUE, SomeEnum)
# -> SomeEnum.SOME_VALUE

enum_value = enum_from_proto(0, SomeEnum)
# -> 0

enum_value = enum_from_proto(42, SomeEnum)
# -> 42

enum_value = enum_from_proto(
    proto_pb2.SomeEnum.SOME_ENUM_UNKNOWN_VALUE, SomeEnum, allow_invalid=False
)
# -> ValueError

PARAMETER	DESCRIPTION
value	The protobuf int enum value. TYPE: int
enum_type	The python enum type to convert to. TYPE: type[EnumT]
allow_invalid	If True, return the value as an int if the value is not a valid member of the enum (this allows for forward-compatibility with new enum values defined in the protocol but not added to the Python enum yet). If False, raise a ValueError if the value is not a valid member of the enum. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
EnumT | int	The resulting python enum value if the protobuf value is known, otherwise the input value converted to a plain int.

RAISES	DESCRIPTION
ValueError	If allow_invalid is False and the value is not a valid member of the enum.

Source code in src/frequenz/client/common/proto/_enum.py

def enum_from_proto(
    value: int, enum_type: type[EnumT], *, allow_invalid: bool = True
) -> EnumT | int:
    """Convert a protobuf int enum value to a python enum.

    Example:
        ```python
        from frequenz.core.enum import Enum, unique

        from proto import proto_pb2  # Just an example. pylint: disable=import-error

        @unique
        class SomeEnum(Enum):
            # These values should match the protobuf enum values.
            SOME_VALUE = 1

        enum_value = enum_from_proto(proto_pb2.SomeEnum.SOME_ENUM_SOME_VALUE, SomeEnum)
        # -> SomeEnum.SOME_VALUE

        enum_value = enum_from_proto(0, SomeEnum)
        # -> 0

        enum_value = enum_from_proto(42, SomeEnum)
        # -> 42

        enum_value = enum_from_proto(
            proto_pb2.SomeEnum.SOME_ENUM_UNKNOWN_VALUE, SomeEnum, allow_invalid=False
        )
        # -> ValueError
        ```

    Args:
        value: The protobuf int enum value.
        enum_type: The python enum type to convert to.
        allow_invalid: If `True`, return the value as an [`int`][] if the value is
            not a valid member of the enum (this allows for forward-compatibility with
            new enum values defined in the protocol but not added to the Python enum
            yet). If `False`, raise a [`ValueError`][] if the value is not a valid
            member of the enum.

    Returns:
        The resulting python enum value if the protobuf value is known, otherwise
            the input value converted to a plain [`int`][].

    Raises:
        ValueError: If `allow_invalid` is `False` and the value is not a valid member
            of the enum.
    """
    try:
        return enum_type(value)
    except ValueError:
        if allow_invalid:
            return value
        raise