types

frequenz.client.dispatch.types

Type wrappers for the generated protobuf messages.

Attributes

frequenz.client.dispatch.types.TargetCategoryInputType module-attribute

TargetCategoryInputType = (
    TargetCategory
    | ComponentCategory
    | BatteryType
    | InverterType
    | EvChargerType
)

Type for the input to TargetCategories constructor.

frequenz.client.dispatch.types.TargetComponents module-attribute

TargetComponents: TypeAlias = TargetIds | TargetCategories

Target components.

Can be one of the following:

• A set of target component IDs (TargetIds)
• A set of target component categories with opt. types (TargetCategories)

This is a frozen set, so it is immutable. The target components are used to specify the components that a dispatch should target.

Classes

frequenz.client.dispatch.types.BatteryType

Bases: Enum

Enum representing the type of battery.

Source code in frequenz/client/dispatch/types.py

class BatteryType(Enum):
    """Enum representing the type of battery."""

    UNSPECIFIED = PBBatteryType.BATTERY_TYPE_UNSPECIFIED
    """Unspecified type of battery."""

    LI_ION = PBBatteryType.BATTERY_TYPE_LI_ION
    """Lithium-ion battery."""

    NA_ION = PBBatteryType.BATTERY_TYPE_NA_ION
    """Sodium-ion battery."""

Attributes

LI_ION class-attribute instance-attribute

LI_ION = BATTERY_TYPE_LI_ION

Lithium-ion battery.

NA_ION class-attribute instance-attribute

NA_ION = BATTERY_TYPE_NA_ION

Sodium-ion battery.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = BATTERY_TYPE_UNSPECIFIED

Unspecified type of battery.

frequenz.client.dispatch.types.Dispatch dataclass

Represents a dispatch operation within a microgrid system.

Source code in frequenz/client/dispatch/types.py

@dataclass(kw_only=True, frozen=True)
class Dispatch:  # pylint: disable=too-many-instance-attributes
    """Represents a dispatch operation within a microgrid system."""

    id: DispatchId
    """The unique identifier for the dispatch."""

    type: str
    """User-defined information about the type of dispatch.

    This is understood and processed by downstream applications."""

    start_time: datetime
    """The start time of the dispatch in UTC."""

    duration: timedelta | None
    """The duration of the dispatch, represented as a timedelta."""

    target: TargetComponents
    """The target components of the dispatch."""

    active: bool
    """Indicates whether the dispatch is active and eligible for processing."""

    dry_run: bool
    """Indicates if the dispatch is a dry run.

    Executed for logging and monitoring without affecting actual component states."""

    payload: dict[str, Any]
    """The dispatch payload containing arbitrary data.

    It is structured as needed for the dispatch operation."""

    recurrence: RecurrenceRule
    """The recurrence rule for the dispatch.

    Defining any repeating patterns or schedules."""

    create_time: datetime
    """The creation time of the dispatch in UTC. Set when a dispatch is created."""

    update_time: datetime
    """The last update time of the dispatch in UTC. Set when a dispatch is modified."""

    end_time: datetime | None = None
    """The end time of the dispatch in UTC.

    Calculated and sent by the backend service.
    """

    @property
    def started(self) -> bool:
        """Check if the dispatch has started.

        A dispatch is considered started if the current time is after the start
        time but before the end time.

        Recurring dispatches are considered started if the current time is after
        the start time of the last occurrence but before the end time of the
        last occurrence.
        """
        if not self.active:
            return False

        now = datetime.now(tz=timezone.utc)

        if now < self.start_time:
            return False

        # A dispatch without duration is always running, once it started
        if self.duration is None:
            return True

        if until := self._until(now):
            return now < until

        return False

    @property
    def until(self) -> datetime | None:
        """Time when the dispatch should end.

        Returns the time that a running dispatch should end.
        If the dispatch is not running, None is returned.

        Returns:
            The time when the dispatch should end or None if the dispatch is not running.
        """
        if not self.active:
            return None

        now = datetime.now(tz=timezone.utc)
        return self._until(now)

    @property
    def next_run(self) -> datetime | None:
        """Calculate the next run of a dispatch.

        Returns:
            The next run of the dispatch or None if the dispatch is finished.
        """
        return self.next_run_after(datetime.now(tz=timezone.utc))

    def next_run_after(self, after: datetime) -> datetime | None:
        """Calculate the next run of a dispatch.

        Args:
            after: The time to calculate the next run from.

        Returns:
            The next run of the dispatch or None if the dispatch is finished.
        """
        if (
            not self.recurrence.frequency
            or self.recurrence.frequency == Frequency.UNSPECIFIED
            or self.duration is None  # Infinite duration
        ):
            if after > self.start_time:
                return None
            return self.start_time

        # Make sure no weekday is UNSPECIFIED
        if Weekday.UNSPECIFIED in self.recurrence.byweekdays:
            return None

        # No type information for rrule, so we need to cast
        return cast(
            datetime | None,
            self.recurrence._as_rrule(  # pylint: disable=protected-access
                self.start_time
            ).after(after, inc=True),
        )

    def _until(self, now: datetime) -> datetime | None:
        """Calculate the time when the dispatch should end.

        If no previous run is found, None is returned.

        Args:
            now: The current time.

        Returns:
            The time when the dispatch should end or None if the dispatch is not running.

        Raises:
            ValueError: If the dispatch has no duration.
        """
        if self.duration is None:
            raise ValueError("_until: Dispatch has no duration")

        if (
            not self.recurrence.frequency
            or self.recurrence.frequency == Frequency.UNSPECIFIED
        ):
            return self.start_time + self.duration

        latest_past_start: datetime | None = (
            self.recurrence._as_rrule(  # pylint: disable=protected-access
                self.start_time
            ).before(now, inc=True)
        )

        if not latest_past_start:
            return None

        return latest_past_start + self.duration

    @classmethod
    def from_protobuf(cls, pb_object: PBDispatch) -> "Dispatch":
        """Convert a protobuf dispatch to a dispatch.

        Args:
            pb_object: The protobuf dispatch to convert.

        Returns:
            The converted dispatch.
        """
        return Dispatch(
            id=DispatchId(pb_object.metadata.dispatch_id),
            type=pb_object.data.type,
            create_time=to_datetime(pb_object.metadata.create_time),
            update_time=to_datetime(pb_object.metadata.update_time),
            end_time=(
                to_datetime(pb_object.metadata.end_time)
                if pb_object.metadata.HasField("end_time")
                else None
            ),
            start_time=to_datetime(pb_object.data.start_time),
            duration=(
                timedelta(seconds=pb_object.data.duration)
                if pb_object.data.HasField("duration")
                else None
            ),
            target=_target_components_from_protobuf(pb_object.data.target),
            active=pb_object.data.is_active,
            dry_run=pb_object.data.is_dry_run,
            payload=MessageToDict(pb_object.data.payload),
            recurrence=RecurrenceRule.from_protobuf(pb_object.data.recurrence),
        )

    def to_protobuf(self) -> PBDispatch:
        """Convert a dispatch to a protobuf dispatch.

        Returns:
            The converted protobuf dispatch.
        """
        payload = Struct()
        payload.update(self.payload)

        return PBDispatch(
            metadata=DispatchMetadata(
                dispatch_id=int(self.id),
                create_time=to_timestamp(self.create_time),
                update_time=to_timestamp(self.update_time),
                end_time=(
                    to_timestamp(self.end_time) if self.end_time is not None else None
                ),
            ),
            data=DispatchData(
                type=self.type,
                start_time=to_timestamp(self.start_time),
                duration=(
                    None
                    if self.duration is None
                    else round(self.duration.total_seconds())
                ),
                target=_target_components_to_protobuf(self.target),
                is_active=self.active,
                is_dry_run=self.dry_run,
                payload=payload,
                recurrence=self.recurrence.to_protobuf() if self.recurrence else None,
            ),
        )

Attributes

active instance-attribute

active: bool

Indicates whether the dispatch is active and eligible for processing.

create_time instance-attribute

create_time: datetime

The creation time of the dispatch in UTC. Set when a dispatch is created.

dry_run instance-attribute

dry_run: bool

Indicates if the dispatch is a dry run.

Executed for logging and monitoring without affecting actual component states.

duration instance-attribute

duration: timedelta | None

The duration of the dispatch, represented as a timedelta.

end_time class-attribute instance-attribute

end_time: datetime | None = None

The end time of the dispatch in UTC.

Calculated and sent by the backend service.

id instance-attribute

id: DispatchId

The unique identifier for the dispatch.

next_run property

next_run: datetime | None

Calculate the next run of a dispatch.

RETURNS	DESCRIPTION
datetime | None	The next run of the dispatch or None if the dispatch is finished.

payload instance-attribute

payload: dict[str, Any]

The dispatch payload containing arbitrary data.

It is structured as needed for the dispatch operation.

recurrence instance-attribute

recurrence: RecurrenceRule

The recurrence rule for the dispatch.

Defining any repeating patterns or schedules.

start_time instance-attribute

start_time: datetime

The start time of the dispatch in UTC.

started property

started: bool

Check if the dispatch has started.

A dispatch is considered started if the current time is after the start time but before the end time.

Recurring dispatches are considered started if the current time is after the start time of the last occurrence but before the end time of the last occurrence.

target instance-attribute

target: TargetComponents

The target components of the dispatch.

type instance-attribute

type: str

User-defined information about the type of dispatch.

This is understood and processed by downstream applications.

until property

until: datetime | None

Time when the dispatch should end.

Returns the time that a running dispatch should end. If the dispatch is not running, None is returned.

RETURNS	DESCRIPTION
datetime | None	The time when the dispatch should end or None if the dispatch is not running.

update_time instance-attribute

update_time: datetime

The last update time of the dispatch in UTC. Set when a dispatch is modified.

Functions

from_protobuf classmethod

from_protobuf(pb_object: Dispatch) -> Dispatch

Convert a protobuf dispatch to a dispatch.

PARAMETER	DESCRIPTION
pb_object	The protobuf dispatch to convert. TYPE: Dispatch

RETURNS	DESCRIPTION
Dispatch	The converted dispatch.

Source code in frequenz/client/dispatch/types.py

@classmethod
def from_protobuf(cls, pb_object: PBDispatch) -> "Dispatch":
    """Convert a protobuf dispatch to a dispatch.

    Args:
        pb_object: The protobuf dispatch to convert.

    Returns:
        The converted dispatch.
    """
    return Dispatch(
        id=DispatchId(pb_object.metadata.dispatch_id),
        type=pb_object.data.type,
        create_time=to_datetime(pb_object.metadata.create_time),
        update_time=to_datetime(pb_object.metadata.update_time),
        end_time=(
            to_datetime(pb_object.metadata.end_time)
            if pb_object.metadata.HasField("end_time")
            else None
        ),
        start_time=to_datetime(pb_object.data.start_time),
        duration=(
            timedelta(seconds=pb_object.data.duration)
            if pb_object.data.HasField("duration")
            else None
        ),
        target=_target_components_from_protobuf(pb_object.data.target),
        active=pb_object.data.is_active,
        dry_run=pb_object.data.is_dry_run,
        payload=MessageToDict(pb_object.data.payload),
        recurrence=RecurrenceRule.from_protobuf(pb_object.data.recurrence),
    )

next_run_after

next_run_after(after: datetime) -> datetime | None

Calculate the next run of a dispatch.

PARAMETER	DESCRIPTION
after	The time to calculate the next run from. TYPE: datetime

RETURNS	DESCRIPTION
datetime | None	The next run of the dispatch or None if the dispatch is finished.

Source code in frequenz/client/dispatch/types.py

def next_run_after(self, after: datetime) -> datetime | None:
    """Calculate the next run of a dispatch.

    Args:
        after: The time to calculate the next run from.

    Returns:
        The next run of the dispatch or None if the dispatch is finished.
    """
    if (
        not self.recurrence.frequency
        or self.recurrence.frequency == Frequency.UNSPECIFIED
        or self.duration is None  # Infinite duration
    ):
        if after > self.start_time:
            return None
        return self.start_time

    # Make sure no weekday is UNSPECIFIED
    if Weekday.UNSPECIFIED in self.recurrence.byweekdays:
        return None

    # No type information for rrule, so we need to cast
    return cast(
        datetime | None,
        self.recurrence._as_rrule(  # pylint: disable=protected-access
            self.start_time
        ).after(after, inc=True),
    )

to_protobuf

to_protobuf() -> Dispatch

Convert a dispatch to a protobuf dispatch.

RETURNS	DESCRIPTION
Dispatch	The converted protobuf dispatch.

Source code in frequenz/client/dispatch/types.py

def to_protobuf(self) -> PBDispatch:
    """Convert a dispatch to a protobuf dispatch.

    Returns:
        The converted protobuf dispatch.
    """
    payload = Struct()
    payload.update(self.payload)

    return PBDispatch(
        metadata=DispatchMetadata(
            dispatch_id=int(self.id),
            create_time=to_timestamp(self.create_time),
            update_time=to_timestamp(self.update_time),
            end_time=(
                to_timestamp(self.end_time) if self.end_time is not None else None
            ),
        ),
        data=DispatchData(
            type=self.type,
            start_time=to_timestamp(self.start_time),
            duration=(
                None
                if self.duration is None
                else round(self.duration.total_seconds())
            ),
            target=_target_components_to_protobuf(self.target),
            is_active=self.active,
            is_dry_run=self.dry_run,
            payload=payload,
            recurrence=self.recurrence.to_protobuf() if self.recurrence else None,
        ),
    )

frequenz.client.dispatch.types.DispatchEvent dataclass

Represents an event that occurred during a dispatch operation.

Source code in frequenz/client/dispatch/types.py

@dataclass(kw_only=True, frozen=True)
class DispatchEvent:
    """Represents an event that occurred during a dispatch operation."""

    dispatch: Dispatch
    """The dispatch associated with the event."""

    event: Event
    """The type of event that occurred."""

    @classmethod
    def from_protobuf(
        cls, pb_object: StreamMicrogridDispatchesResponse
    ) -> "DispatchEvent":
        """Convert a protobuf dispatch event to a dispatch event.

        Args:
            pb_object: The protobuf dispatch event to convert.

        Returns:
            The converted dispatch event.
        """
        return DispatchEvent(
            dispatch=Dispatch.from_protobuf(pb_object.dispatch),
            event=Event(pb_object.event),
        )

Attributes

dispatch instance-attribute

dispatch: Dispatch

The dispatch associated with the event.

event instance-attribute

event: Event

The type of event that occurred.

Functions

from_protobuf classmethod

from_protobuf(
    pb_object: StreamMicrogridDispatchesResponse,
) -> DispatchEvent

Convert a protobuf dispatch event to a dispatch event.

PARAMETER	DESCRIPTION
pb_object	The protobuf dispatch event to convert. TYPE: StreamMicrogridDispatchesResponse

RETURNS	DESCRIPTION
DispatchEvent	The converted dispatch event.

Source code in frequenz/client/dispatch/types.py

@classmethod
def from_protobuf(
    cls, pb_object: StreamMicrogridDispatchesResponse
) -> "DispatchEvent":
    """Convert a protobuf dispatch event to a dispatch event.

    Args:
        pb_object: The protobuf dispatch event to convert.

    Returns:
        The converted dispatch event.
    """
    return DispatchEvent(
        dispatch=Dispatch.from_protobuf(pb_object.dispatch),
        event=Event(pb_object.event),
    )

frequenz.client.dispatch.types.DispatchId

Bases: BaseId

A unique identifier for a dispatch.

Source code in frequenz/client/dispatch/types.py

@final
class DispatchId(BaseId, str_prefix="DID"):
    """A unique identifier for a dispatch."""

Attributes

str_prefix property

str_prefix: str

The prefix used for the string representation of this ID.

Functions

__eq__

__eq__(other: object) -> bool

Check if this instance is equal to another object.

Equality is defined as being of the exact same type and having the same underlying ID.

Source code in frequenz/core/id.py

def __eq__(self, other: object) -> bool:
    """Check if this instance is equal to another object.

    Equality is defined as being of the exact same type and having the same
    underlying ID.
    """
    # pylint thinks this is not an unidiomatic typecheck, but in this case
    # it is not. isinstance() returns True for subclasses, which is not
    # what we want here, as different ID types should never be equal.
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    # We already checked type(other) is type(self), but mypy doesn't
    # understand that, so we need to cast it to Self.
    other_id = cast(Self, other)
    return self._id == other_id._id

__hash__

__hash__() -> int

Return the hash of this instance.

The hash is based on the exact type and the underlying ID to ensure that IDs of different types but with the same numeric value have different hashes.

Source code in frequenz/core/id.py

def __hash__(self) -> int:
    """Return the hash of this instance.

    The hash is based on the exact type and the underlying ID to ensure
    that IDs of different types but with the same numeric value have different hashes.
    """
    return hash((type(self), self._id))

__init__

__init__(id_: int) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
id_	The numeric unique identifier. TYPE: int

RAISES	DESCRIPTION
ValueError	If the ID is negative.

Source code in frequenz/core/id.py

def __init__(self, id_: int, /) -> None:
    """Initialize this instance.

    Args:
        id_: The numeric unique identifier.

    Raises:
        ValueError: If the ID is negative.
    """
    if id_ < 0:
        raise ValueError(f"{type(self).__name__} can't be negative.")
    self._id = id_

__init_subclass__

__init_subclass__(
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any
) -> None

Initialize a subclass, set its string prefix, and perform checks.

PARAMETER	DESCRIPTION
str_prefix	The string prefix for the ID type (e.g., "MID"). Must be unique across all ID types. TYPE: str
allow_custom_name	If True, bypasses the check that the class name must end with "Id". Defaults to False. TYPE: bool DEFAULT: False
**kwargs	Forwarded to the parent's init_subclass. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
TypeError	If allow_custom_name is False and the class name does not end with "Id".

Source code in frequenz/core/id.py

def __init_subclass__(
    cls,
    *,
    str_prefix: str,
    allow_custom_name: bool = False,
    **kwargs: Any,
) -> None:
    """Initialize a subclass, set its string prefix, and perform checks.

    Args:
        str_prefix: The string prefix for the ID type (e.g., "MID").
            Must be unique across all ID types.
        allow_custom_name: If True, bypasses the check that the class name
            must end with "Id". Defaults to False.
        **kwargs: Forwarded to the parent's __init_subclass__.

    Raises:
        TypeError: If `allow_custom_name` is False and the class name
            does not end with "Id".
    """
    super().__init_subclass__(**kwargs)

    if str_prefix in BaseId._registered_prefixes:
        # We want to raise an exception here, but currently can't due to
        # https://github.com/frequenz-floss/frequenz-repo-config-python/issues/421
        _logger.warning(
            "Prefix '%s' is already registered. ID prefixes must be unique.",
            str_prefix,
        )
    BaseId._registered_prefixes.add(str_prefix)

    if not allow_custom_name and not cls.__name__.endswith("Id"):
        raise TypeError(
            f"Class name '{cls.__name__}' for an ID class must end with 'Id' "
            "(e.g., 'SomeId'), or use `allow_custom_name=True`."
        )

    cls._str_prefix = str_prefix

__int__

__int__() -> int

Return the numeric ID of this instance.

Source code in frequenz/core/id.py

def __int__(self) -> int:
    """Return the numeric ID of this instance."""
    return self._id

__lt__

__lt__(other: object) -> bool

Check if this instance is less than another object.

Comparison is only defined between instances of the exact same type.

Source code in frequenz/core/id.py

def __lt__(self, other: object) -> bool:
    """Check if this instance is less than another object.

    Comparison is only defined between instances of the exact same type.
    """
    # pylint: disable-next=unidiomatic-typecheck
    if type(other) is not type(self):
        return NotImplemented
    other_id = cast(Self, other)
    return self._id < other_id._id

__new__

__new__(*_: Any, **__: Any) -> Self

Create a new instance of the ID class, only if it is a subclass of BaseId.

Source code in frequenz/core/id.py

def __new__(cls, *_: Any, **__: Any) -> Self:
    """Create a new instance of the ID class, only if it is a subclass of BaseId."""
    if cls is BaseId:
        raise TypeError("BaseId cannot be instantiated directly. Use a subclass.")
    return super().__new__(cls)

__repr__

__repr__() -> str

Return the string representation of this instance.

Source code in frequenz/core/id.py

def __repr__(self) -> str:
    """Return the string representation of this instance."""
    return f"{type(self).__name__}({self._id!r})"

__str__

__str__() -> str

Return the short string representation of this instance.

Source code in frequenz/core/id.py

def __str__(self) -> str:
    """Return the short string representation of this instance."""
    return f"{self._str_prefix}{self._id}"

frequenz.client.dispatch.types.EvChargerType

Bases: Enum

Enum representing the type of EV charger.

Source code in frequenz/client/dispatch/types.py

class EvChargerType(Enum):
    """Enum representing the type of EV charger."""

    UNSPECIFIED = PBEvChargerType.EV_CHARGER_TYPE_UNSPECIFIED
    """Unspecified type of EV charger."""

    AC = PBEvChargerType.EV_CHARGER_TYPE_AC
    """AC EV charger."""

    DC = PBEvChargerType.EV_CHARGER_TYPE_DC
    """DC EV charger."""

    HYBRID = PBEvChargerType.EV_CHARGER_TYPE_HYBRID
    """Hybrid EV charger."""

Attributes

AC class-attribute instance-attribute

AC = EV_CHARGER_TYPE_AC

AC EV charger.

DC class-attribute instance-attribute

DC = EV_CHARGER_TYPE_DC

DC EV charger.

HYBRID class-attribute instance-attribute

HYBRID = EV_CHARGER_TYPE_HYBRID

Hybrid EV charger.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = EV_CHARGER_TYPE_UNSPECIFIED

Unspecified type of EV charger.

frequenz.client.dispatch.types.Event

Bases: Enum

Enum representing the type of event that occurred during a dispatch operation.

Source code in frequenz/client/dispatch/types.py

class Event(Enum):
    """Enum representing the type of event that occurred during a dispatch operation."""

    UNSPECIFIED = StreamMicrogridDispatchesResponse.Event.EVENT_UNSPECIFIED
    CREATED = StreamMicrogridDispatchesResponse.Event.EVENT_CREATED
    UPDATED = StreamMicrogridDispatchesResponse.Event.EVENT_UPDATED
    DELETED = StreamMicrogridDispatchesResponse.Event.EVENT_DELETED

frequenz.client.dispatch.types.InverterType

Bases: Enum

Enum representing the type of inverter.

Source code in frequenz/client/dispatch/types.py

class InverterType(Enum):
    """Enum representing the type of inverter."""

    UNSPECIFIED = PBInverterType.INVERTER_TYPE_UNSPECIFIED
    """Unspecified type of inverter."""

    BATTERY = PBInverterType.INVERTER_TYPE_BATTERY
    """Battery inverter."""

    SOLAR = PBInverterType.INVERTER_TYPE_SOLAR
    """Solar inverter."""

    HYBRID = PBInverterType.INVERTER_TYPE_HYBRID
    """Hybrid inverter."""

Attributes

BATTERY class-attribute instance-attribute

BATTERY = INVERTER_TYPE_BATTERY

Battery inverter.

HYBRID class-attribute instance-attribute

HYBRID = INVERTER_TYPE_HYBRID

Hybrid inverter.

SOLAR class-attribute instance-attribute

SOLAR = INVERTER_TYPE_SOLAR

Solar inverter.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = INVERTER_TYPE_UNSPECIFIED

Unspecified type of inverter.

frequenz.client.dispatch.types.TargetCategories

Bases: frozenset[TargetCategory]

A set of target component categories and types.

This is a frozen set, so it is immutable.

Source code in frequenz/client/dispatch/types.py

class TargetCategories(frozenset[TargetCategory]):
    """A set of target component categories and types.

    This is a frozen set, so it is immutable.
    """

    def __new__(cls, *categories_input: TargetCategoryInputType) -> Self:
        """Create a new TargetCategories instance.

        Args:
            *categories_input: TargetCategory instances or raw ComponentCategory/specific types
                               (BatteryType, InverterType, EvChargerType) to be wrapped.

        Returns:
            A new TargetCategories instance.

        Raises:
            TypeError: If an item in categories_input is not a TargetCategory
                       nor one of the wrappable types.
        """
        processed_elements = []
        for item in categories_input:
            if isinstance(item, TargetCategory):
                processed_elements.append(item)
            elif isinstance(
                item, (ComponentCategory, BatteryType, InverterType, EvChargerType)
            ):
                # Wrap raw categories/types into TargetCategory instances
                processed_elements.append(TargetCategory(target=item))
            else:
                # This case should ideally be caught by static type checkers
                # if call sites adhere to type hints.
                raise TypeError(
                    f"Invalid type for TargetCategories constructor: {type(item)}. "
                    f"Expected TargetCategory, ComponentCategory, BatteryType, "
                    f"InverterType, or EvChargerType."
                )
        # `super().__new__` for frozenset expects an iterable of elements for the set
        return super().__new__(cls, processed_elements)

    def __repr__(self) -> str:
        """Return an ordered string representation."""
        ordered = sorted(list(self), key=lambda cat: cat.target.value)
        return str([cat.target.name for cat in ordered])

Functions

__new__

__new__(*categories_input: TargetCategoryInputType) -> Self

Create a new TargetCategories instance.

PARAMETER	DESCRIPTION
*categories_input	TargetCategory instances or raw ComponentCategory/specific types (BatteryType, InverterType, EvChargerType) to be wrapped. TYPE: TargetCategoryInputType DEFAULT: ()

RETURNS	DESCRIPTION
Self	A new TargetCategories instance.

RAISES	DESCRIPTION
TypeError	If an item in categories_input is not a TargetCategory nor one of the wrappable types.

Source code in frequenz/client/dispatch/types.py

def __new__(cls, *categories_input: TargetCategoryInputType) -> Self:
    """Create a new TargetCategories instance.

    Args:
        *categories_input: TargetCategory instances or raw ComponentCategory/specific types
                           (BatteryType, InverterType, EvChargerType) to be wrapped.

    Returns:
        A new TargetCategories instance.

    Raises:
        TypeError: If an item in categories_input is not a TargetCategory
                   nor one of the wrappable types.
    """
    processed_elements = []
    for item in categories_input:
        if isinstance(item, TargetCategory):
            processed_elements.append(item)
        elif isinstance(
            item, (ComponentCategory, BatteryType, InverterType, EvChargerType)
        ):
            # Wrap raw categories/types into TargetCategory instances
            processed_elements.append(TargetCategory(target=item))
        else:
            # This case should ideally be caught by static type checkers
            # if call sites adhere to type hints.
            raise TypeError(
                f"Invalid type for TargetCategories constructor: {type(item)}. "
                f"Expected TargetCategory, ComponentCategory, BatteryType, "
                f"InverterType, or EvChargerType."
            )
    # `super().__new__` for frozenset expects an iterable of elements for the set
    return super().__new__(cls, processed_elements)

__repr__

__repr__() -> str

Return an ordered string representation.

Source code in frequenz/client/dispatch/types.py

def __repr__(self) -> str:
    """Return an ordered string representation."""
    ordered = sorted(list(self), key=lambda cat: cat.target.value)
    return str([cat.target.name for cat in ordered])

frequenz.client.dispatch.types.TargetCategory dataclass

Represents a category and optionally a type.

Source code in frequenz/client/dispatch/types.py

@dataclass(frozen=True)
class TargetCategory:
    """Represents a category and optionally a type."""

    target: ComponentCategory | BatteryType | EvChargerType | InverterType
    """The target category of the dispatch.

    Implicitly derived from the types.
    """

    @property
    def category(self) -> ComponentCategory:
        """Get the category of the target.

        Returns:
            The category of the target.
        """
        match self.target:
            case ComponentCategory():
                return self.target
            case BatteryType():
                return ComponentCategory.BATTERY
            case EvChargerType():
                return ComponentCategory.EV_CHARGER
            case InverterType():
                return ComponentCategory.INVERTER

    @property
    def type(self) -> BatteryType | EvChargerType | InverterType | None:
        """Get the type of the category.

        Returns:
            The type of the category.
        """
        match self.target:
            case BatteryType() | EvChargerType() | InverterType():
                return self.target
            case _:
                return None

Attributes

category property

category: ComponentCategory

Get the category of the target.

RETURNS	DESCRIPTION
ComponentCategory	The category of the target.

target instance-attribute

target: (
    ComponentCategory
    | BatteryType
    | EvChargerType
    | InverterType
)

The target category of the dispatch.

Implicitly derived from the types.

type property

type: BatteryType | EvChargerType | InverterType | None

Get the type of the category.

RETURNS	DESCRIPTION
BatteryType | EvChargerType | InverterType | None	The type of the category.

frequenz.client.dispatch.types.TargetIds

Bases: frozenset[int]

A set of target component IDs.

This is a frozen set, so it is immutable.

Source code in frequenz/client/dispatch/types.py

class TargetIds(frozenset[int]):
    """A set of target component IDs.

    This is a frozen set, so it is immutable.
    """

    def __new__(cls, *ids: SupportsInt) -> Self:
        """Create a new TargetIds instance.

        Args:
            *ids: The target IDs to initialize.

        Returns:
            A new TargetIds instance.
        """
        # Convert all provided ids to integers before creating the frozenset
        processed_ids = tuple(int(id_val) for id_val in ids)
        return super().__new__(cls, processed_ids)

Functions

__new__

__new__(*ids: SupportsInt) -> Self

Create a new TargetIds instance.

PARAMETER	DESCRIPTION
*ids	The target IDs to initialize. TYPE: SupportsInt DEFAULT: ()

RETURNS	DESCRIPTION
Self	A new TargetIds instance.

Source code in frequenz/client/dispatch/types.py

def __new__(cls, *ids: SupportsInt) -> Self:
    """Create a new TargetIds instance.

    Args:
        *ids: The target IDs to initialize.

    Returns:
        A new TargetIds instance.
    """
    # Convert all provided ids to integers before creating the frozenset
    processed_ids = tuple(int(id_val) for id_val in ids)
    return super().__new__(cls, processed_ids)

frequenz.client.dispatch.types.TimeIntervalFilter dataclass

Filter for a time interval.

Source code in frequenz/client/dispatch/types.py

@dataclass(frozen=True, kw_only=True)
class TimeIntervalFilter:
    """Filter for a time interval."""

    start_from: datetime | None
    """Filter by start_time >= start_from."""

    start_to: datetime | None
    """Filter by start_time < start_to."""

    end_from: datetime | None
    """Filter by end_time >= end_from."""

    end_to: datetime | None
    """Filter by end_time < end_to."""

Attributes

end_from instance-attribute

end_from: datetime | None

Filter by end_time >= end_from.

end_to instance-attribute

end_to: datetime | None

Filter by end_time < end_to.

start_from instance-attribute

start_from: datetime | None

Filter by start_time >= start_from.

start_to instance-attribute

start_to: datetime | None

Filter by start_time < start_to.

Functions