Skip to content

Conversion

AnnotationConverter

Converts runtime values to match rich annotations (including nested containers).

Source code in src/hassette/conversion/annotation_converter.py
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class AnnotationConverter:
    """Converts runtime values to match rich annotations (including nested containers)."""

    def convert(self, value: Any, annotation: Any) -> Any:
        # TODO(#892): break circular dependency — conversion/__init__ imports this module
        from hassette.conversion import TYPE_MATCHER, TYPE_REGISTRY
        from hassette.models.states import BaseState

        tp = normalize_annotation(annotation, constructible=True)

        # Already correct (deep) => no-op
        if TYPE_MATCHER.matches(value, tp):
            return value

        # Union / Optional: try arms
        if is_union(tp):
            last_err: Exception | None = None
            for arm in get_args(tp):
                try:
                    return self.convert(value, arm)
                except Exception as e:
                    last_err = e
            raise UnableToConvertValueError(f"Unable to convert {value!r} to {tp!r}") from last_err

        # Literal: match exact allowed values (conversion generally not meaningful)
        if get_origin(tp) is Literal:
            allowed = get_args(tp)
            if value in allowed:
                return value
            raise UnableToConvertValueError(f"{value!r} is not one of {allowed!r}")

        origin = get_origin(tp)
        if origin is not None:
            entry = ContainerConverterRegistry.get(origin)
            if entry is not None:
                return entry.convert(self, value, tp)

        # Special case: BaseState subclass conversion from dict
        if isinstance(tp, type) and issubclass(tp, BaseState):
            # NOTE: this expects tp is the concrete model class (LightState, etc.)
            return convert_state_dict_to_model(value, tp)

        # Leaf conversion: tp should be a runtime type
        if isinstance(tp, type):
            return TYPE_REGISTRY.convert(value, tp)

        # Fall back: try converting to origin if that's a runtime type
        if origin is not None and isinstance(origin, type):
            return TYPE_REGISTRY.convert(value, origin)

        raise UnableToConvertValueError(f"Unable to convert {value!r} to {tp!r}")

StateKey dataclass

Source code in src/hassette/conversion/state_registry.py
22
23
24
25
26
27
28
@dataclass(frozen=True)
class StateKey:
    domain: Hashable | None = None
    """The domain of the entity (e.g., 'light', 'sensor')."""

    device_class: Hashable | None = None
    """Optional device class of the entity (e.g., 'temperature', 'humidity'). Not yet being used."""

domain: Hashable | None = None class-attribute instance-attribute

The domain of the entity (e.g., 'light', 'sensor').

device_class: Hashable | None = None class-attribute instance-attribute

Optional device class of the entity (e.g., 'temperature', 'humidity'). Not yet being used.

StateRegistry

Registry for mapping domains to their state classes.

This class maintains a mapping of Home Assistant domains to their corresponding BaseState subclasses. State classes get registered during the after_initialize phase by scanning all subclasses of BaseState.

Source code in src/hassette/conversion/state_registry.py
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class StateRegistry:
    """Registry for mapping domains to their state classes.

    This class maintains a mapping of Home Assistant domains to their corresponding
    BaseState subclasses. State classes get registered during the `after_initialize` phase
    by scanning all subclasses of BaseState.
    """

    _registry: ClassVar[dict[StateKey, type["BaseState"]]] = {}

    def _get_entity_id(self, data: "HassStateDict", entity_id: str | None = None) -> str:
        if not entity_id:
            # specifically this way so we also handle empty strings/None
            entity_id = data.get("entity_id") or "<unknown>"

        if not isinstance(entity_id, str):
            LOGGER.error("State data has invalid 'entity_id' field: %s", data, stacklevel=2)
            raise InvalidEntityIdError(entity_id)

        if "." not in entity_id:
            LOGGER.error("State data has malformed 'entity_id' (missing domain): %s", entity_id, stacklevel=2)
            raise InvalidEntityIdError(entity_id)

        return entity_id

    def try_convert_state(self, data: "HassStateDict", entity_id: str | None = None) -> "BaseState":
        """Convert a dictionary representation of a state into a specific state type.

        This function uses the state registry to look up the appropriate state class
        based on the entity's domain. If no specific class is registered for the domain,
        it falls back to the generic BaseState.

        Args:
            data: Dictionary containing state data from Home Assistant.
            entity_id: Optional entity ID to assist in domain determination.

        Returns:
            A properly typed state object (e.g., LightState, SensorState) or BaseState
            for unknown domains.

        Raises:
            InvalidDataForStateConversionError: If the provided data is invalid or malformed.
            InvalidEntityIdError: If the entity_id is invalid or malformed.
            UnableToConvertStateError: If conversion to the determined state class fails.

        Example:
            ```python
            state_dict = {"entity_id": "light.bedroom", "state": "on", ...}
            light_state = try_convert_state(state_dict)  # Returns LightState instance
            ```
        """
        from hassette.models.states.base import BaseState

        if "event" in data:
            LOGGER.error(
                "Data contains 'event' key, expected state data, not event data. "
                "To convert state from an event, extract the state data from event.payload.data.new_state "
                "or event.payload.data.old_state.",
                stacklevel=2,
            )
            raise InvalidDataForStateConversionError(data)

        entity_id = self._get_entity_id(data, entity_id=entity_id)
        domain = entity_id.split(".", 1)[0]

        # Look up the appropriate state class from the registry
        state_class = self.resolve(domain=domain)

        classes = [state_class, BaseState] if state_class is not None else [BaseState]

        final_idx = len(classes) - 1
        for i, cls in enumerate(classes):
            try:
                return self._conversion_with_error_handling(cls, data, entity_id, domain)
            except UnableToConvertStateError:
                if i == final_idx:
                    raise
                LOGGER.debug(
                    "Falling back to next state class after failure to convert to '%s' for entity '%s'",
                    cls.__name__,
                    entity_id,
                )

        raise RuntimeError("Unreachable code reached in try_convert_state")

    @classmethod
    def register(
        cls, state_class: type["BaseState"], *, domain: Hashable | None = None, device_class: Hashable | None = None
    ) -> None:
        """Register a state class for a given domain and optional device_class combination.

        Args:
            state_class: The state class to register. Must be a subclass of BaseState.
            domain: The Home Assistant domain (e.g., "light", "sensor"). If None, matches any domain.
            device_class: The device class (e.g., "temperature", "motion"). If None, matches any device class.
        """
        key = StateKey(domain=domain, device_class=device_class)
        cls._registry[key] = state_class

    @classmethod
    def resolve(
        cls, *, domain: Hashable | None = None, device_class: Hashable | None = None
    ) -> type["BaseState"] | None:
        """Resolve a state class from the registry based on domain and device_class."""
        candidates = [StateKey(domain=domain, device_class=device_class)]
        if device_class is not None:
            candidates.append(StateKey(domain=domain, device_class=None))

        for k in candidates:
            if k in cls._registry:
                return cls._registry[k]
        return None

    def _conversion_with_error_handling(
        self, state_class: type["BaseState"], data: "HassStateDict", entity_id: str, domain: str
    ) -> "BaseState":
        """Convert state data, logging and re-raising as UnableToConvertStateError on failure."""

        class_name = state_class.__name__
        truncated_data = repr(data)
        if len(truncated_data) > STATE_REPR_MAX_LENGTH:
            truncated_data = truncated_data[:STATE_REPR_MAX_LENGTH] + "...[truncated]"

        try:
            return convert_state_dict_to_model(data, state_class)
        except Exception as e:
            tb = get_short_traceback()

            LOGGER.error(
                CONVERSION_FAIL_TEMPLATE,
                entity_id,
                domain,
                class_name,
                truncated_data,
                e,
                tb,
            )
            raise UnableToConvertStateError(entity_id, state_class) from e

    def __contains__(self, model: type["BaseState"]) -> bool:
        """Check if the registry contains a state class for the given model."""
        return any(cls is model for cls in self._registry.values())

    def __iter__(self) -> Iterator[tuple[StateKey, type["BaseState"]]]:
        """Iterate over all registered state classes with their keys."""
        return iter(self._registry.items())

    def items(self) -> Iterator[tuple[StateKey, type["BaseState"]]]:
        return iter(self._registry.items())

    def values(self) -> Iterator[type["BaseState"]]:
        return (state_class for state_class in self._registry.values())

    def keys(self) -> Iterator[StateKey]:
        return (key for key in self._registry)

    @classmethod
    def snapshot(cls) -> dict[StateKey, type["BaseState"]]:
        """Return a shallow copy of the current registry state."""
        return dict(cls._registry)

    @classmethod
    def restore(cls, snapshot: dict[StateKey, type["BaseState"]]) -> None:
        """Replace the registry with a previously captured snapshot."""
        cls._registry = snapshot

try_convert_state(data: HassStateDict, entity_id: str | None = None) -> BaseState

Convert a dictionary representation of a state into a specific state type.

This function uses the state registry to look up the appropriate state class based on the entity's domain. If no specific class is registered for the domain, it falls back to the generic BaseState.

Parameters:

Name Type Description Default
data HassStateDict

Dictionary containing state data from Home Assistant.

required
entity_id str | None

Optional entity ID to assist in domain determination.

None

Returns:

Type Description
BaseState

A properly typed state object (e.g., LightState, SensorState) or BaseState

BaseState

for unknown domains.

Raises:

Type Description
InvalidDataForStateConversionError

If the provided data is invalid or malformed.

InvalidEntityIdError

If the entity_id is invalid or malformed.

UnableToConvertStateError

If conversion to the determined state class fails.

Example
state_dict = {"entity_id": "light.bedroom", "state": "on", ...}
light_state = try_convert_state(state_dict)  # Returns LightState instance
Source code in src/hassette/conversion/state_registry.py
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def try_convert_state(self, data: "HassStateDict", entity_id: str | None = None) -> "BaseState":
    """Convert a dictionary representation of a state into a specific state type.

    This function uses the state registry to look up the appropriate state class
    based on the entity's domain. If no specific class is registered for the domain,
    it falls back to the generic BaseState.

    Args:
        data: Dictionary containing state data from Home Assistant.
        entity_id: Optional entity ID to assist in domain determination.

    Returns:
        A properly typed state object (e.g., LightState, SensorState) or BaseState
        for unknown domains.

    Raises:
        InvalidDataForStateConversionError: If the provided data is invalid or malformed.
        InvalidEntityIdError: If the entity_id is invalid or malformed.
        UnableToConvertStateError: If conversion to the determined state class fails.

    Example:
        ```python
        state_dict = {"entity_id": "light.bedroom", "state": "on", ...}
        light_state = try_convert_state(state_dict)  # Returns LightState instance
        ```
    """
    from hassette.models.states.base import BaseState

    if "event" in data:
        LOGGER.error(
            "Data contains 'event' key, expected state data, not event data. "
            "To convert state from an event, extract the state data from event.payload.data.new_state "
            "or event.payload.data.old_state.",
            stacklevel=2,
        )
        raise InvalidDataForStateConversionError(data)

    entity_id = self._get_entity_id(data, entity_id=entity_id)
    domain = entity_id.split(".", 1)[0]

    # Look up the appropriate state class from the registry
    state_class = self.resolve(domain=domain)

    classes = [state_class, BaseState] if state_class is not None else [BaseState]

    final_idx = len(classes) - 1
    for i, cls in enumerate(classes):
        try:
            return self._conversion_with_error_handling(cls, data, entity_id, domain)
        except UnableToConvertStateError:
            if i == final_idx:
                raise
            LOGGER.debug(
                "Falling back to next state class after failure to convert to '%s' for entity '%s'",
                cls.__name__,
                entity_id,
            )

    raise RuntimeError("Unreachable code reached in try_convert_state")

register(state_class: type[BaseState], *, domain: Hashable | None = None, device_class: Hashable | None = None) -> None classmethod

Register a state class for a given domain and optional device_class combination.

Parameters:

Name Type Description Default
state_class type[BaseState]

The state class to register. Must be a subclass of BaseState.

required
domain Hashable | None

The Home Assistant domain (e.g., "light", "sensor"). If None, matches any domain.

None
device_class Hashable | None

The device class (e.g., "temperature", "motion"). If None, matches any device class.

None
Source code in src/hassette/conversion/state_registry.py
123
124
125
126
127
128
129
130
131
132
133
134
135
@classmethod
def register(
    cls, state_class: type["BaseState"], *, domain: Hashable | None = None, device_class: Hashable | None = None
) -> None:
    """Register a state class for a given domain and optional device_class combination.

    Args:
        state_class: The state class to register. Must be a subclass of BaseState.
        domain: The Home Assistant domain (e.g., "light", "sensor"). If None, matches any domain.
        device_class: The device class (e.g., "temperature", "motion"). If None, matches any device class.
    """
    key = StateKey(domain=domain, device_class=device_class)
    cls._registry[key] = state_class

resolve(*, domain: Hashable | None = None, device_class: Hashable | None = None) -> type[BaseState] | None classmethod

Resolve a state class from the registry based on domain and device_class.

Source code in src/hassette/conversion/state_registry.py
137
138
139
140
141
142
143
144
145
146
147
148
149
@classmethod
def resolve(
    cls, *, domain: Hashable | None = None, device_class: Hashable | None = None
) -> type["BaseState"] | None:
    """Resolve a state class from the registry based on domain and device_class."""
    candidates = [StateKey(domain=domain, device_class=device_class)]
    if device_class is not None:
        candidates.append(StateKey(domain=domain, device_class=None))

    for k in candidates:
        if k in cls._registry:
            return cls._registry[k]
    return None

__contains__(model: type[BaseState]) -> bool

Check if the registry contains a state class for the given model.

Source code in src/hassette/conversion/state_registry.py
177
178
179
def __contains__(self, model: type["BaseState"]) -> bool:
    """Check if the registry contains a state class for the given model."""
    return any(cls is model for cls in self._registry.values())

__iter__() -> Iterator[tuple[StateKey, type[BaseState]]]

Iterate over all registered state classes with their keys.

Source code in src/hassette/conversion/state_registry.py
181
182
183
def __iter__(self) -> Iterator[tuple[StateKey, type["BaseState"]]]:
    """Iterate over all registered state classes with their keys."""
    return iter(self._registry.items())

snapshot() -> dict[StateKey, type[BaseState]] classmethod

Return a shallow copy of the current registry state.

Source code in src/hassette/conversion/state_registry.py
194
195
196
197
@classmethod
def snapshot(cls) -> dict[StateKey, type["BaseState"]]:
    """Return a shallow copy of the current registry state."""
    return dict(cls._registry)

restore(snapshot: dict[StateKey, type[BaseState]]) -> None classmethod

Replace the registry with a previously captured snapshot.

Source code in src/hassette/conversion/state_registry.py
199
200
201
202
@classmethod
def restore(cls, snapshot: dict[StateKey, type["BaseState"]]) -> None:
    """Replace the registry with a previously captured snapshot."""
    cls._registry = snapshot

TypeMatcher

Runtime matcher for checking if values already satisfy nested type annotations.

Source code in src/hassette/conversion/type_matcher.py
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class TypeMatcher:
    """Runtime matcher for checking if values already satisfy nested type annotations."""

    def matches(self, value: Any, tp: Any) -> bool:
        tp = normalize_annotation(tp)

        # Any matches everything
        if tp is Any:
            return True

        # Check optional prior to union to short-circuit
        if is_optional(tp):
            return value is None or any(self.matches(value, arg) for arg in get_args(tp) if arg is not NoneType)

        # Union: only return "True" if it matches the first type
        # otherwise we risk saying "sure, '6' matches [int | str]" when the caller really wanted an int
        if is_union(tp):
            args = get_args(tp)
            if not args:
                return False
            return self.matches(value, args[0])

        # Literal[...] is *very* useful for DI-style validation
        origin = get_origin(tp)
        if origin is Literal:
            return value in get_args(tp)

        # Non-parameterized type: use normalize_for_isinstance
        if origin is None:
            with suppress(TypeError):
                norm = normalize_for_isinstance(tp)
                return isinstance(value, norm)
            return False

        # Parameterized type: must satisfy outer container/protocol first
        with suppress(TypeError):
            if not isinstance(value, origin):
                return False
        # If origin can't be used with isinstance, treat as non-match (forces conversion).
        if not safe_isinstance(value, origin):
            return False

        # Deep match for known container origins
        entry = TypeMatcherRegistry.get(origin)
        if entry is not None:
            return entry.match_fn(self, value, tp)

        # Unknown generic: outer match is the best we can do
        return True

TypeConverterEntry dataclass

Bases: Generic[T, R]

Represents a type conversion function and its associated metadata.

Source code in src/hassette/conversion/type_registry.py
40
41
42
43
44
45
46
47
48
@dataclass
class TypeConverterEntry(Generic[T, R]):
    """Represents a type conversion function and its associated metadata."""

    func: Callable[[T], R]
    from_type: type[T]
    to_type: type[R]
    error_types: tuple[type[BaseException], ...] = (ValueError,)
    error_message: str | None = None

TypeRegistry

Registry for converting between types, used by State models and the Dependency Injection system.

Source code in src/hassette/conversion/type_registry.py
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class TypeRegistry:
    """Registry for converting between types, used by State models and the Dependency Injection system."""

    conversion_map: ClassVar[dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]] = {}

    @classmethod
    def register(cls, type_converter: TypeConverterEntry[Any, Any]) -> None:
        """Register a type converter in the registry."""
        from_type = type_converter.from_type
        to_type = type_converter.to_type
        key = (from_type, to_type)
        if key in cls.conversion_map:
            LOGGER.warning("Overwriting existing conversion from %s to %s", from_type.__name__, to_type.__name__)
        cls.conversion_map[key] = type_converter

    def convert(self, value: Any, to_type: type[Any] | tuple[type[Any], ...]) -> Any:
        """Convert a StateValue to a target Python type.

        Args:
            value: The StateValue instance to convert.
            to_type: The target Python type.

        Returns:
            The converted value.
        """

        # handle tuple
        if isinstance(to_type, tuple):
            if isinstance(value, to_type):
                return value

            for tt in to_type:
                with suppress(UnableToConvertValueError):
                    return self.convert(value, tt)
            raise UnableToConvertValueError(f"Unable to convert {value!r} to any of the types {to_type}")

        # handle single type

        from_type = type(value)
        key = (from_type, to_type)

        # handle Any type
        if to_type is type(Any):
            return value

        # handle exact type match
        if to_type is from_type:
            return value

        # handle None value
        if value is None:
            return value

        if to_type is type(None) and value is not None:
            LOGGER.debug("Not attempting to convert %r to NoneType", value)
            raise UnableToConvertValueError(f"Cannot convert {value!r} to NoneType")

        # if we don't have this in our map, attempt to just convert using it as a constructor
        if key not in self.conversion_map and to_type is not type(None):
            try:
                new_value = to_type(value)
            except Exception as e:
                raise UnableToConvertValueError(f"Unable to convert {value!r} to {to_type}") from e
            TypeRegistry.register(
                TypeConverterEntry(func=to_type, from_type=from_type, to_type=to_type, error_types=(Exception,))
            )
            LOGGER.debug(
                "Converted %r (%s) to %r (%s) using constructor (auto-registered)",
                value,
                type(value).__name__,
                new_value,
                to_type.__name__,
            )
            return new_value

        fn = self.conversion_map[key]

        try:
            new_value = fn.func(value)
            LOGGER.debug(
                "Converted %r (%s) to %r (%s) using registered converter %s",
                value,
                type(value).__name__,
                new_value,
                to_type.__name__,
                fn.func.__name__,
            )
            return new_value
        except fn.error_types as e:
            default_err_msg = f"Error converting {value!r} ({type(value).__name__}) to {to_type.__name__}"
            err_msg = fn.error_message or default_err_msg
            if get_format_fields(err_msg):
                err_msg = err_msg.format(value=value, from_type=from_type, to_type=to_type)

            LOGGER.debug("Error converting %r (%s) to %s: %s", value, type(value).__name__, to_type.__name__, err_msg)

            raise UnableToConvertValueError(err_msg) from e
        except Exception as e:
            raise RuntimeError(f"Error converting {value!r} ({type(value).__name__}) to {to_type.__name__}") from e

    def list_conversions(self) -> list[tuple[type, type, TypeConverterEntry]]:
        """List all registered type conversions.

        Returns a sorted list of all registered type conversions with their metadata.
        Useful for debugging and inspection of available converters.

        Returns:
            List of (from_type, to_type, entry) tuples sorted by from_type name then to_type name.

        Example:
            ```python
            from hassette import TYPE_REGISTRY

            conversions = TYPE_REGISTRY.list_conversions()
            for from_type, to_type, entry in conversions:
                print(f"{from_type.__name__} → {to_type.__name__}: {entry.description}")
            ```
        """
        items = []
        for (from_type, to_type), entry in self.conversion_map.items():
            items.append((from_type, to_type, entry))
        items.sort(key=lambda x: (x[0].__name__, x[1].__name__))
        return items

    @classmethod
    def snapshot(cls) -> dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]:
        """Return a shallow copy of the current conversion map."""
        return dict(cls.conversion_map)

    @classmethod
    def restore(cls, snapshot: dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]) -> None:
        """Replace the conversion map with a previously captured snapshot."""
        cls.conversion_map = snapshot

register(type_converter: TypeConverterEntry[Any, Any]) -> None classmethod

Register a type converter in the registry.

Source code in src/hassette/conversion/type_registry.py
147
148
149
150
151
152
153
154
155
@classmethod
def register(cls, type_converter: TypeConverterEntry[Any, Any]) -> None:
    """Register a type converter in the registry."""
    from_type = type_converter.from_type
    to_type = type_converter.to_type
    key = (from_type, to_type)
    if key in cls.conversion_map:
        LOGGER.warning("Overwriting existing conversion from %s to %s", from_type.__name__, to_type.__name__)
    cls.conversion_map[key] = type_converter

convert(value: Any, to_type: type[Any] | tuple[type[Any], ...]) -> Any

Convert a StateValue to a target Python type.

Parameters:

Name Type Description Default
value Any

The StateValue instance to convert.

required
to_type type[Any] | tuple[type[Any], ...]

The target Python type.

required

Returns:

Type Description
Any

The converted value.

Source code in src/hassette/conversion/type_registry.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def convert(self, value: Any, to_type: type[Any] | tuple[type[Any], ...]) -> Any:
    """Convert a StateValue to a target Python type.

    Args:
        value: The StateValue instance to convert.
        to_type: The target Python type.

    Returns:
        The converted value.
    """

    # handle tuple
    if isinstance(to_type, tuple):
        if isinstance(value, to_type):
            return value

        for tt in to_type:
            with suppress(UnableToConvertValueError):
                return self.convert(value, tt)
        raise UnableToConvertValueError(f"Unable to convert {value!r} to any of the types {to_type}")

    # handle single type

    from_type = type(value)
    key = (from_type, to_type)

    # handle Any type
    if to_type is type(Any):
        return value

    # handle exact type match
    if to_type is from_type:
        return value

    # handle None value
    if value is None:
        return value

    if to_type is type(None) and value is not None:
        LOGGER.debug("Not attempting to convert %r to NoneType", value)
        raise UnableToConvertValueError(f"Cannot convert {value!r} to NoneType")

    # if we don't have this in our map, attempt to just convert using it as a constructor
    if key not in self.conversion_map and to_type is not type(None):
        try:
            new_value = to_type(value)
        except Exception as e:
            raise UnableToConvertValueError(f"Unable to convert {value!r} to {to_type}") from e
        TypeRegistry.register(
            TypeConverterEntry(func=to_type, from_type=from_type, to_type=to_type, error_types=(Exception,))
        )
        LOGGER.debug(
            "Converted %r (%s) to %r (%s) using constructor (auto-registered)",
            value,
            type(value).__name__,
            new_value,
            to_type.__name__,
        )
        return new_value

    fn = self.conversion_map[key]

    try:
        new_value = fn.func(value)
        LOGGER.debug(
            "Converted %r (%s) to %r (%s) using registered converter %s",
            value,
            type(value).__name__,
            new_value,
            to_type.__name__,
            fn.func.__name__,
        )
        return new_value
    except fn.error_types as e:
        default_err_msg = f"Error converting {value!r} ({type(value).__name__}) to {to_type.__name__}"
        err_msg = fn.error_message or default_err_msg
        if get_format_fields(err_msg):
            err_msg = err_msg.format(value=value, from_type=from_type, to_type=to_type)

        LOGGER.debug("Error converting %r (%s) to %s: %s", value, type(value).__name__, to_type.__name__, err_msg)

        raise UnableToConvertValueError(err_msg) from e
    except Exception as e:
        raise RuntimeError(f"Error converting {value!r} ({type(value).__name__}) to {to_type.__name__}") from e

list_conversions() -> list[tuple[type, type, TypeConverterEntry]]

List all registered type conversions.

Returns a sorted list of all registered type conversions with their metadata. Useful for debugging and inspection of available converters.

Returns:

Type Description
list[tuple[type, type, TypeConverterEntry]]

List of (from_type, to_type, entry) tuples sorted by from_type name then to_type name.

Example
from hassette import TYPE_REGISTRY

conversions = TYPE_REGISTRY.list_conversions()
for from_type, to_type, entry in conversions:
    print(f"{from_type.__name__}{to_type.__name__}: {entry.description}")
Source code in src/hassette/conversion/type_registry.py
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def list_conversions(self) -> list[tuple[type, type, TypeConverterEntry]]:
    """List all registered type conversions.

    Returns a sorted list of all registered type conversions with their metadata.
    Useful for debugging and inspection of available converters.

    Returns:
        List of (from_type, to_type, entry) tuples sorted by from_type name then to_type name.

    Example:
        ```python
        from hassette import TYPE_REGISTRY

        conversions = TYPE_REGISTRY.list_conversions()
        for from_type, to_type, entry in conversions:
            print(f"{from_type.__name__} → {to_type.__name__}: {entry.description}")
        ```
    """
    items = []
    for (from_type, to_type), entry in self.conversion_map.items():
        items.append((from_type, to_type, entry))
    items.sort(key=lambda x: (x[0].__name__, x[1].__name__))
    return items

snapshot() -> dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]] classmethod

Return a shallow copy of the current conversion map.

Source code in src/hassette/conversion/type_registry.py
266
267
268
269
@classmethod
def snapshot(cls) -> dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]:
    """Return a shallow copy of the current conversion map."""
    return dict(cls.conversion_map)

restore(snapshot: dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]) -> None classmethod

Replace the conversion map with a previously captured snapshot.

Source code in src/hassette/conversion/type_registry.py
271
272
273
274
@classmethod
def restore(cls, snapshot: dict[tuple[type[Any], type[Any]], TypeConverterEntry[Any, Any]]) -> None:
    """Replace the conversion map with a previously captured snapshot."""
    cls.conversion_map = snapshot

RegistryValidationIssue dataclass

A single issue found during registry validation.

Attributes:

Name Type Description
registry str

Which registry produced the issue — "STATE_REGISTRY" or "TYPE_REGISTRY".

severity str

"error" for issues that indicate broken/missing registrations; "warning" for non-fatal anomalies such as duplicate domain registrations.

message str

Human-readable description of the issue.

Source code in src/hassette/conversion/validation.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True)
class RegistryValidationIssue:
    """A single issue found during registry validation.

    Attributes:
        registry: Which registry produced the issue — ``"STATE_REGISTRY"`` or
            ``"TYPE_REGISTRY"``.
        severity: ``"error"`` for issues that indicate broken/missing
            registrations; ``"warning"`` for non-fatal anomalies such as
            duplicate domain registrations.
        message: Human-readable description of the issue.
    """

    registry: str
    severity: str
    message: str

convert_state_dict_to_model(value: typing.Any, model: type[BaseState]) -> BaseState

Convert a raw Home Assistant state dict to a typed state model.

This converter is used by state object extractors (StateNew, StateOld, etc.) to transform the raw state dictionary from Home Assistant into a strongly-typed Pydantic model.

Parameters:

Name Type Description Default
value Any

The raw state dict from Home Assistant

required
model type[BaseState]

The target state model class (e.g., LightState, SensorState)

required

Returns:

Type Description
BaseState

The typed state model instance

Raises:

Type Description
TypeError

If value is not a dict or model instance

ValidationError

If the state dict doesn't match the model schema

Source code in src/hassette/conversion/state_registry.py
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def convert_state_dict_to_model(value: typing.Any, model: type["BaseState"]) -> "BaseState":
    """Convert a raw Home Assistant state dict to a typed state model.

    This converter is used by state object extractors (StateNew, StateOld, etc.) to transform
    the raw state dictionary from Home Assistant into a strongly-typed Pydantic model.

    Args:
        value: The raw state dict from Home Assistant
        model: The target state model class (e.g., LightState, SensorState)

    Returns:
        The typed state model instance

    Raises:
        TypeError: If value is not a dict or model instance
        ValidationError: If the state dict doesn't match the model schema
    """
    if isinstance(value, model):
        return value

    if not isinstance(value, dict):
        raise TypeError(f"Cannot convert {type(value).__name__} to {model.__name__}, expected dict")

    return model.model_validate(value)

register_state_converter(state_class: type[BaseState], domain: Hashable, device_class: Hashable | None = None) -> None

Register a state converter class for a specific domain and optional device class.

Source code in src/hassette/conversion/state_registry.py
31
32
33
34
35
def register_state_converter(
    state_class: type["BaseState"], domain: Hashable, device_class: Hashable | None = None
) -> None:
    """Register a state converter class for a specific domain and optional device class."""
    StateRegistry.register(state_class, domain=domain, device_class=device_class)

register_simple_type_converter(from_type: type[T], to_type: type[R], fn: Callable[[T], R] | None = None, error_message: str | None = None, error_types: tuple[type[BaseException], ...] = (ValueError,))

Register a simple type conversion function from a non-user defined function, such as a constructor.

Parameters:

Name Type Description Default
from_type type[T]

The source type to convert from.

required
to_type type[R]

The target type to convert to.

required
fn Callable[[T], R] | None

The function to use for conversion. If None, the target type constructor is used.

None
error_message str | None

Optional custom error message if conversion fails.

None
error_types tuple[type[BaseException], ...]

Tuple of exception types to catch and wrap in UnableToConvertValueError.

(ValueError,)
Example

register_simple_type_converter(int, float, error_message="Failed to convert int to float") register_simple_type_converter(ZonedDateTime, str, fn=ZonedDateTime.format_iso)

Source code in src/hassette/conversion/type_registry.py
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def register_simple_type_converter(
    from_type: type[T],
    to_type: type[R],
    fn: Callable[[T], R] | None = None,
    error_message: str | None = None,
    error_types: tuple[type[BaseException], ...] = (ValueError,),
):
    """Register a simple type conversion function from a non-user defined function, such as a constructor.

    Args:
        from_type: The source type to convert from.
        to_type: The target type to convert to.
        fn: The function to use for conversion. If None, the target type constructor is used.
        error_message: Optional custom error message if conversion fails.
        error_types: Tuple of exception types to catch and wrap in UnableToConvertValueError.

    Example:
        register_simple_type_converter(int, float, error_message="Failed to convert int to float")
        register_simple_type_converter(ZonedDateTime, str, fn=ZonedDateTime.format_iso)
    """
    if error_message is not None:
        fields = get_format_fields(error_message)
        invalid_fields = set(fields) - ALLOWED_FORMAT_FIELDS
        if invalid_fields:
            raise ValueError(f"Invalid format fields in error_message: {invalid_fields}")

    fn = fn or (lambda x: to_type(x))  # pyright: ignore[reportCallIssue]

    TypeRegistry.register(
        TypeConverterEntry(
            func=fn,
            from_type=from_type,
            to_type=to_type,
            error_message=error_message,
            error_types=error_types,
        )
    )

register_type_converter_fn(fn: Callable[[T], R] | None = None, *, error_message: str | None = None, error_types: tuple[type[BaseException], ...] = (ValueError,)) -> Callable[[T], R] | Callable[[Callable[[T], R]], Callable[[T], R]]

register_type_converter_fn(
    fn: Callable[[T], R],
) -> Callable[[T], R]
register_type_converter_fn(
    fn: None = None,
    *,
    error_message: str | None = None,
    error_types: tuple[type[BaseException], ...] = (
        ValueError,
    ),
) -> Callable[[Callable[[T], R]], Callable[[T], R]]

Register a type conversion function with the TypeRegistry.

Can be used as:

@register_type_converter
def convert_x(value: T) -> R: ...

or:

@register_type_converter(error_message="failed to convert X")
def convert_x(value: T) -> R: ...
Source code in src/hassette/conversion/type_registry.py
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def register_type_converter_fn(
    fn: Callable[[T], R] | None = None,
    *,
    error_message: str | None = None,
    error_types: tuple[type[BaseException], ...] = (ValueError,),
) -> Callable[[T], R] | Callable[[Callable[[T], R]], Callable[[T], R]]:
    """Register a type conversion function with the TypeRegistry.

    Can be used as:

        @register_type_converter
        def convert_x(value: T) -> R: ...

    or:

        @register_type_converter(error_message="failed to convert X")
        def convert_x(value: T) -> R: ...
    """
    if error_message is not None:
        fields = get_format_fields(error_message)
        invalid_fields = set(fields) - ALLOWED_FORMAT_FIELDS
        if invalid_fields:
            raise ValueError(f"Invalid format fields in error_message: {invalid_fields}")

    def decorator(func: Callable[[T], R]) -> Callable[[T], R]:
        from_type = func.__annotations__["value"]
        to_type = func.__annotations__["return"]
        TypeRegistry.register(
            TypeConverterEntry(
                func=func, from_type=from_type, to_type=to_type, error_message=error_message, error_types=error_types
            )
        )
        return func

    # Used as bare @register_type_converter
    if fn is not None:
        return decorator(fn)

    # Used as @register_type_converter(...)
    return decorator

validate_registries(state_registry: StateRegistry, type_registry: TypeRegistry, *, strict: bool = False) -> list[RegistryValidationIssue]

Validate STATE_REGISTRY and TYPE_REGISTRY contents at startup.

Collects ALL issues before raising or logging — never fail-fast.

Parameters:

Name Type Description Default
state_registry StateRegistry

The StateRegistry singleton to validate.

required
type_registry TypeRegistry

The TypeRegistry singleton to validate.

required
strict bool

When True, raise RegistryValidationError if any error-severity issues are found. When False (default), log each issue as a WARNING.

False

Returns:

Type Description
list[RegistryValidationIssue]

A list of RegistryValidationIssue instances (empty when everything

list[RegistryValidationIssue]

is healthy).

Raises:

Type Description
RegistryValidationError

In strict mode when at least one error-level issue is present.

Source code in src/hassette/conversion/validation.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def validate_registries(
    state_registry: StateRegistry,
    type_registry: TypeRegistry,
    *,
    strict: bool = False,
) -> list[RegistryValidationIssue]:
    """Validate STATE_REGISTRY and TYPE_REGISTRY contents at startup.

    Collects ALL issues before raising or logging — never fail-fast.

    Args:
        state_registry: The ``StateRegistry`` singleton to validate.
        type_registry: The ``TypeRegistry`` singleton to validate.
        strict: When ``True``, raise ``RegistryValidationError`` if any
            error-severity issues are found.  When ``False`` (default), log
            each issue as a WARNING.

    Returns:
        A list of ``RegistryValidationIssue`` instances (empty when everything
        is healthy).

    Raises:
        RegistryValidationError: In strict mode when at least one error-level
            issue is present.
    """
    issues: list[RegistryValidationIssue] = []

    issues.extend(_validate_state_registry(state_registry))
    issues.extend(_validate_type_registry(type_registry))

    _apply_mode(issues, strict=strict)
    return issues