<commit_before>"""Provide pre-made queries on top of the recorder component."""
from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Iterator, MutableMapping
from datetime import datetime
from itertools import groupby
import logging
import time
from typing import Any, cast

from sqlalchemy import Column, Text, and_, bindparam, func, or_
from sqlalchemy.ext import baked
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.expression import literal

from homeassistant.components import recorder
from homeassistant.core import HomeAssistant, State, split_entity_id
import homeassistant.util.dt as dt_util

from .models import (
    LazyState,
    RecorderRuns,
    StateAttributes,
    States,
    process_timestamp_to_utc_isoformat,
)
from .util import execute, session_scope

# mypy: allow-untyped-defs, no-check-untyped-defs

_LOGGER = logging.getLogger(__name__)

STATE_KEY = "state"
LAST_CHANGED_KEY = "last_changed"

SIGNIFICANT_DOMAINS = {
    "climate",
    "device_tracker",
    "humidifier",
    "thermostat",
    "water_heater",
}
SIGNIFICANT_DOMAINS_ENTITY_ID_LIKE = [f"{domain}.%" for domain in SIGNIFICANT_DOMAINS]
IGNORE_DOMAINS = {"zone", "scene"}
IGNORE_DOMAINS_ENTITY_ID_LIKE = [f"{domain}.%" for domain in IGNORE_DOMAINS]
NEED_ATTRIBUTE_DOMAINS = {
    "climate",
    "humidifier",
    "input_datetime",
    "thermostat",
    "water_heater",
}

BASE_STATES = [
    States.entity_id,
    States.state,
    States.last_changed,
    States.last_updated,
]
QUERY_STATE_NO_ATTR = [
    *BASE_STATES,
    literal(value=None, type_=Text).label("attributes"),
    literal(value=None, type_=Text).label("shared_attrs"),
]
# Remove QUERY_STATES_PRE_SCHEMA_25
# and the migration_in_progress check
# once schema 26 is created
QUERY_STATES_PRE_SCHEMA_25 = [
    *BASE_STATES,
    States.attributes,
    literal(value=None, type_=Text).label("shared_attrs"),
]
QUERY_STATES = [
    *BASE_STATES,
    # Remove States.attributes once all attributes are in StateAttributes.shared_attrs
    States.attributes,
    StateAttributes.shared_attrs,
]

HISTORY_BAKERY = "recorder_history_bakery"


def query_and_join_attributes(
    hass: HomeAssistant, no_attributes: bool
) -> tuple[list[Column], bool]:
    """Return the query keys and if StateAttributes should be joined."""
    # If no_attributes was requested we do the query
    # without the attributes fields and do not join the
    # state_attributes table
    if no_attributes:
        return QUERY_STATE_NO_ATTR, False
    # If we in the process of migrating schema we do
    # not want to join the state_attributes table as we
    # do not know if it will be there yet
    if recorder.get_instance(hass).migration_in_progress:
        return QUERY_STATES_PRE_SCHEMA_25, False
    # Finally if no migration is in progress and no_attributes
    # was not requested, we query both attributes columns and
    # join state_attributes
    return QUERY_STATES, True


def bake_query_and_join_attributes(
    hass: HomeAssistant, no_attributes: bool
) -> tuple[Any, bool]:
    """Return the initial backed query and if StateAttributes should be joined.

    Because these are baked queries the values inside the lambdas need
    to be explicitly written out to avoid caching the wrong values.
    """
    bakery: baked.bakery = hass.data[HISTORY_BAKERY]
    # If no_attributes was requested we do the query
    # without the attributes fields and do not join the
    # state_attributes table
    if no_attributes:
        return bakery(lambda session: session.query(*QUERY_STATE_NO_ATTR)), False
    # If we in the process of migrating schema we do
    # not want to join the state_attributes table as we
    # do not know if it will be there yet
    if recorder.get_instance(hass).migration_in_progress:
        return bakery(lambda session: session.query(*QUERY_STATES_PRE_SCHEMA_25)), False
    # Finally if no migration is in progress and no_attributes
    # was not requested, we query both attributes columns and
    # join state_attributes
    return bakery(lambda session: session.query(*QUERY_STATES)), True


def async_setup(hass: HomeAssistant) -> None:
    """Set up the history hooks."""
    hass.data[HISTORY_BAKERY] = baked.bakery()


def get_significant_states(
    hass: HomeAssistant,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any | None = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]:
    """Wrap get_significant_states_with_session with an sql session."""
    with session_scope(hass=hass) as session:
        return get_significant_states_with_session(
            hass,
            session,
            start_time,
            end_time,
            entity_ids,
            filters,
            include_start_time_state,
            significant_changes_only,
            minimal_response,
            no_attributes,
        )


def _query_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    significant_changes_only: bool = True,
    no_attributes: bool = False,
) -> list[States]:
    """Query the database for significant state changes."""
    if _LOGGER.isEnabledFor(logging.DEBUG):
        timer_start = time.perf_counter()

    baked_query, join_attributes = bake_query_and_join_attributes(hass, no_attributes)

    if entity_ids is not None and len(entity_ids) == 1:
        if (
            significant_changes_only
            and split_entity_id(entity_ids[0])[0] not in SIGNIFICANT_DOMAINS
        ):
            baked_query += lambda q: q.filter(
                States.last_changed == States.last_updated
            )
    elif significant_changes_only:
        baked_query += lambda q: q.filter(
            or_(
                *[
                    States.entity_id.like(entity_domain)
                    for entity_domain in SIGNIFICANT_DOMAINS_ENTITY_ID_LIKE
                ],
                (States.last_changed == States.last_updated),
            )
        )

    if entity_ids is not None:
        baked_query += lambda q: q.filter(
            States.entity_id.in_(bindparam("entity_ids", expanding=True))
        )
    else:
        baked_query += lambda q: q.filter(
            and_(
                *[
                    ~States.entity_id.like(entity_domain)
                    for entity_domain in IGNORE_DOMAINS_ENTITY_ID_LIKE
                ]
            )
        )
        if filters:
            filters.bake(baked_query)

    baked_query += lambda q: q.filter(States.last_updated > bindparam("start_time"))
    if end_time is not None:
        baked_query += lambda q: q.filter(States.last_updated < bindparam("end_time"))

    if join_attributes:
        baked_query += lambda q: q.outerjoin(
            StateAttributes, States.attributes_id == StateAttributes.attributes_id
        )
    baked_query += lambda q: q.order_by(States.entity_id, States.last_updated)

    states = execute(
        baked_query(session).params(
            start_time=start_time, end_time=end_time, entity_ids=entity_ids
        )
    )

    if _LOGGER.isEnabledFor(logging.DEBUG):
        elapsed = time.perf_counter() - timer_start
        _LOGGER.debug("get_significant_states took %fs", elapsed)

    return states


def get_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]:
    """
    Return states changes during UTC period start_time - end_time.

    entity_ids is an optional iterable of entities to include in the results.

    filters is an optional SQLAlchemy filter which will be applied to the database
    queries unless entity_ids is given, in which case its ignored.

    Significant states are all states where there is a state change,
    as well as all states from certain domains (for instance
    thermostat so that we get current temperature in our graphs).
    """
    states = _query_significant_states_with_session(
        hass,
        session,
        start_time,
        end_time,
        entity_ids,
        filters,
        significant_changes_only,
        no_attributes,
    )
    return _sorted_states_to_dict(
        hass,
        session,
        states,
        start_time,
        entity_ids,
        filters,
        include_start_time_state,
        minimal_response,
        no_attributes,
    )


def get_full_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State]]:
    """Variant of get_significant_states_with_session that does not return minimal responses."""
    return cast(
        MutableMapping[str, list[State]],
        get_significant_states_with_session(
            hass=hass,
            session=session,
            start_time=start_time,
            end_time=end_time,
            entity_ids=entity_ids,
            filters=filters,
            include_start_time_state=include_start_time_state,
            significant_changes_only=significant_changes_only,
            minimal_response=False,
            no_attributes=no_attributes,
        ),
    )


def state_changes_during_period(
    hass: HomeAssistant,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_id: str | None = None,
    no_attributes: bool = False,
    descending: bool = False,
    limit: int | None = None,
    include_start_time_state: bool = True,
) -> MutableMapping[str, list[State]]:
    """Return states changes during UTC period start_time - end_time."""
    with session_scope(hass=hass) as session:
        baked_query, join_attributes = bake_query_and_join_attributes(
            hass, no_attributes
        )

        baked_query += lambda q: q.filter(
            (States.last_changed == States.last_updated)
            & (States.last_updated > bindparam("start_time"))
        )

        if end_time is not None:
            baked_query += lambda q: q.filter(
                States.last_updated < bindparam("end_time")
            )

        if entity_id is not None:
            baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
            entity_id = entity_id.lower()

        if join_attributes:
            baked_query += lambda q: q.outerjoin(
                StateAttributes, States.attributes_id == StateAttributes.attributes_id
            )

        last_updated = States.last_updated.desc() if descending else States.last_updated
        baked_query += lambda q: q.order_by(States.entity_id, last_updated)

        if limit:
            baked_query += lambda q: q.limit(limit)

        states = execute(
            baked_query(session).params(
                start_time=start_time, end_time=end_time, entity_id=entity_id
            )
        )

        entity_ids = [entity_id] if entity_id is not None else None

        return cast(
            MutableMapping[str, list[State]],
            _sorted_states_to_dict(
                hass,
                session,
                states,
                start_time,
                entity_ids,
                include_start_time_state=include_start_time_state,
            ),
        )


def get_last_state_changes(
    hass: HomeAssistant, number_of_states: int, entity_id: str
) -> MutableMapping[str, list[State]]:
    """Return the last number_of_states."""
    start_time = dt_util.utcnow()

    with session_scope(hass=hass) as session:
        baked_query, join_attributes = bake_query_and_join_attributes(hass, False)

        baked_query += lambda q: q.filter(States.last_changed == States.last_updated)

        if entity_id is not None:
            baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
            entity_id = entity_id.lower()

        if join_attributes:
            baked_query += lambda q: q.outerjoin(
                StateAttributes, States.attributes_id == StateAttributes.attributes_id
            )
        baked_query += lambda q: q.order_by(
            States.entity_id, States.last_updated.desc()
        )

        baked_query += lambda q: q.limit(bindparam("number_of_states"))

        states = execute(
            baked_query(session).params(
                number_of_states=number_of_states, entity_id=entity_id
            )
        )

        entity_ids = [entity_id] if entity_id is not None else None

        return cast(
            MutableMapping[str, list[State]],
            _sorted_states_to_dict(
                hass,
                session,
                reversed(states),
                start_time,
                entity_ids,
                include_start_time_state=False,
            ),
        )


def get_states(
    hass: HomeAssistant,
    utc_point_in_time: datetime,
    entity_ids: list[str] | None = None,
    run: RecorderRuns | None = None,
    filters: Any = None,
    no_attributes: bool = False,
) -> list[State]:
    """Return the states at a specific point in time."""
    if (
        run is None
        and (run := (recorder.run_information_from_instance(hass, utc_point_in_time)))
        is None
    ):
        # History did not run before utc_point_in_time
        return []

    with session_scope(hass=hass) as session:
        return _get_states_with_session(
            hass, session, utc_point_in_time, entity_ids, run, filters, no_attributes
        )


def _get_states_with_session(
    hass: HomeAssistant,
    session: Session,
    utc_point_in_time: datetime,
    entity_ids: list[str] | None = None,
    run: RecorderRuns | None = None,
    filters: Any | None = None,
    no_attributes: bool = False,
) -> list[State]:
    """Return the states at a specific point in time."""
    if entity_ids and len(entity_ids) == 1:
        return _get_single_entity_states_with_session(
            hass, session, utc_point_in_time, entity_ids[0], no_attributes
        )

    if (
        run is None
        and (run := (recorder.run_information_from_instance(hass, utc_point_in_time)))
        is None
    ):
        # History did not run before utc_point_in_time
        return []<commit_msg>Fix history not including start time state (#70447)<commit_after>"""Provide pre-made queries on top of the recorder component."""
from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Iterator, MutableMapping
from datetime import datetime
from itertools import groupby
import logging
import time
from typing import Any, cast

from sqlalchemy import Column, Text, and_, bindparam, func, or_
from sqlalchemy.ext import baked
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.expression import literal

from homeassistant.components import recorder
from homeassistant.core import HomeAssistant, State, split_entity_id
import homeassistant.util.dt as dt_util

from .models import (
    LazyState,
    RecorderRuns,
    StateAttributes,
    States,
    process_timestamp_to_utc_isoformat,
)
from .util import execute, session_scope

# mypy: allow-untyped-defs, no-check-untyped-defs

_LOGGER = logging.getLogger(__name__)

STATE_KEY = "state"
LAST_CHANGED_KEY = "last_changed"

SIGNIFICANT_DOMAINS = {
    "climate",
    "device_tracker",
    "humidifier",
    "thermostat",
    "water_heater",
}
SIGNIFICANT_DOMAINS_ENTITY_ID_LIKE = [f"{domain}.%" for domain in SIGNIFICANT_DOMAINS]
IGNORE_DOMAINS = {"zone", "scene"}
IGNORE_DOMAINS_ENTITY_ID_LIKE = [f"{domain}.%" for domain in IGNORE_DOMAINS]
NEED_ATTRIBUTE_DOMAINS = {
    "climate",
    "humidifier",
    "input_datetime",
    "thermostat",
    "water_heater",
}

BASE_STATES = [
    States.entity_id,
    States.state,
    States.last_changed,
    States.last_updated,
]
QUERY_STATE_NO_ATTR = [
    *BASE_STATES,
    literal(value=None, type_=Text).label("attributes"),
    literal(value=None, type_=Text).label("shared_attrs"),
]
# Remove QUERY_STATES_PRE_SCHEMA_25
# and the migration_in_progress check
# once schema 26 is created
QUERY_STATES_PRE_SCHEMA_25 = [
    *BASE_STATES,
    States.attributes,
    literal(value=None, type_=Text).label("shared_attrs"),
]
QUERY_STATES = [
    *BASE_STATES,
    # Remove States.attributes once all attributes are in StateAttributes.shared_attrs
    States.attributes,
    StateAttributes.shared_attrs,
]

HISTORY_BAKERY = "recorder_history_bakery"


def query_and_join_attributes(
    hass: HomeAssistant, no_attributes: bool
) -> tuple[list[Column], bool]:
    """Return the query keys and if StateAttributes should be joined."""
    # If no_attributes was requested we do the query
    # without the attributes fields and do not join the
    # state_attributes table
    if no_attributes:
        return QUERY_STATE_NO_ATTR, False
    # If we in the process of migrating schema we do
    # not want to join the state_attributes table as we
    # do not know if it will be there yet
    if recorder.get_instance(hass).migration_in_progress:
        return QUERY_STATES_PRE_SCHEMA_25, False
    # Finally if no migration is in progress and no_attributes
    # was not requested, we query both attributes columns and
    # join state_attributes
    return QUERY_STATES, True


def bake_query_and_join_attributes(
    hass: HomeAssistant, no_attributes: bool
) -> tuple[Any, bool]:
    """Return the initial backed query and if StateAttributes should be joined.

    Because these are baked queries the values inside the lambdas need
    to be explicitly written out to avoid caching the wrong values.
    """
    bakery: baked.bakery = hass.data[HISTORY_BAKERY]
    # If no_attributes was requested we do the query
    # without the attributes fields and do not join the
    # state_attributes table
    if no_attributes:
        return bakery(lambda session: session.query(*QUERY_STATE_NO_ATTR)), False
    # If we in the process of migrating schema we do
    # not want to join the state_attributes table as we
    # do not know if it will be there yet
    if recorder.get_instance(hass).migration_in_progress:
        return bakery(lambda session: session.query(*QUERY_STATES_PRE_SCHEMA_25)), False
    # Finally if no migration is in progress and no_attributes
    # was not requested, we query both attributes columns and
    # join state_attributes
    return bakery(lambda session: session.query(*QUERY_STATES)), True


def async_setup(hass: HomeAssistant) -> None:
    """Set up the history hooks."""
    hass.data[HISTORY_BAKERY] = baked.bakery()


def get_significant_states(
    hass: HomeAssistant,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any | None = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]:
    """Wrap get_significant_states_with_session with an sql session."""
    with session_scope(hass=hass) as session:
        return get_significant_states_with_session(
            hass,
            session,
            start_time,
            end_time,
            entity_ids,
            filters,
            include_start_time_state,
            significant_changes_only,
            minimal_response,
            no_attributes,
        )


def _query_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    significant_changes_only: bool = True,
    no_attributes: bool = False,
) -> list[States]:
    """Query the database for significant state changes."""
    if _LOGGER.isEnabledFor(logging.DEBUG):
        timer_start = time.perf_counter()

    baked_query, join_attributes = bake_query_and_join_attributes(hass, no_attributes)

    if entity_ids is not None and len(entity_ids) == 1:
        if (
            significant_changes_only
            and split_entity_id(entity_ids[0])[0] not in SIGNIFICANT_DOMAINS
        ):
            baked_query += lambda q: q.filter(
                States.last_changed == States.last_updated
            )
    elif significant_changes_only:
        baked_query += lambda q: q.filter(
            or_(
                *[
                    States.entity_id.like(entity_domain)
                    for entity_domain in SIGNIFICANT_DOMAINS_ENTITY_ID_LIKE
                ],
                (States.last_changed == States.last_updated),
            )
        )

    if entity_ids is not None:
        baked_query += lambda q: q.filter(
            States.entity_id.in_(bindparam("entity_ids", expanding=True))
        )
    else:
        baked_query += lambda q: q.filter(
            and_(
                *[
                    ~States.entity_id.like(entity_domain)
                    for entity_domain in IGNORE_DOMAINS_ENTITY_ID_LIKE
                ]
            )
        )
        if filters:
            filters.bake(baked_query)

    baked_query += lambda q: q.filter(States.last_updated > bindparam("start_time"))
    if end_time is not None:
        baked_query += lambda q: q.filter(States.last_updated < bindparam("end_time"))

    if join_attributes:
        baked_query += lambda q: q.outerjoin(
            StateAttributes, States.attributes_id == StateAttributes.attributes_id
        )
    baked_query += lambda q: q.order_by(States.entity_id, States.last_updated)

    states = execute(
        baked_query(session).params(
            start_time=start_time, end_time=end_time, entity_ids=entity_ids
        )
    )

    if _LOGGER.isEnabledFor(logging.DEBUG):
        elapsed = time.perf_counter() - timer_start
        _LOGGER.debug("get_significant_states took %fs", elapsed)

    return states


def get_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]:
    """
    Return states changes during UTC period start_time - end_time.

    entity_ids is an optional iterable of entities to include in the results.

    filters is an optional SQLAlchemy filter which will be applied to the database
    queries unless entity_ids is given, in which case its ignored.

    Significant states are all states where there is a state change,
    as well as all states from certain domains (for instance
    thermostat so that we get current temperature in our graphs).
    """
    states = _query_significant_states_with_session(
        hass,
        session,
        start_time,
        end_time,
        entity_ids,
        filters,
        significant_changes_only,
        no_attributes,
    )
    return _sorted_states_to_dict(
        hass,
        session,
        states,
        start_time,
        entity_ids,
        filters,
        include_start_time_state,
        minimal_response,
        no_attributes,
    )


def get_full_significant_states_with_session(
    hass: HomeAssistant,
    session: Session,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_ids: list[str] | None = None,
    filters: Any = None,
    include_start_time_state: bool = True,
    significant_changes_only: bool = True,
    no_attributes: bool = False,
) -> MutableMapping[str, list[State]]:
    """Variant of get_significant_states_with_session that does not return minimal responses."""
    return cast(
        MutableMapping[str, list[State]],
        get_significant_states_with_session(
            hass=hass,
            session=session,
            start_time=start_time,
            end_time=end_time,
            entity_ids=entity_ids,
            filters=filters,
            include_start_time_state=include_start_time_state,
            significant_changes_only=significant_changes_only,
            minimal_response=False,
            no_attributes=no_attributes,
        ),
    )


def state_changes_during_period(
    hass: HomeAssistant,
    start_time: datetime,
    end_time: datetime | None = None,
    entity_id: str | None = None,
    no_attributes: bool = False,
    descending: bool = False,
    limit: int | None = None,
    include_start_time_state: bool = True,
) -> MutableMapping[str, list[State]]:
    """Return states changes during UTC period start_time - end_time."""
    with session_scope(hass=hass) as session:
        baked_query, join_attributes = bake_query_and_join_attributes(
            hass, no_attributes
        )

        baked_query += lambda q: q.filter(
            (States.last_changed == States.last_updated)
            & (States.last_updated > bindparam("start_time"))
        )

        if end_time is not None:
            baked_query += lambda q: q.filter(
                States.last_updated < bindparam("end_time")
            )

        if entity_id is not None:
            baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
            entity_id = entity_id.lower()

        if join_attributes:
            baked_query += lambda q: q.outerjoin(
                StateAttributes, States.attributes_id == StateAttributes.attributes_id
            )

        last_updated = States.last_updated.desc() if descending else States.last_updated
        baked_query += lambda q: q.order_by(States.entity_id, last_updated)

        if limit:
            baked_query += lambda q: q.limit(limit)

        states = execute(
            baked_query(session).params(
                start_time=start_time, end_time=end_time, entity_id=entity_id
            )
        )

        entity_ids = [entity_id] if entity_id is not None else None

        return cast(
            MutableMapping[str, list[State]],
            _sorted_states_to_dict(
                hass,
                session,
                states,
                start_time,
                entity_ids,
                include_start_time_state=include_start_time_state,
            ),
        )


def get_last_state_changes(
    hass: HomeAssistant, number_of_states: int, entity_id: str
) -> MutableMapping[str, list[State]]:
    """Return the last number_of_states."""
    start_time = dt_util.utcnow()

    with session_scope(hass=hass) as session:
        baked_query, join_attributes = bake_query_and_join_attributes(hass, False)

        baked_query += lambda q: q.filter(States.last_changed == States.last_updated)

        if entity_id is not None:
            baked_query += lambda q: q.filter_by(entity_id=bindparam("entity_id"))
            entity_id = entity_id.lower()

        if join_attributes:
            baked_query += lambda q: q.outerjoin(
                StateAttributes, States.attributes_id == StateAttributes.attributes_id
            )
        baked_query += lambda q: q.order_by(
            States.entity_id, States.last_updated.desc()
        )

        baked_query += lambda q: q.limit(bindparam("number_of_states"))

        states = execute(
            baked_query(session).params(
                number_of_states=number_of_states, entity_id=entity_id
            )
        )

        entity_ids = [entity_id] if entity_id is not None else None

        return cast(
            MutableMapping[str, list[State]],
            _sorted_states_to_dict(
                hass,
                session,
                reversed(states),
                start_time,
                entity_ids,
                include_start_time_state=False,
            ),
        )


def get_states(
    hass: HomeAssistant,
    utc_point_in_time: datetime,
    entity_ids: list[str] | None = None,
    run: RecorderRuns | None = None,
    filters: Any = None,
    no_attributes: bool = False,
) -> list[State]:
    """Return the states at a specific point in time."""
    if (
        run is None
        and (run := (recorder.run_information_from_instance(hass, utc_point_in_time)))
        is None
    ):
        # History did not run before utc_point_in_time
        return []

    with session_scope(hass=hass) as session:
        return _get_states_with_session(
            hass, session, utc_point_in_time, entity_ids, run, filters, no_attributes
        )


def _get_states_with_session(
    hass: HomeAssistant,
    session: Session,
    utc_point_in_time: datetime,
    entity_ids: list[str] | None = None,
    run: RecorderRuns | None = None,
    filters: Any | None = None,
    no_attributes: bool = False,
) -> list[State]:
    """Return the states at a specific point in time."""
    if entity_ids and len(entity_ids) == 1:
        return _get_single_entity_states_with_session(
            hass, session, utc_point_in_time, entity_ids[0], no_attributes
        )

    if (
        run is None
        and (run := (recorder.run_information_with_session(session, utc_point_in_time)))
        is None
    ):
        # History did not run before utc_point_in_time
        return []
{'fraction_non_alphanumeric': 0.07942830365510778, 'fraction_numerical': 0.0011715089034676663, 'mean_word_length': 2.679433497536946, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5991234', 'n_tokens_mistral': 9012, 'n_tokens_neox': 8772, 'n_words': 2257}