Predicates

Predicates combine accessors and conditions to form reusable boolean functions.

A predicate takes a source callable that extracts a value from an event, and a condition that tests the extracted value. The condition may be a literal value, a callable, or a more complex condition object. Conditions can be composed of other conditions to form complex logic.

Examples:

Basic value comparison

ValueIs(source=get_entity_id, condition="light.kitchen")

With a callable condition

def is_kitchen_light(entity_id: str) -> bool:
    return entity_id == "light.kitchen"

ValueIs(source=get_entity_id, condition=is_kitchen_light)

With a condition object

ValueIs(
    source=get_entity_id,
    condition=C.IsIn(collection=["light.kitchen", "light.living"]),
)

Combining multiple predicates

P.AllOf(predicates=[
    P.DomainMatches("light"),
    P.EntityMatches("light.kitchen"),
    P.StateTo("on"),
])

Guard dataclass

Bases: Generic[EventT]

Wraps a predicate function to be used in combinators.

Allows for passing any callable as a predicate. Generic over EventT to allow type checkers to understand the expected event type.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class Guard(typing.Generic[EventT]):
    """Wraps a predicate function to be used in combinators.

    Allows for passing any callable as a predicate. Generic over EventT to allow type checkers to understand the
    expected event type.
    """

    fn: "Predicate[EventT]"

    def __call__(self, value: "EventT") -> bool:
        return self.fn(value)

    def summarize(self) -> str:
        return "custom condition"

AllOf dataclass

Predicate that evaluates to True if all of the contained predicates evaluate to True.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AllOf:
    """Predicate that evaluates to True if all of the contained predicates evaluate to True."""

    predicates: tuple["Predicate", ...]
    """The predicates to evaluate."""

    def __call__(self, value: "Event") -> bool:
        return all(p(value) for p in self.predicates)

    def summarize(self) -> str:
        joined = " and ".join(_summarize_predicate(p) for p in self.predicates)
        if len(self.predicates) >= 2:
            return f"({joined})"
        return joined

    @classmethod
    def ensure_iterable(cls, where: "Predicate | Sequence[Predicate] | list[Predicate]") -> "AllOf":
        return cls(ensure_tuple(where))

predicates: tuple[Predicate, ...] instance-attribute

The predicates to evaluate.

AnyOf dataclass

Predicate that evaluates to True if any of the contained predicates evaluate to True.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AnyOf:
    """Predicate that evaluates to True if any of the contained predicates evaluate to True."""

    predicates: tuple["Predicate", ...]
    """The predicates to evaluate."""

    def __call__(self, event: "Event") -> bool:
        return any(p(event) for p in self.predicates)

    def summarize(self) -> str:
        joined = " or ".join(_summarize_predicate(p) for p in self.predicates)
        if len(self.predicates) >= 2:
            return f"({joined})"
        return joined

    @classmethod
    def ensure_iterable(cls, where: "Predicate | Sequence[Predicate]") -> "AnyOf":
        return cls(ensure_tuple(where))

predicates: tuple[Predicate, ...] instance-attribute

The predicates to evaluate.

Not dataclass

Negates the result of the predicate.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class Not:
    """Negates the result of the predicate."""

    predicate: "Predicate"

    def __call__(self, value: "Event", /) -> bool:
        return not self.predicate(value)

    def summarize(self) -> str:
        return "not " + _summarize_predicate(self.predicate)

ValueIs dataclass

Bases: Generic[EventT, V]

Checks whether a value extracted from an event satisfies a condition.

Parameters:

Name	Type	Description	Default
source	Callable[[EventT], V]	Callable that extracts the value to compare from the event.	required
condition	ChangeType	A literal or callable tested against the extracted value. If ANY_VALUE, always True.	ANY_VALUE

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class ValueIs(Generic[EventT, V]):
    """Checks whether a value extracted from an event satisfies a condition.

    Args:
        source: Callable that extracts the value to compare from the event.
        condition: A literal or callable tested against the extracted value. If ANY_VALUE, always True.
    """

    source: Callable[[EventT], V]
    condition: "ChangeType" = ANY_VALUE

    def __call__(self, value: EventT, /) -> bool:
        if self.condition is ANY_VALUE:
            return True
        extracted = self.source(value)
        return compare_value(extracted, self.condition)

    def summarize(self) -> str:
        source_name = callable_name(self.source)
        if callable(self.condition):
            return f"custom condition from {source_name}"
        return f"value is {self.condition} from {source_name}"

DidChange dataclass

Bases: Generic[EventT]

Predicate that is True when two extracted values differ.

Typical use is an accessor that returns (old_value, new_value).

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class DidChange(Generic[EventT]):
    """Predicate that is True when two extracted values differ.

    Typical use is an accessor that returns (old_value, new_value).
    """

    source: Callable[[EventT], tuple[Any, Any]]

    def __call__(self, value: EventT, /) -> bool:
        old_v, new_v = self.source(value)
        return old_v != new_v

    def summarize(self) -> str:
        return "changed"

IsPresent dataclass

Checks if a value extracted from an event is present (not MISSING_VALUE).

This will generally be used when comparing state changes, where either the old or new state may be missing.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class IsPresent:
    """Checks if a value extracted from an event is present (not MISSING_VALUE).

    This will generally be used when comparing state changes, where either the old or new state may be missing.

    """

    source: Callable[[Any], Any]

    def __call__(self, value: Any, /) -> bool:
        return self.source(value) is not MISSING_VALUE

    def summarize(self) -> str:
        return "is present"

IsMissing dataclass

Checks if a value extracted from an event is missing (MISSING_VALUE).

This will generally be used when comparing state changes, where either the old or new state may be missing.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class IsMissing:
    """Checks if a value extracted from an event is missing (MISSING_VALUE).

    This will generally be used when comparing state changes, where either the old or new state may be missing.

    """

    source: Callable[[Any], Any]

    def __call__(self, value: Any, /) -> bool:
        return self.source(value) is MISSING_VALUE

    def summarize(self) -> str:
        return "is missing"

StateFrom dataclass

Checks if a value extracted from a RawStateChangeEvent satisfies a condition on the 'old' value.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class StateFrom:
    """Checks if a value extracted from a RawStateChangeEvent satisfies a condition on the 'old' value."""

    condition: "ChangeType"

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return ValueIs(source=get_state_value_old, condition=self.condition)(value)

    def summarize(self) -> str:
        return f"from {_summarize_condition(self.condition)}"

StateTo dataclass

Checks if a value extracted from a RawStateChangeEvent satisfies a condition on the 'new' value.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class StateTo:
    """Checks if a value extracted from a RawStateChangeEvent satisfies a condition on the 'new' value."""

    condition: "ChangeType"

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return ValueIs(source=get_state_value_new, condition=self.condition)(value)

    def summarize(self) -> str:
        return f"{ARROW} {_summarize_condition(self.condition)}"

StateComparison dataclass

Checks if a comparison between from_state and to_state satisfies a condition.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class StateComparison:
    """Checks if a comparison between from_state and to_state satisfies a condition."""

    condition: ComparisonCondition

    def __post_init__(self) -> None:
        if inspect.isclass(self.condition):
            LOGGER.warning("StateComparison was passed a class instead of an instance.", stacklevel=2)
            object.__setattr__(self, "condition", self.condition())

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return self.condition(get_state_value_old(value), get_state_value_new(value))

    def summarize(self) -> str:
        return f"state {_summarize_condition(self.condition)}"

AttrFrom dataclass

Checks if a specific attribute changed in a RawStateChangeEvent.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AttrFrom:
    """Checks if a specific attribute changed in a RawStateChangeEvent."""

    attr_name: str
    condition: "ChangeType"

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return ValueIs(source=get_attr_old(self.attr_name), condition=self.condition)(value)

    def summarize(self) -> str:
        return f"attr {self.attr_name} from {_summarize_condition(self.condition)}"

AttrTo dataclass

Checks if a specific attribute changed in a RawStateChangeEvent.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AttrTo:
    """Checks if a specific attribute changed in a RawStateChangeEvent."""

    attr_name: str
    condition: "ChangeType"

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return ValueIs(source=get_attr_new(self.attr_name), condition=self.condition)(value)

    def summarize(self) -> str:
        return f"attr {self.attr_name} {ARROW} {_summarize_condition(self.condition)}"

AttrComparison dataclass

Checks if a comparison between from_attr and to_attr satisfies a condition.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AttrComparison:
    """Checks if a comparison between from_attr and to_attr satisfies a condition."""

    attr_name: str
    condition: ComparisonCondition

    def __post_init__(self) -> None:
        if inspect.isclass(self.condition):
            LOGGER.warning("AttrComparison was passed a class instead of an instance.", stacklevel=2)
            object.__setattr__(self, "condition", self.condition())

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        old_attr = get_attr_old(self.attr_name)(value)
        new_attr = get_attr_new(self.attr_name)(value)
        return self.condition(old_attr, new_attr)

    def summarize(self) -> str:
        return f"attr {self.attr_name} {_summarize_condition(self.condition)}"

StateDidChange dataclass

Checks if the state changed in a RawStateChangeEvent.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class StateDidChange:
    """Checks if the state changed in a RawStateChangeEvent."""

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        return DidChange(get_state_value_old_new)(value)

    def summarize(self) -> str:
        return "state changed"

AttrDidChange dataclass

Checks if a specific attribute changed in a RawStateChangeEvent.

When old_state is None, the attribute is treated as having changed if it is present on new_state. This applies to synthetic immediate-fire/bootstrap events, the cancel handler's old_state stripping in DurationTimer, and real state-change events where there is no prior state yet (first observation / first-time state set).

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class AttrDidChange:
    """Checks if a specific attribute changed in a RawStateChangeEvent.

    When ``old_state`` is None, the attribute is treated as having changed
    if it is present on ``new_state``.  This applies to synthetic
    immediate-fire/bootstrap events, the cancel handler's old_state stripping
    in DurationTimer, and real state-change events where there is no prior
    state yet (first observation / first-time state set).
    """

    attr_name: str

    def __call__(self, value: "RawStateChangeEvent", /) -> bool:
        # old_state=None arises in two paths: (1) synthetic immediate-fire events
        # and (2) the cancel handler's old_state stripping in DurationTimer.
        # Both need "attribute present = changed" semantics — but only if the
        # attribute actually exists on new_state.
        if value.payload.data.old_state is None:
            return get_attr_new(self.attr_name)(value) is not MISSING_VALUE
        return DidChange(get_attr_old_new(self.attr_name))(value)

    def summarize(self) -> str:
        return f"attr {self.attr_name} changed"

DomainMatches dataclass

Checks if the event domain matches a specific value.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class DomainMatches:
    """Checks if the event domain matches a specific value."""

    domain: str

    def __call__(self, value: "HassEvent", /) -> bool:
        cond = Glob(self.domain) if is_glob(self.domain) else self.domain
        return ValueIs(source=get_domain, condition=cond)(value)

    def summarize(self) -> str:
        return f"domain {self.domain}"

    def __repr__(self) -> str:
        return f"DomainMatches(domain={self.domain!r})"

EntityMatches dataclass

Checks if the event entity_id matches a specific value.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class EntityMatches:
    """Checks if the event entity_id matches a specific value."""

    entity_id: str

    def __call__(self, value: "HassEvent", /) -> bool:
        cond = Glob(self.entity_id) if is_glob(self.entity_id) else self.entity_id
        return ValueIs(source=get_entity_id, condition=cond)(value)

    def summarize(self) -> str:
        return f"entity {self.entity_id}"

    def __repr__(self) -> str:
        return f"EntityMatches(entity_id={self.entity_id!r})"

ServiceMatches dataclass

Checks if the event service matches a specific value.

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class ServiceMatches:
    """Checks if the event service matches a specific value."""

    service: str

    def __call__(self, value: "HassEvent", /) -> bool:
        cond = Glob(self.service) if is_glob(self.service) else self.service
        return ValueIs(source=get_path("payload.data.service"), condition=cond)(value)

    def summarize(self) -> str:
        return f"service {self.service}"

    def __repr__(self) -> str:
        return f"ServiceMatches(service={self.service!r})"

ServiceDataWhere dataclass

Predicate that applies a mapping of service_data conditions to a CallServiceEvent.

Examples

Exact matches only

ServiceDataWhere({"entity_id": "light.kitchen", "transition": 1})

With a callable condition

ServiceDataWhere({"brightness": lambda v: isinstance(v, int) and v >= 150})

With globs (auto-wrapped)

ServiceDataWhere({"entity_id": "light.*"})

Using conditions

ServiceDataWhere({"entity_id": Glob("switch.*")})

ServiceDataWhere({"brightness": IsIn([100, 200, 255])})

Source code in src/hassette/event_handling/predicates.py

@dataclass(frozen=True)
class ServiceDataWhere:
    """Predicate that applies a mapping of service_data conditions to a CallServiceEvent.

    Examples
    --------
    Exact matches only

        ServiceDataWhere({"entity_id": "light.kitchen", "transition": 1})

    With a callable condition

        ServiceDataWhere({"brightness": lambda v: isinstance(v, int) and v >= 150})

    With globs (auto-wrapped)

        ServiceDataWhere({"entity_id": "light.*"})

    Using conditions

        ServiceDataWhere({"entity_id": Glob("switch.*")})

        ServiceDataWhere({"brightness": IsIn([100, 200, 255])})
    """

    spec: Mapping[str, "ChangeType"]
    auto_glob: bool = True
    _predicates: tuple["Predicate[CallServiceEvent]", ...] = field(init=False, repr=False)

    def __post_init__(self) -> None:
        preds: list[Predicate[CallServiceEvent]] = []

        for k, cond in self.spec.items():
            source = get_service_data_key(k)
            c: ChangeType
            # presence check
            if cond is ANY_VALUE:
                c = Present()
            # auto-glob wrapping
            elif self.auto_glob and isinstance(cond, str) and is_glob(cond):
                c = Glob(cond)
            # literal or callable condition
            else:
                c = cond
            preds.append(ValueIs(source=source, condition=c))

        object.__setattr__(self, "_predicates", tuple(preds))

    def __call__(self, value: "CallServiceEvent", /) -> bool:
        return all(p(value) for p in self._predicates)

    def summarize(self) -> str:
        def _fmt(v: Any) -> str:
            if callable(v):
                return _summarize_condition(v) if hasattr(v, "summarize") else callable_name(v)
            return str(v)

        parts = [f"{k} = {_fmt(v)}" for k, v in self.spec.items()]
        return ", ".join(parts)

    @classmethod
    def from_kwargs(cls, *, auto_glob: bool = True, **spec: "ChangeType") -> "ServiceDataWhere":
        """Ergonomic constructor for literal kwargs.

        Example
        -------
        >>> ServiceDataWhere.from_kwargs(entity_id="light.*", brightness=200)
        """
        return cls(spec=spec, auto_glob=auto_glob)

from_kwargs(*, auto_glob: bool = True, **spec: ChangeType) -> ServiceDataWhere classmethod

Ergonomic constructor for literal kwargs.

Example

ServiceDataWhere.from_kwargs(entity_id="light.*", brightness=200)

Source code in src/hassette/event_handling/predicates.py

@classmethod
def from_kwargs(cls, *, auto_glob: bool = True, **spec: "ChangeType") -> "ServiceDataWhere":
    """Ergonomic constructor for literal kwargs.

    Example
    -------
    >>> ServiceDataWhere.from_kwargs(entity_id="light.*", brightness=200)
    """
    return cls(spec=spec, auto_glob=auto_glob)

summarize_top_level(predicate: Predicate) -> str

Return a human-readable summary suitable for display as a top-level label.

Calls summarize() on the predicate, then strips balanced outer parentheses so top-level combinators don't produce redundant wrapping.

Source code in src/hassette/event_handling/predicates.py

def summarize_top_level(predicate: "Predicate") -> str:
    """Return a human-readable summary suitable for display as a top-level label.

    Calls ``summarize()`` on the predicate, then strips balanced outer
    parentheses so top-level combinators don't produce redundant wrapping.
    """
    return _strip_outer_parens(_summarize_predicate(predicate))

compare_value(actual: Any, condition: ChangeType) -> bool

Compare an actual value against a condition.

Parameters:

Name	Type	Description	Default
actual	Any	The actual value to compare.	required
condition	ChangeType	The condition to compare against. Can be a literal value or a callable.	required

Returns:

Type	Description
bool	True if the actual value matches the condition, False otherwise.

Behavior

• If condition is NOT_PROVIDED, treat as 'no constraint' (True).
• If condition is a non-callable, compare for equality only.
• If condition is a callable, call and ensure bool.
• Async/coroutine predicates are explicitly disallowed (raise).

Note

This function does not handle collections any differently than other literals — it compares them for equality only. Use specific conditions like IsIn/NotIn/Intersects for collection membership tests.

Source code in src/hassette/event_handling/predicates.py

def compare_value(actual: Any, condition: "ChangeType") -> bool:
    """Compare an actual value against a condition.

    Args:
        actual: The actual value to compare.
        condition: The condition to compare against. Can be a literal value or a callable.

    Returns:
        True if the actual value matches the condition, False otherwise.

    Behavior:
        - If condition is NOT_PROVIDED, treat as 'no constraint' (True).
        - If condition is a non-callable, compare for equality only.
        - If condition is a callable, call and ensure bool.
        - Async/coroutine predicates are explicitly disallowed (raise).

    Note:
        This function does not handle collections any differently than other literals — it compares
        them for equality only. Use specific conditions like IsIn/NotIn/Intersects for collection membership tests.
    """
    if condition is NOT_PROVIDED:
        return True

    if not callable(condition):
        return actual == condition

    # Disallow async predicates to keep filters pure/fast.
    if iscoroutinefunction(condition):
        raise TypeError("Async predicates are not supported; make the condition synchronous.")

    if typing.TYPE_CHECKING:
        condition = typing.cast("Callable[[Any], bool]", condition)

    result = condition(actual)

    if isawaitable(result):
        raise TypeError("Predicate returned an awaitable; make it return bool.")

    # Fallback: callable but not declared as PredicateCallable; still require bool.
    if not isinstance(result, bool):
        raise TypeError(f"Predicate must return bool, got {type(result)}")
    return result

ensure_tuple(where: Predicate | Sequence[Predicate]) -> tuple[Predicate, ...]

Ensure the 'where' is a flat tuple of predicates, flattening only predicate collections.

Recurses into list/tuple/set/frozenset; leaves Mapping, strings/bytes, and callables intact.

Source code in src/hassette/event_handling/predicates.py

def ensure_tuple(where: "Predicate | Sequence[Predicate]") -> tuple["Predicate", ...]:
    """Ensure the 'where' is a flat tuple of predicates, flattening *only* predicate collections.

    Recurses into list/tuple/set/frozenset; leaves Mapping, strings/bytes, and callables intact.
    """
    if is_predicate_collection(where):
        out: list[Predicate] = []
        # mypy/pyright: guarded by _is_predicate_collection, so safe to iterate
        for item in typing.cast("Sequence[Predicate | Sequence[Predicate]]", where):
            out.extend(ensure_tuple(item))
        return tuple(out)

    return (typing.cast("Predicate", where),)

is_predicate_collection(obj: Any) -> TypeGuard[Sequence[Predicate]]

Return True for predicate collections we want to recurse into.

We treat only list/tuple/set/frozenset-like things as collections of predicates. We explicitly DO NOT recurse into: - mappings (those feed ServiceDataWhere elsewhere), - strings/bytes, - callables (predicates are callables; don't explode them), - None.

Source code in src/hassette/event_handling/predicates.py

def is_predicate_collection(obj: Any) -> TypeGuard[Sequence["Predicate"]]:
    """Return True for *predicate collections* we want to recurse into.

    We treat only list/tuple/set/frozenset-like things as collections of predicates.
    We explicitly DO NOT recurse into:
      - mappings (those feed ServiceDataWhere elsewhere),
      - strings/bytes,
      - callables (predicates are callables; don't explode them),
      - None.
    """
    if obj is None:
        return False
    if callable(obj):
        return False
    if isinstance(obj, (str, bytes, Mapping)):
        return False
    # boltons.is_collection filters out scalars for us; we just fence off types we don't want
    return is_collection(obj)

normalize_where(where: Predicate | Sequence[Predicate] | None) -> Predicate | None

Normalize a 'where' clause into a single Predicate (usually AllOf.ensure_iterable), or None.

• If where is None → None
• If where is a predicate collection (list/tuple/set/...) → AllOf.ensure_iterable(where)
• Otherwise (single predicate or mapping handled elsewhere) → where

Source code in src/hassette/event_handling/predicates.py

def normalize_where(where: "Predicate | Sequence[Predicate] | None") -> "Predicate | None":
    """Normalize a 'where' clause into a single Predicate (usually AllOf.ensure_iterable), or None.

    - If where is None → None
    - If where is a predicate collection (list/tuple/set/...) → AllOf.ensure_iterable(where)
    - Otherwise (single predicate or mapping handled elsewhere) → where
    """
    if where is None:
        return None

    # prevent circular import only when needed
    if is_predicate_collection(where):
        return AllOf.ensure_iterable(where)

    # help the type checker know that `where` is not an Sequence here
    if typing.TYPE_CHECKING:
        assert not isinstance(where, Sequence)

    return where  # single predicate or mapping gets handled by the caller