State Machine

Centralized management and storage of all entity states in Home Assistant

The State Machine is Home Assistant’s central storage system for all entity states. It maintains a complete picture of your home automation system’s current state and efficiently tracks changes over time.

StateMachine Overview

The StateMachine class provides thread-safe access to entity states:

homeassistant/core.py

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

The StateMachine maintains two references to state storage: _states (the States container) and _states_data (direct dict access) for performance optimization.

State Object Structure

Each entity’s state is represented by a State object:

homeassistant/core.py

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

State Components

entity_id

Unique identifier in format domain.object_id

state

String representing the current state value

attributes

Dictionary of additional state metadata

timestamps

Three timestamps tracking different types of changes

Setting States

Basic State Setting

# Async context (preferred)
hass.states.async_set(
    "sensor.temperature",
    "23.5",
    {"unit_of_measurement": "°C", "friendly_name": "Temperature"}
)

# Synchronous context (blocks until complete)
hass.states.set(
    "sensor.temperature",
    "23.5",
    {"unit_of_measurement": "°C", "friendly_name": "Temperature"}
)

Advanced State Setting

homeassistant/core.py

@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(),
    )

Parameters

• entity_id: Entity identifier (automatically lowercased)
• new_state: New state value (converted to string)
• attributes: Optional dictionary of attributes
• force_update: Force last_changed update even if state unchanged
• context: Context tracking the change origin
• timestamp: Override the timestamp (for replay scenarios)

1. Entity ID Normalized

   The entity_id is converted to lowercase for consistency.

2. State Validated

   The state string length is validated (max 255 characters).

3. Old State Retrieved

   The existing state is retrieved to compare changes.

4. State Object Created

   A new State object is created with the updated values.

5. Event Fired

   Either EVENT_STATE_CHANGED or EVENT_STATE_REPORTED is fired.

State Change Detection

The state machine intelligently detects different types of state changes:

homeassistant/core.py

@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,
    )

Event Types

STATE_CHANGED Event

# Fired when state value or attributes change
{
    "entity_id": "light.living_room",
    "old_state": State(entity_id="light.living_room", state="off", ...),
    "new_state": State(entity_id="light.living_room", state="on", ...)
}

EVENT_STATE_REPORTED was introduced to allow tracking of entity updates even when the state value doesn’t change. This is useful for detecting that a device is still communicating.

Reading States

Getting a Single State

# Get state (returns None if not found)
state = hass.states.get("light.living_room")

if state:
    print(f"State: {state.state}")
    print(f"Attributes: {state.attributes}")
    print(f"Last changed: {state.last_changed}")

Checking State Value

# Check if entity is in a specific state
if hass.states.is_state("light.living_room", "on"):
    print("Light is on")

Getting All States

# Get all states (async context)
all_states = hass.states.async_all()

# Get states for a specific domain
light_states = hass.states.async_all("light")

# Get states for multiple domains  
states = hass.states.async_all(["light", "switch"])

Getting Entity IDs

# Get all entity IDs
all_entity_ids = hass.states.async_entity_ids()

# Get entity IDs for a domain
light_entity_ids = hass.states.async_entity_ids("light")

# Get count of entities
total_entities = hass.states.async_entity_ids_count()
light_count = hass.states.async_entity_ids_count("light")

Removing States

Remove an entity from the state machine:

# Async context
removed = hass.states.async_remove("sensor.old_sensor")

if removed:
    print("Entity removed")
else:
    print("Entity did not exist")

# Synchronous context
removed = hass.states.remove("sensor.old_sensor")

Removing a state fires a STATE_CHANGED event with new_state=None. This signals to listeners that the entity no longer exists.

State Reservations

Reserve an entity_id before creating it to prevent race conditions:

from homeassistant.exceptions import HomeAssistantError

try:
    # Reserve the entity_id
    hass.states.async_reserve("sensor.new_sensor")
    
    # Perform setup that might take time
    await setup_sensor()
    
    # Set the initial state
    hass.states.async_set(
        "sensor.new_sensor",
        "initializing",
        {"friendly_name": "New Sensor"}
    )
except HomeAssistantError:
    _LOGGER.error("Entity ID already in use")

Reservations prevent multiple components from trying to create the same entity_id simultaneously, which could cause data corruption or unexpected behavior.

States Container

The States class provides efficient domain-based indexing:

homeassistant/core.py

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()

The domain index enables O(1) lookups when retrieving all entities of a specific domain, making operations like hass.states.async_all("light") very fast even with thousands of entities.

State Attributes

Attributes provide additional context about an entity’s state:

# Common attributes
attributes = {
    "friendly_name": "Living Room Temperature",
    "unit_of_measurement": "°C",
    "device_class": "temperature",
    "icon": "mdi:thermometer",
    
    # Location attributes
    "latitude": 37.7749,
    "longitude": -122.4194,
    
    # Device information
    "manufacturer": "Acme Corp",
    "model": "TH-100",
    "sw_version": "1.2.3",
}

hass.states.async_set(
    "sensor.temperature",
    "23.5",
    attributes
)

Attribute Best Practices

• Use standard attribute names defined in homeassistant.const
• Keep attribute values JSON-serializable
• Don’t store large data structures in attributes
• Use ReadOnlyDict to prevent accidental modification

Timestamp Semantics

The State object maintains three timestamps:

last_changed

Updated when the state value changes:

# Initial state
hass.states.async_set("sensor.temp", "20", {"unit": "°C"})
# last_changed = now

# Update attributes only
hass.states.async_set("sensor.temp", "20", {"unit": "°C", "battery": 90})
# last_changed = unchanged

# Update state value
hass.states.async_set("sensor.temp", "21", {"unit": "°C", "battery": 90})
# last_changed = now

last_updated

Updated when the state value or attributes change:

# Update attributes
hass.states.async_set("sensor.temp", "20", {"battery": 90})
# last_updated = now, last_changed = unchanged

last_reported

Updated every time async_set is called, even if nothing changed:

# Report same state
hass.states.async_set("sensor.temp", "20", {"unit": "°C"})
# last_reported = now
# Fires EVENT_STATE_REPORTED (not EVENT_STATE_CHANGED)

Use last_reported to detect if a device is still communicating, even when its state hasn’t changed. This is particularly useful for detecting unavailable devices.

State Serialization

States can be serialized for storage or API responses:

# Get as dictionary
state_dict = state.as_dict()
# Returns ReadOnlyDict with ISO-formatted timestamps

# Get as JSON bytes
state_json = state.as_dict_json

# Get compressed state (for WebSocket API)
compressed = state.as_compressed_state
# Omits last_updated if equal to last_changed
# Compresses context if simple

Compressed State Format

homeassistant/core.py

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)

Performance Optimizations

The state machine includes several optimizations:

Direct dict access: Internal code uses _states_data to bypass container overhead
Cached properties: Expensive computations are cached with @cached_property
Domain indexing: O(1) lookups for domain-filtered queries
Timestamp caching: Timestamp conversions are cached on State objects

Best Practices

1. Use lowercase entity_ids: The state machine automatically lowercases, but it’s faster if you do it yourself
2. Preserve contexts: Pass contexts through automation chains for proper tracking
3. Validate before setting: Check entity_id format before calling async_set
4. Use async methods: Always prefer async_set over set when in the event loop
5. Handle None states: When listening to state changes, old_state or new_state may be None