State Machine

class StateMachine:
    """Helper class that tracks the state of different entities."""
    
    def __init__(self, bus: EventBus, loop: asyncio.events.AbstractEventLoop) -> None:
        """Initialize state machine."""
        self._states = States()
        self._states_data = self._states.data
        self._reservations: set[str] = set()
        self._bus = bus
        self._loop = loop

class State:
    """Object to represent a state within the state machine."""
    
    entity_id: str                    # Entity identifier (e.g., "light.living_room")
    domain: str                       # Domain portion (e.g., "light")
    object_id: str                    # Object ID portion (e.g., "living_room")
    state: str                        # Current state value (e.g., "on", "23.5", "home")
    attributes: ReadOnlyDict[str, Any]  # Additional state information
    last_changed: datetime.datetime   # When state value last changed
    last_reported: datetime.datetime  # When state was last reported
    last_updated: datetime.datetime   # When state or attributes last changed
    context: Context                  # Context that created this state

@callback
def async_set(
    self,
    entity_id: str,
    new_state: str,
    attributes: Mapping[str, Any] | None = None,
    force_update: bool = False,
    context: Context | None = None,
    state_info: StateInfo | None = None,
    timestamp: float | None = None,
) -> None:
    """Set the state of an entity, add entity if it does not exist.
    
    If you just update the attributes and not the state, last_changed
    will not be affected.
    """
    state = str(new_state)
    validate_state(state)
    self.async_set_internal(
        entity_id.lower(),
        state,
        attributes or {},
        force_update,
        context,
        state_info,
        timestamp or time.time(),
    )

@callback
def async_set_internal(
    self,
    entity_id: str,
    new_state: str,
    attributes: Mapping[str, Any] | None,
    force_update: bool,
    context: Context | None,
    state_info: StateInfo | None,
    timestamp: float,
) -> None:
    """Set the state of an entity."""
    old_state: State | None
    try:
        old_state = self._states_data[entity_id]
    except KeyError:
        old_state = None
        same_state = False
        same_attr = False
        last_changed = None
    else:
        same_state = old_state.state == new_state and not force_update
        same_attr = old_state.attributes == attributes
        last_changed = old_state.last_changed if same_state else None
    
    if same_state and same_attr:
        # State unchanged - update last_reported and fire STATE_REPORTED
        old_state.last_reported = now
        self._bus.async_fire_internal(
            EVENT_STATE_REPORTED,
            {
                "entity_id": entity_id,
                "last_reported": now,
                "old_last_reported": old_last_reported,
                "new_state": old_state,
            },
            context=context,
        )
        return
    
    # State or attributes changed - create new state and fire STATE_CHANGED
    state = State(
        entity_id,
        new_state,
        attributes,
        last_changed,
        now,
        now,
        context,
        old_state is None,
        state_info,
        timestamp,
    )
    self._states[entity_id] = state
    self._bus.async_fire_internal(
        EVENT_STATE_CHANGED,
        {
            "entity_id": entity_id,
            "old_state": old_state,
            "new_state": state,
        },
        context=context,
    )

class States(UserDict[str, State]):
    """Container for states, maps entity_id -> State.
    
    Maintains an additional index:
    - domain -> dict[str, State]
    """
    
    def __init__(self) -> None:
        super().__init__()
        self._domain_index: defaultdict[str, dict[str, State]] = defaultdict(dict)
    
    def __setitem__(self, key: str, entry: State) -> None:
        """Add an item."""
        self.data[key] = entry
        self._domain_index[entry.domain][entry.entity_id] = entry
    
    def domain_entity_ids(self, key: str) -> KeysView[str] | tuple[()]:
        """Get all entity_ids for a domain."""
        if key not in self._domain_index:
            return ()
        return self._domain_index[key].keys()

class CompressedState(TypedDict):
    """Compressed dict of a state."""
    
    s: str                         # state
    a: ReadOnlyDict[str, Any]      # attributes  
    c: str | dict[str, Any]        # context (string if simple)
    lc: float                      # last_changed timestamp
    lu: NotRequired[float]         # last_updated (omitted if == lc)