Improve typing of State.as_compressed_state (#113540)

pull/113583/head
Erik Montnemery 2024-03-16 05:01:46 +01:00 committed by GitHub
parent 27e844e3bf
commit a5cde8a61e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 29 additions and 8 deletions

View File

@ -1456,11 +1456,11 @@ _DEPRECATED_DATA_RATE_GIBIBYTES_PER_SECOND: Final = DeprecatedConstantEnum(
# States # States
COMPRESSED_STATE_STATE = "s" COMPRESSED_STATE_STATE: Final = "s"
COMPRESSED_STATE_ATTRIBUTES = "a" COMPRESSED_STATE_ATTRIBUTES: Final = "a"
COMPRESSED_STATE_CONTEXT = "c" COMPRESSED_STATE_CONTEXT: Final = "c"
COMPRESSED_STATE_LAST_CHANGED = "lc" COMPRESSED_STATE_LAST_CHANGED: Final = "lc"
COMPRESSED_STATE_LAST_UPDATED = "lu" COMPRESSED_STATE_LAST_UPDATED: Final = "lu"
# #### SERVICES #### # #### SERVICES ####
SERVICE_TURN_ON: Final = "turn_on" SERVICE_TURN_ON: Final = "turn_on"

View File

@ -31,7 +31,18 @@ import re
import threading import threading
import time import time
from time import monotonic from time import monotonic
from typing import TYPE_CHECKING, Any, Generic, Literal, ParamSpec, Self, cast, overload from typing import (
TYPE_CHECKING,
Any,
Generic,
Literal,
NotRequired,
ParamSpec,
Self,
TypedDict,
cast,
overload,
)
from urllib.parse import urlparse from urllib.parse import urlparse
from typing_extensions import TypeVar from typing_extensions import TypeVar
@ -1534,6 +1545,16 @@ class EventBus:
) )
class CompressedState(TypedDict):
"""Compressed dict of a state."""
s: str # COMPRESSED_STATE_STATE
a: ReadOnlyDict[str, Any] # COMPRESSED_STATE_ATTRIBUTES
c: str | dict[str, Any] # COMPRESSED_STATE_CONTEXT
lc: float # COMPRESSED_STATE_LAST_CHANGED
lu: NotRequired[float] # COMPRESSED_STATE_LAST_UPDATED
class State: class State:
"""Object to represent a state within the state machine. """Object to represent a state within the state machine.
@ -1663,7 +1684,7 @@ class State:
return json_fragment(self.as_dict_json) return json_fragment(self.as_dict_json)
@cached_property @cached_property
def as_compressed_state(self) -> dict[str, Any]: def as_compressed_state(self) -> CompressedState:
"""Build a compressed dict of a state for adds. """Build a compressed dict of a state for adds.
Omits the lu (last_updated) if it matches (lc) last_changed. Omits the lu (last_updated) if it matches (lc) last_changed.
@ -1678,7 +1699,7 @@ class State:
# to avoid callers outside of this module # to avoid callers outside of this module
# from misusing it by mistake. # from misusing it by mistake.
context = state_context._as_dict # pylint: disable=protected-access context = state_context._as_dict # pylint: disable=protected-access
compressed_state = { compressed_state: CompressedState = {
COMPRESSED_STATE_STATE: self.state, COMPRESSED_STATE_STATE: self.state,
COMPRESSED_STATE_ATTRIBUTES: self.attributes, COMPRESSED_STATE_ATTRIBUTES: self.attributes,
COMPRESSED_STATE_CONTEXT: context, COMPRESSED_STATE_CONTEXT: context,