"""The tests for sensor recorder platform."""

from collections.abc import Callable, Iterable
from datetime import datetime, timedelta
import logging
import math
from statistics import mean
from typing import Any, Literal
from unittest.mock import ANY, Mock, patch

from freezegun import freeze_time
from freezegun.api import FrozenDateTimeFactory
import pytest

from homeassistant import loader
from homeassistant.components.recorder import (
    CONF_COMMIT_INTERVAL,
    DOMAIN as RECORDER_DOMAIN,
    Recorder,
    history,
)
from homeassistant.components.recorder.db_schema import (
    StateAttributes,
    States,
    StatesMeta,
    StatisticsMeta,
)
from homeassistant.components.recorder.models import (
    StatisticData,
    StatisticMeanType,
    StatisticMetaData,
    process_timestamp,
)
from homeassistant.components.recorder.statistics import (
    DEG_TO_RAD,
    RAD_TO_DEG,
    async_import_statistics,
    get_metadata,
    list_statistic_ids,
)
from homeassistant.components.recorder.util import get_instance, session_scope
from homeassistant.components.sensor import (
    ATTR_OPTIONS,
    DOMAIN,
    SensorDeviceClass,
    SensorStateClass,
)
from homeassistant.components.sensor.recorder import (
    MEAN_TYPE_CHANGED_ISSUE,
    STATE_CLASS_REMOVED_ISSUE,
    UNITS_CHANGED_ISSUE,
)
from homeassistant.const import ATTR_FRIENDLY_NAME, DEGREE, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_system import METRIC_SYSTEM, US_CUSTOMARY_SYSTEM

from .common import MockSensor

from tests.common import mock_platform, setup_test_component_platform
from tests.components.recorder.common import (
    assert_dict_of_states_equal_without_context_and_last_changed,
    assert_multiple_states_equal_without_context_and_last_changed,
    async_recorder_block_till_done,
    async_wait_recording_done,
    db_state_attributes_to_native,
    db_state_to_native,
    do_adhoc_statistics,
    get_start_time,
    statistics_during_period,
)
from tests.typing import (
    MockHAClientWebSocket,
    RecorderInstanceContextManager,
    WebSocketGenerator,
)

BATTERY_SENSOR_ATTRIBUTES = {
    "device_class": "battery",
    "state_class": "measurement",
    "unit_of_measurement": "%",
}
ENERGY_SENSOR_ATTRIBUTES = {
    "device_class": "energy",
    "state_class": "total",
    "unit_of_measurement": "kWh",
}
NONE_SENSOR_ATTRIBUTES = {
    "state_class": "measurement",
}
POWER_SENSOR_ATTRIBUTES = {
    "device_class": "power",
    "state_class": "measurement",
    "unit_of_measurement": "kW",
}
PRESSURE_SENSOR_ATTRIBUTES = {
    "device_class": "pressure",
    "state_class": "measurement",
    "unit_of_measurement": "hPa",
}
TEMPERATURE_SENSOR_ATTRIBUTES = {
    "device_class": "temperature",
    "state_class": "measurement",
    "unit_of_measurement": "°C",
}
GAS_SENSOR_ATTRIBUTES = {
    "device_class": "gas",
    "state_class": "total",
    "unit_of_measurement": "m³",
}
KW_SENSOR_ATTRIBUTES = {
    "state_class": "measurement",
    "unit_of_measurement": "kW",
}
WIND_DIRECTION_ATTRIBUTES = {
    "device_class": SensorDeviceClass.WIND_DIRECTION,
    "state_class": SensorStateClass.MEASUREMENT_ANGLE,
    "unit_of_measurement": DEGREE,
}
WIND_DIRECTION_STATES_SEQ = [350, 0, 15]
TEMP_STATES_SEQ = [-10, 15, 30, 60]


@pytest.fixture
async def mock_recorder_before_hass(
    async_test_recorder: RecorderInstanceContextManager,
) -> None:
    """Set up recorder patches."""


@pytest.fixture(autouse=True)
def setup_recorder(recorder_mock: Recorder) -> Recorder:
    """Set up recorder."""


@pytest.fixture(autouse=True)
def disable_mariadb_issue() -> None:
    """Disable creating issue about outdated MariaDB version."""
    with patch(
        "homeassistant.components.recorder.util._async_create_mariadb_range_index_regression_issue"
    ):
        yield


async def async_list_statistic_ids(
    hass: HomeAssistant,
    statistic_ids: set[str] | None = None,
    statistic_type: Literal["mean", "sum"] | None = None,
) -> list[dict]:
    """Return all statistic_ids and unit of measurement."""
    return await hass.async_add_executor_job(
        list_statistic_ids, hass, statistic_ids, statistic_type
    )


async def assert_statistic_ids(
    hass: HomeAssistant,
    expected_result: list[dict[str, Any]],
) -> None:
    """Assert statistic ids."""
    with session_scope(hass=hass, read_only=True) as session:
        db_states = list(session.query(StatisticsMeta))
        assert len(db_states) == len(expected_result)
        for i, db_state in enumerate(db_states):
            assert db_state.statistic_id == expected_result[i]["statistic_id"]
            assert (
                db_state.unit_of_measurement
                == expected_result[i]["unit_of_measurement"]
            )


def assert_issues(
    hass: HomeAssistant,
    expected_issues: dict[str, dict[str, Any]],
) -> None:
    """Assert statistics issues."""
    issue_registry = ir.async_get(hass)
    assert len(issue_registry.issues) == len(expected_issues)
    for issue_id, expected_issue_data in expected_issues.items():
        expected_translation_placeholders = dict(expected_issue_data)
        expected_translation_placeholders.pop("issue_type")
        expected_issue = ir.IssueEntry(
            active=True,
            breaks_in_ha_version=None,
            created=ANY,
            data=expected_issue_data,
            dismissed_version=None,
            domain=DOMAIN,
            is_fixable=False,
            is_persistent=False,
            issue_domain=None,
            issue_id=issue_id,
            learn_more_url=None,
            severity=ir.IssueSeverity.WARNING,
            translation_key=expected_issue_data["issue_type"],
            translation_placeholders=expected_translation_placeholders,
        )
        assert (DOMAIN, issue_id) in issue_registry.issues
        assert issue_registry.issues[(DOMAIN, issue_id)] == expected_issue


async def assert_validation_result(
    hass: HomeAssistant,
    client: MockHAClientWebSocket,
    expected_validation_result: dict[str, list[dict[str, Any]]],
    expected_issues: Iterable[str],
) -> None:
    """Assert statistics validation result."""
    await client.send_json_auto_id({"type": "recorder/validate_statistics"})
    response = await client.receive_json()
    assert response["success"]
    assert response["result"] == expected_validation_result
    await hass.async_block_till_done()

    # Check we get corresponding issues
    await client.send_json_auto_id({"type": "recorder/update_statistics_issues"})
    response = await client.receive_json()
    assert response["success"]
    expected_issue_registry_issues = {
        f"{issue['type']}_{statistic_id}": issue["data"] | {"issue_type": issue["type"]}
        for statistic_id, issues in expected_validation_result.items()
        for issue in issues
        if issue["type"] in expected_issues
    }

    assert_issues(hass, expected_issue_registry_issues)


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        (None, "%", "%", "%", "unitless", 13.050847, -10, 30),
        (None, "ppm", "ppm", "ppm", "unitless", 13.050847, -10, 30),
        (None, "g/m³", "g/m³", "g/m³", "concentration", 13.050847, -10, 30),
        (None, "mg/m³", "mg/m³", "mg/m³", "concentration", 13.050847, -10, 30),
        ("area", "m²", "m²", "m²", "area", 13.050847, -10, 30),
        ("area", "mi²", "mi²", "mi²", "area", 13.050847, -10, 30),
        ("battery", "%", "%", "%", "unitless", 13.050847, -10, 30),
        ("battery", None, None, None, "unitless", 13.050847, -10, 30),
        (
            "carbon_monoxide",
            "mg/m³",
            "mg/m³",
            "mg/m³",
            "carbon_monoxide",
            13.050847,
            -10,
            30,
        ),
        ("carbon_monoxide", "ppm", "ppm", "ppm", "carbon_monoxide", 13.050847, -10, 30),
        ("distance", "m", "m", "m", "distance", 13.050847, -10, 30),
        ("distance", "mi", "mi", "mi", "distance", 13.050847, -10, 30),
        ("humidity", "%", "%", "%", "unitless", 13.050847, -10, 30),
        ("humidity", None, None, None, "unitless", 13.050847, -10, 30),
        ("pressure", "mPa", "mPa", "mPa", "pressure", 13.050847, -10, 30),
        ("pressure", "Pa", "Pa", "Pa", "pressure", 13.050847, -10, 30),
        ("pressure", "hPa", "hPa", "hPa", "pressure", 13.050847, -10, 30),
        ("pressure", "mbar", "mbar", "mbar", "pressure", 13.050847, -10, 30),
        ("pressure", "inHg", "inHg", "inHg", "pressure", 13.050847, -10, 30),
        ("pressure", "psi", "psi", "psi", "pressure", 13.050847, -10, 30),
        ("speed", "m/s", "m/s", "m/s", "speed", 13.050847, -10, 30),
        ("speed", "mph", "mph", "mph", "speed", 13.050847, -10, 30),
        ("temperature", "°C", "°C", "°C", "temperature", 13.050847, -10, 30),
        ("temperature", "°F", "°F", "°F", "temperature", 13.050847, -10, 30),
        (
            "temperature_delta",
            "°C",
            "°C",
            "°C",
            "temperature_delta",
            13.050847,
            -10,
            30,
        ),
        (
            "temperature_delta",
            "°F",
            "°F",
            "°F",
            "temperature_delta",
            13.050847,
            -10,
            30,
        ),
        ("volume", "m³", "m³", "m³", "volume", 13.050847, -10, 30),
        ("volume", "ft³", "ft³", "ft³", "volume", 13.050847, -10, 30),
        ("weight", "g", "g", "g", "mass", 13.050847, -10, 30),
        ("weight", "oz", "oz", "oz", "mass", 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_statistics_angle(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test compiling hourly statistics for measurement_angle."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass,
            freezer,
            zero,
            "sensor.test1",
            WIND_DIRECTION_ATTRIBUTES,
            seq=WIND_DIRECTION_STATES_SEQ,
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(0.5802544),
                "min": None,
                "max": None,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        ("temperature", "°C", "°C", "°C", "temperature", 27.796610169491526, -10, 60),
        ("temperature", "°F", "°F", "°F", "temperature", 27.796610169491526, -10, 60),
    ],
)
async def test_compile_hourly_statistics_with_some_same_last_updated(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics with the some of the same last updated value.

    If the last updated value is the same we will have a zero duration.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    entity_id = "sensor.test1"
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    attributes = dict(attributes)
    seq = TEMP_STATES_SEQ

    async def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        await async_wait_recording_done(hass)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=10 * 5)
    three = two + timedelta(seconds=40 * 5)
    four = three + timedelta(seconds=10 * 5)

    states = {entity_id: []}
    with freeze_time(one) as freezer:
        states[entity_id].append(
            await set_state(entity_id, str(seq[0]), attributes=attributes)
        )

        # Record two states at the exact same time
        freezer.move_to(two)
        states[entity_id].append(
            await set_state(entity_id, str(seq[1]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[2]), attributes=attributes)
        )

        freezer.move_to(three)
        states[entity_id].append(
            await set_state(entity_id, str(seq[3]), attributes=attributes)
        )

    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_statistics_with_some_same_last_updated_angle(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test compiling stats with same last_updated for angle.

    If the last updated value is the same we will have a zero duration.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    entity_id = "sensor.test1"
    seq = [350, 2, 15, 345]

    async def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        await async_wait_recording_done(hass)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=10 * 5)
    three = two + timedelta(seconds=40 * 5)
    four = three + timedelta(seconds=10 * 5)

    states = {entity_id: []}
    with freeze_time(one) as freezer:
        states[entity_id].append(
            await set_state(
                entity_id, str(seq[0]), attributes=WIND_DIRECTION_ATTRIBUTES
            )
        )

        # Record two states at the exact same time
        freezer.move_to(two)
        states[entity_id].append(
            await set_state(
                entity_id, str(seq[1]), attributes=WIND_DIRECTION_ATTRIBUTES
            )
        )
        states[entity_id].append(
            await set_state(
                entity_id, str(seq[2]), attributes=WIND_DIRECTION_ATTRIBUTES
            )
        )

        freezer.move_to(three)
        states[entity_id].append(
            await set_state(
                entity_id, str(seq[3]), attributes=WIND_DIRECTION_ATTRIBUTES
            )
        )

    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(6.274605),
                "min": None,
                "max": None,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "attributes",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
        "mean_type",
        "seq",
    ),
    [
        (
            {
                "device_class": "temperature",
                "state_class": "measurement",
                "unit_of_measurement": "°C",
            },
            "°C",
            "°C",
            "temperature",
            60,
            -10,
            60,
            StatisticMeanType.ARITHMETIC,
            TEMP_STATES_SEQ,
        ),
        (
            {
                "device_class": "temperature",
                "state_class": "measurement",
                "unit_of_measurement": "°F",
            },
            "°F",
            "°F",
            "temperature",
            60,
            -10,
            60,
            StatisticMeanType.ARITHMETIC,
            TEMP_STATES_SEQ,
        ),
        (
            WIND_DIRECTION_ATTRIBUTES,
            DEGREE,
            DEGREE,
            None,
            15,
            None,
            None,
            StatisticMeanType.CIRCULAR,
            [350, 0, 355, 15],
        ),
    ],
)
async def test_compile_hourly_statistics_with_all_same_last_updated(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    attributes: dict[str, Any],
    display_unit: str,
    statistics_unit: str,
    unit_class: str | None,
    mean: float | None,
    min: float | None,
    max: float | None,
    mean_type: StatisticMeanType,
    seq: list[float],
) -> None:
    """Test compiling hourly statistics with the all of the same last updated value.

    If the last updated value is the same we will have a zero duration.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    entity_id = "sensor.test1"

    async def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        await async_wait_recording_done(hass)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=10 * 5)
    three = two + timedelta(seconds=40 * 5)
    four = three + timedelta(seconds=10 * 5)

    states = {entity_id: []}
    with freeze_time(two):
        states[entity_id].append(
            await set_state(entity_id, str(seq[0]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[1]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[2]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[3]), attributes=attributes)
        )

    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": mean_type is StatisticMeanType.ARITHMETIC,
            "mean_type": mean_type,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "attributes",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
        "mean_type",
        "seq",
    ),
    [
        (
            {
                "device_class": "temperature",
                "state_class": "measurement",
                "unit_of_measurement": "°C",
            },
            "°C",
            "°C",
            "temperature",
            60,
            -10,
            60,
            StatisticMeanType.ARITHMETIC,
            TEMP_STATES_SEQ,
        ),
        (
            {
                "device_class": "temperature",
                "state_class": "measurement",
                "unit_of_measurement": "°F",
            },
            "°F",
            "°F",
            "temperature",
            60,
            -10,
            60,
            StatisticMeanType.ARITHMETIC,
            TEMP_STATES_SEQ,
        ),
        (
            WIND_DIRECTION_ATTRIBUTES,
            DEGREE,
            DEGREE,
            None,
            15,
            None,
            None,
            StatisticMeanType.CIRCULAR,
            [350, 0, 355, 15],
        ),
    ],
)
async def test_compile_hourly_statistics_only_state_is_at_end_of_period(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    attributes: dict[str, Any],
    display_unit: str,
    statistics_unit: str,
    unit_class: str | None,
    mean: float | None,
    min: float | None,
    max: float | None,
    mean_type: StatisticMeanType,
    seq: list[float],
) -> None:
    """Test compiling hourly statistics when the only states are at end of period."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    entity_id = "sensor.test1"

    async def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        await async_wait_recording_done(hass)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=10 * 5)
    three = two + timedelta(seconds=40 * 5)
    four = three + timedelta(seconds=10 * 5)
    end = zero + timedelta(minutes=5)

    states = {entity_id: []}
    with freeze_time(end):
        states[entity_id].append(
            await set_state(entity_id, str(seq[0]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[1]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[2]), attributes=attributes)
        )
        states[entity_id].append(
            await set_state(entity_id, str(seq[3]), attributes=attributes)
        )

    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=5))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": mean_type is StatisticMeanType.ARITHMETIC,
            "mean_type": mean_type,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("device_class", "state_unit", "display_unit", "statistics_unit", "unit_class"),
    [
        (None, "%", "%", "%", "unitless"),
    ],
)
async def test_compile_hourly_statistics_purged_state_changes(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
) -> None:
    """Test compiling hourly statistics.

    This tests statistics falls back to the state machine when states are purged.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    mean = min_value = max_value = float(hist["sensor.test1"][-1].state)

    # Purge all states from the database
    with freeze_time(four):
        await hass.services.async_call("recorder", "purge", {"keep_days": 0})
        await hass.async_block_till_done()
        await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert not hist

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min_value),
                "max": pytest.approx(max_value),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        (None, "%", "%", "%", "unitless", 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics_ignore_future_state(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics.

    This tests statistics does not fall back to the state machine if the state
    in the state machine is newer than the end of the statistics period.
    """
    zero = get_start_time(dt_util.utcnow() + timedelta(minutes=5))
    previous_period = zero - timedelta(minutes=5)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=previous_period)
    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, previous_period, period="5minute")
    # Check we get no stats from the previous period
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize("attributes", [TEMPERATURE_SENSOR_ATTRIBUTES])
async def test_compile_hourly_statistics_wrong_unit(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    attributes,
) -> None:
    """Test stats for sensor with unit not matching device class."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )

        attributes_tmp = dict(attributes)
        attributes_tmp["unit_of_measurement"] = "invalid"
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test2", attributes_tmp
        )
        states = {**states, **_states}
        attributes_tmp.pop("unit_of_measurement")
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test3", attributes_tmp
        )
        states = {**states, **_states}

        attributes_tmp = dict(attributes)
        attributes_tmp["state_class"] = "invalid"
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test4", attributes_tmp
        )
        states = {**states, **_states}
        attributes_tmp.pop("state_class")
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test5", attributes_tmp
        )
        states = {**states, **_states}

        attributes_tmp = dict(attributes)
        attributes_tmp["device_class"] = "invalid"
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test6", attributes_tmp
        )
        states = {**states, **_states}
        attributes_tmp.pop("device_class")
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test7", attributes_tmp
        )
        states = {**states, **_states}
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": "°C",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "°C",
            "unit_class": "temperature",
        },
        {
            "display_unit_of_measurement": "invalid",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistic_id": "sensor.test2",
            "statistics_unit_of_measurement": "invalid",
            "unit_class": None,
        },
        {
            "display_unit_of_measurement": None,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistic_id": "sensor.test3",
            "statistics_unit_of_measurement": None,
            "unit_class": "unitless",
        },
        {
            "statistic_id": "sensor.test6",
            "display_unit_of_measurement": "°C",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "°C",
            "unit_class": "temperature",
        },
        {
            "statistic_id": "sensor.test7",
            "display_unit_of_measurement": "°C",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "°C",
            "unit_class": "temperature",
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(13.050847),
                "min": pytest.approx(-10.0),
                "max": pytest.approx(30.0),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ],
        "sensor.test2": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": 13.05084745762712,
                "min": -10.0,
                "max": 30.0,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ],
        "sensor.test3": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": 13.05084745762712,
                "min": -10.0,
                "max": 30.0,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ],
        "sensor.test6": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(13.050847),
                "min": pytest.approx(-10.0),
                "max": pytest.approx(30.0),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ],
        "sensor.test7": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(13.050847),
                "min": pytest.approx(-10.0),
                "max": pytest.approx(30.0),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ],
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize("state_class", ["total"])
@pytest.mark.parametrize(
    (
        "units",
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        (US_CUSTOMARY_SYSTEM, "area", "m²", "m²", "m²", "area", 1),
        (US_CUSTOMARY_SYSTEM, "area", "mi²", "mi²", "mi²", "area", 1),
        (US_CUSTOMARY_SYSTEM, "distance", "m", "m", "m", "distance", 1),
        (US_CUSTOMARY_SYSTEM, "distance", "mi", "mi", "mi", "distance", 1),
        (US_CUSTOMARY_SYSTEM, "energy", "kWh", "kWh", "kWh", "energy", 1),
        (US_CUSTOMARY_SYSTEM, "energy", "Wh", "Wh", "Wh", "energy", 1),
        (US_CUSTOMARY_SYSTEM, "gas", "m³", "m³", "m³", "volume", 1),
        (US_CUSTOMARY_SYSTEM, "gas", "ft³", "ft³", "ft³", "volume", 1),
        (US_CUSTOMARY_SYSTEM, "monetary", "EUR", "EUR", "EUR", None, 1),
        (US_CUSTOMARY_SYSTEM, "monetary", "SEK", "SEK", "SEK", None, 1),
        (US_CUSTOMARY_SYSTEM, "volume", "m³", "m³", "m³", "volume", 1),
        (US_CUSTOMARY_SYSTEM, "volume", "ft³", "ft³", "ft³", "volume", 1),
        (US_CUSTOMARY_SYSTEM, "weight", "g", "g", "g", "mass", 1),
        (US_CUSTOMARY_SYSTEM, "weight", "oz", "oz", "oz", "mass", 1),
        (METRIC_SYSTEM, "area", "m²", "m²", "m²", "area", 1),
        (METRIC_SYSTEM, "area", "mi²", "mi²", "mi²", "area", 1),
        (METRIC_SYSTEM, "distance", "m", "m", "m", "distance", 1),
        (METRIC_SYSTEM, "distance", "mi", "mi", "mi", "distance", 1),
        (METRIC_SYSTEM, "energy", "kWh", "kWh", "kWh", "energy", 1),
        (METRIC_SYSTEM, "energy", "Wh", "Wh", "Wh", "energy", 1),
        (METRIC_SYSTEM, "gas", "m³", "m³", "m³", "volume", 1),
        (METRIC_SYSTEM, "gas", "ft³", "ft³", "ft³", "volume", 1),
        (METRIC_SYSTEM, "monetary", "EUR", "EUR", "EUR", None, 1),
        (METRIC_SYSTEM, "monetary", "SEK", "SEK", "SEK", None, 1),
        (METRIC_SYSTEM, "volume", "m³", "m³", "m³", "volume", 1),
        (METRIC_SYSTEM, "volume", "ft³", "ft³", "ft³", "volume", 1),
        (METRIC_SYSTEM, "weight", "g", "g", "g", "mass", 1),
        (METRIC_SYSTEM, "weight", "oz", "oz", "oz", "mass", 1),
    ],
)
async def test_compile_hourly_sum_statistics_amount(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    caplog: pytest.LogCaptureFixture,
    units,
    state_class,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
        "last_reset": None,
    }
    seq = [10, 15, 20, 10, 30, 40, 50, 60, 70]
    with freeze_time(period0) as freezer:
        four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": statistics_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    expected_stats = {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(period0).timestamp(),
                "state": pytest.approx(factor * seq[2]),
                "sum": pytest.approx(factor * 10.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(factor * seq[5]),
                "sum": pytest.approx(factor * 40.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(factor * seq[8]),
                "sum": pytest.approx(factor * 70.0),
            },
        ]
    }
    assert stats == expected_stats

    # With an offset of 1 minute, we expect to get the 2nd and 3rd periods
    stats = statistics_during_period(
        hass, period0 + timedelta(minutes=1), period="5minute"
    )
    assert stats == {"sensor.test1": expected_stats["sensor.test1"][1:3]}

    # With an offset of 5 minutes, we expect to get the 2nd and 3rd periods
    stats = statistics_during_period(
        hass, period0 + timedelta(minutes=5), period="5minute"
    )
    assert stats == {"sensor.test1": expected_stats["sensor.test1"][1:3]}

    # With an offset of 6 minutes, we expect to get the 3rd period
    stats = statistics_during_period(
        hass, period0 + timedelta(minutes=6), period="5minute"
    )
    assert stats == {"sensor.test1": expected_stats["sensor.test1"][2:3]}

    assert "Error while processing event StatisticsTask" not in caplog.text
    assert "Detected new cycle for sensor.test1, last_reset set to" in caplog.text
    assert "Compiling initial sum statistics for sensor.test1" in caplog.text
    assert "Detected new cycle for sensor.test1, value dropped" not in caplog.text

    client = await hass_ws_client()

    # Adjust the inserted statistics
    await client.send_json(
        {
            "id": 1,
            "type": "recorder/adjust_sum_statistics",
            "statistic_id": "sensor.test1",
            "start_time": period1.isoformat(),
            "adjustment": 100.0,
            "adjustment_unit_of_measurement": display_unit,
        }
    )
    response = await client.receive_json()
    assert response["success"]
    await async_wait_recording_done(hass)

    expected_stats["sensor.test1"][1]["sum"] = pytest.approx(factor * 40.0 + 100)
    expected_stats["sensor.test1"][2]["sum"] = pytest.approx(factor * 70.0 + 100)
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == expected_stats

    # Adjust the inserted statistics
    await client.send_json(
        {
            "id": 2,
            "type": "recorder/adjust_sum_statistics",
            "statistic_id": "sensor.test1",
            "start_time": period2.isoformat(),
            "adjustment": -400.0,
            "adjustment_unit_of_measurement": display_unit,
        }
    )
    response = await client.receive_json()
    assert response["success"]
    await async_wait_recording_done(hass)

    expected_stats["sensor.test1"][1]["sum"] = pytest.approx(factor * 40.0 + 100)
    expected_stats["sensor.test1"][2]["sum"] = pytest.approx(factor * 70.0 - 300)
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == expected_stats


@pytest.mark.parametrize("state_class", ["total"])
@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
        ("energy", "Wh", "Wh", "Wh", "energy", 1),
        ("monetary", "EUR", "EUR", "EUR", None, 1),
        ("monetary", "SEK", "SEK", "SEK", None, 1),
        ("gas", "m³", "m³", "m³", "volume", 1),
        ("gas", "ft³", "ft³", "ft³", "volume", 1),
        ("weight", "kg", "kg", "kg", "mass", 1),
    ],
)
async def test_compile_hourly_sum_statistics_amount_reset_every_state_change(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    state_class,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
        "last_reset": None,
    }
    seq = [10, 15, 15, 15, 20, 20, 20, 25]
    # Make sure the sequence has consecutive equal states
    assert seq[1] == seq[2] == seq[3]

    # Make sure the first and last state differ
    assert seq[0] != seq[-1]

    states = {"sensor.test1": []}
    with freeze_time(zero) as freezer:
        # Insert states for a 1st statistics period
        one = zero
        for i in range(len(seq)):
            one = one + timedelta(seconds=5)
            attributes = dict(attributes)
            attributes["last_reset"] = dt_util.as_local(one).isoformat()
            _states = await async_record_meter_state(
                hass, freezer, one, "sensor.test1", attributes, seq[i : i + 1]
            )
            states["sensor.test1"].extend(_states["sensor.test1"])

        # Insert states for a 2nd statistics period
        two = zero + timedelta(minutes=5)
        for i in range(len(seq)):
            two = two + timedelta(seconds=5)
            attributes = dict(attributes)
            attributes["last_reset"] = dt_util.as_local(two).isoformat()
            _states = await async_record_meter_state(
                hass, freezer, two, "sensor.test1", attributes, seq[i : i + 1]
            )
            states["sensor.test1"].extend(_states["sensor.test1"])
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass,
        zero - timedelta.resolution,
        two + timedelta.resolution,
        hass.states.async_entity_ids(),
        significant_changes_only=False,
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=zero)
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=5))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(dt_util.as_local(one)).timestamp(),
                "state": pytest.approx(factor * seq[7]),
                "sum": pytest.approx(factor * (sum(seq) - seq[0])),
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(dt_util.as_local(two)).timestamp(),
                "state": pytest.approx(factor * seq[7]),
                "sum": pytest.approx(factor * (2 * sum(seq) - seq[0])),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize("state_class", ["total"])
@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
    ],
)
async def test_compile_hourly_sum_statistics_amount_invalid_last_reset(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    state_class,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
        "last_reset": None,
    }
    seq = [10, 15, 15, 15, 20, 20, 20, 25]

    states = {"sensor.test1": []}

    # Insert states
    with freeze_time(zero) as freezer:
        one = zero
        for i in range(len(seq)):
            one = one + timedelta(seconds=5)
            attributes = dict(attributes)
            attributes["last_reset"] = dt_util.as_local(one).isoformat()
            if i == 3:
                attributes["last_reset"] = "festivus"  # not a valid time
            _states = await async_record_meter_state(
                hass, freezer, one, "sensor.test1", attributes, seq[i : i + 1]
            )
            states["sensor.test1"].extend(_states["sensor.test1"])
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass,
        zero - timedelta.resolution,
        one + timedelta.resolution,
        hass.states.async_entity_ids(),
        significant_changes_only=False,
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(dt_util.as_local(one)).timestamp(),
                "state": pytest.approx(factor * seq[7]),
                "sum": pytest.approx(factor * (sum(seq) - seq[0] - seq[3])),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text
    assert "Ignoring invalid last reset 'festivus' for sensor.test1" in caplog.text


@pytest.mark.parametrize("state_class", ["total"])
@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
    ],
)
async def test_compile_hourly_sum_statistics_nan_inf_state(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    state_class,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics with nan and inf states."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
        "last_reset": None,
    }
    seq = [10, math.nan, 15, 15, 20, math.inf, 20, 10]

    states = {"sensor.test1": []}
    with freeze_time(zero) as freezer:
        one = zero
        for i in range(len(seq)):
            one = one + timedelta(seconds=5)
            attributes = dict(attributes)
            attributes["last_reset"] = dt_util.as_local(one).isoformat()
            _states = await async_record_meter_state(
                hass, freezer, one, "sensor.test1", attributes, seq[i : i + 1]
            )
            states["sensor.test1"].extend(_states["sensor.test1"])
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass,
        zero - timedelta.resolution,
        one + timedelta.resolution,
        hass.states.async_entity_ids(),
        significant_changes_only=False,
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(one).timestamp(),
                "state": pytest.approx(factor * seq[7]),
                "sum": pytest.approx(
                    factor * (seq[2] + seq[3] + seq[4] + seq[6] + seq[7])
                ),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "entity_id",
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "offset",
        "warning_1",
        "warning_2",
    ),
    [
        (
            "sensor.test1",
            "energy",
            "kWh",
            "kWh",
            "kWh",
            "energy",
            0,
            "",
            "bug report at https://github.com/home-assistant/core/issues?q=is%3Aopen+is%3Aissue",
        ),
        (
            "sensor.power_consumption",
            "power",
            "W",
            "W",
            "W",
            "power",
            15,
            "from integration demo ",
            "bug report at https://github.com/home-assistant/core/issues?q=is%3Aopen+is%3Aissue+label%3A%22integration%3A+demo%22",
        ),
        (
            "sensor.custom_sensor",
            "energy",
            "kWh",
            "kWh",
            "kWh",
            "energy",
            0,
            "from integration test ",
            "report it to the author of the 'test' custom integration",
        ),
    ],
)
@pytest.mark.parametrize("state_class", ["total_increasing"])
async def test_compile_hourly_sum_statistics_negative_state(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    entity_id,
    warning_1,
    warning_2,
    state_class,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    offset,
) -> None:
    """Test compiling hourly statistics with negative states."""
    zero = get_start_time(dt_util.utcnow())
    hass.data.pop(loader.DATA_CUSTOM_COMPONENTS)

    mocksensor = MockSensor(name="custom_sensor")
    mocksensor._attr_should_poll = False
    setup_test_component_platform(hass, DOMAIN, [mocksensor], built_in=False)

    await async_setup_component(hass, "homeassistant", {})
    with freeze_time(zero) as freezer:
        await async_setup_component(
            hass, "sensor", {"sensor": [{"platform": "demo"}, {"platform": "test"}]}
        )
        await hass.async_block_till_done()
    attributes = {
        "device_class": device_class,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
    }
    seq = [15, 16, 15, 16, 20, -20, 20, 10]

    states = {entity_id: []}
    offending_state = 5
    if state := hass.states.get(entity_id):
        states[entity_id].append(state)
        offending_state = 6
    one = zero
    with freeze_time(zero) as freezer:
        for i in range(len(seq)):
            one = one + timedelta(seconds=5)
            _states = await async_record_meter_state(
                hass, freezer, one, entity_id, attributes, seq[i : i + 1]
            )
            states[entity_id].extend(_states[entity_id])
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass,
        zero - timedelta.resolution,
        one + timedelta.resolution,
        hass.states.async_entity_ids(),
        significant_changes_only=False,
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)[entity_id], dict(hist)[entity_id]
    )

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert {
        "display_unit_of_measurement": display_unit,
        "has_mean": False,
        "mean_type": StatisticMeanType.NONE,
        "has_sum": True,
        "name": None,
        "source": "recorder",
        "statistic_id": entity_id,
        "statistics_unit_of_measurement": statistics_unit,
        "unit_class": unit_class,
    } in statistic_ids
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats[entity_id] == [
        {
            "start": process_timestamp(zero).timestamp(),
            "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
            "max": None,
            "mean": None,
            "min": None,
            "last_reset": None,
            "state": pytest.approx(seq[7]),
            "sum": pytest.approx(offset + 15),  # (20 - 15) + (10 - 0)
        },
    ]
    assert "Error while processing event StatisticsTask" not in caplog.text
    state = states[entity_id][offending_state].state
    last_updated = states[entity_id][offending_state].last_updated.isoformat()
    assert (
        f"Entity {entity_id} {warning_1}has state class"
        f" total_increasing, but its state "
        f"is negative. Triggered by state {state}"
        f" with last_updated set to {last_updated}." in caplog.text
    )
    assert warning_2 in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
        ("energy", "Wh", "Wh", "Wh", "energy", 1),
        ("monetary", "EUR", "EUR", "EUR", None, 1),
        ("monetary", "SEK", "SEK", "SEK", None, 1),
        ("gas", "m³", "m³", "m³", "volume", 1),
        ("gas", "ft³", "ft³", "ft³", "volume", 1),
        ("weight", "kg", "kg", "kg", "mass", 1),
    ],
)
async def test_compile_hourly_sum_statistics_total_no_reset(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "total",
        "unit_of_measurement": state_unit,
    }
    seq = [10, 15, 20, 10, 30, 40, 50, 60, 70]
    with freeze_time(period0) as freezer:
        _four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[2]),
                "sum": pytest.approx(factor * 10.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[5]),
                "sum": pytest.approx(factor * 30.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[8]),
                "sum": pytest.approx(factor * 60.0),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
        ("energy", "Wh", "Wh", "Wh", "energy", 1),
        ("gas", "m³", "m³", "m³", "volume", 1),
        ("gas", "ft³", "ft³", "ft³", "volume", 1),
        ("weight", "kg", "kg", "kg", "mass", 1),
    ],
)
async def test_compile_hourly_sum_statistics_total_increasing(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test compiling hourly statistics."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "total_increasing",
        "unit_of_measurement": state_unit,
    }
    seq = [10, 15, 20, 10, 30, 40, 50, 60, 70]
    with freeze_time(period0) as freezer:
        _four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[2]),
                "sum": pytest.approx(factor * 10.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[5]),
                "sum": pytest.approx(factor * 50.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": None,
                "state": pytest.approx(factor * seq[8]),
                "sum": pytest.approx(factor * 80.0),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text
    assert "Detected new cycle for sensor.test1, last_reset set to" not in caplog.text
    assert "Compiling initial sum statistics for sensor.test1" in caplog.text
    assert "Detected new cycle for sensor.test1, value dropped" in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "factor",
    ),
    [
        ("energy", "kWh", "kWh", "kWh", "energy", 1),
        ("weight", "kg", "kg", "kg", "mass", 1),
    ],
)
async def test_compile_hourly_sum_statistics_total_increasing_small_dip(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    factor,
) -> None:
    """Test small dips in sensor readings do not trigger a reset."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "total_increasing",
        "unit_of_measurement": state_unit,
    }
    seq = [10, 15, 20, 19, 30, 40, 39, 60, 70]
    with freeze_time(period0) as freezer:
        _four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    assert (
        "Entity sensor.test1 has state class total_increasing, but its state is not "
        "strictly increasing."
    ) not in caplog.text
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    state = states["sensor.test1"][6].state
    previous_state = float(states["sensor.test1"][5].state)
    last_updated = states["sensor.test1"][6].last_updated.isoformat()
    assert (
        "Entity sensor.test1 has state class total_increasing, but its state is not "
        f"strictly increasing. Triggered by state {state}"
        f" (previous state: {previous_state}) "
        f"with last_updated set to {last_updated}. Please create a bug report at "
        "https://github.com/home-assistant/core/issues?q=is%3Aopen+is%3Aissue"
    ) in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        }
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "last_reset": None,
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "state": pytest.approx(factor * seq[2]),
                "sum": pytest.approx(factor * 10.0),
            },
            {
                "last_reset": None,
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "state": pytest.approx(factor * seq[5]),
                "sum": pytest.approx(factor * 30.0),
            },
            {
                "last_reset": None,
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "state": pytest.approx(factor * seq[8]),
                "sum": pytest.approx(factor * 60.0),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_energy_statistics_unsupported(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test compiling hourly statistics."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    sns1_attr = {
        "device_class": "energy",
        "state_class": "total",
        "unit_of_measurement": "kWh",
        "last_reset": None,
    }
    sns2_attr = {"device_class": "energy"}
    sns3_attr = {}
    seq1 = [10, 15, 20, 10, 30, 40, 50, 60, 70]
    seq2 = [110, 120, 130, 0, 30, 45, 55, 65, 75]
    seq3 = [0, 0, 5, 10, 30, 50, 60, 80, 90]

    with freeze_time(period0) as freezer:
        four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", sns1_attr, seq1
        )
        _, _, _states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test2", sns2_attr, seq2
        )
        states = {**states, **_states}
        _, _, _states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test3", sns3_attr, seq3
        )
    states = {**states, **_states}
    await async_wait_recording_done(hass)

    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": "kWh",
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "kWh",
            "unit_class": "energy",
        }
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(period0).timestamp(),
                "state": pytest.approx(20.0),
                "sum": pytest.approx(10.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(40.0),
                "sum": pytest.approx(40.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(70.0),
                "sum": pytest.approx(70.0),
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_energy_statistics_multiple(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test compiling multiple hourly statistics."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period2 = period0 + timedelta(minutes=10)
    period2_end = period0 + timedelta(minutes=15)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    sns1_attr = {**ENERGY_SENSOR_ATTRIBUTES, "last_reset": None}
    sns2_attr = {**ENERGY_SENSOR_ATTRIBUTES, "last_reset": None}
    sns3_attr = {
        **ENERGY_SENSOR_ATTRIBUTES,
        "unit_of_measurement": "Wh",
        "last_reset": None,
    }
    seq1 = [10, 15, 20, 10, 30, 40, 50, 60, 70]
    seq2 = [110, 120, 130, 0, 30, 45, 55, 65, 75]
    seq3 = [0, 0, 5, 10, 30, 50, 60, 80, 90]

    with freeze_time(period0) as freezer:
        four, eight, states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test1", sns1_attr, seq1
        )
        _, _, _states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test2", sns2_attr, seq2
        )
        states = {**states, **_states}
        _, _, _states = await async_record_meter_states(
            hass, freezer, period0, "sensor.test3", sns3_attr, seq3
        )
    states = {**states, **_states}
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        period0 - timedelta.resolution,
        eight + timedelta.resolution,
        hass.states.async_entity_ids(),
    )
    assert_multiple_states_equal_without_context_and_last_changed(
        dict(states)["sensor.test1"], dict(hist)["sensor.test1"]
    )

    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period2)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": "kWh",
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "kWh",
            "unit_class": "energy",
        },
        {
            "statistic_id": "sensor.test2",
            "display_unit_of_measurement": "kWh",
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "kWh",
            "unit_class": "energy",
        },
        {
            "statistic_id": "sensor.test3",
            "display_unit_of_measurement": "Wh",
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "Wh",
            "unit_class": "energy",
        },
    ]
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(period0).timestamp(),
                "state": pytest.approx(20.0),
                "sum": pytest.approx(10.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(40.0),
                "sum": pytest.approx(40.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(70.0),
                "sum": pytest.approx(70.0),
            },
        ],
        "sensor.test2": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(period0).timestamp(),
                "state": pytest.approx(130.0),
                "sum": pytest.approx(20.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(45.0),
                "sum": pytest.approx(-65.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(75.0),
                "sum": pytest.approx(-35.0),
            },
        ],
        "sensor.test3": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(period0).timestamp(),
                "state": pytest.approx(5.0),
                "sum": pytest.approx(5.0),
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(50.0),
                "sum": pytest.approx(60.0),
            },
            {
                "start": process_timestamp(period2).timestamp(),
                "end": process_timestamp(period2_end).timestamp(),
                "max": None,
                "mean": None,
                "min": None,
                "last_reset": process_timestamp(four).timestamp(),
                "state": pytest.approx(90.0),
                "sum": pytest.approx(100.0),
            },
        ],
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("device_class", "state_unit", "value"),
    [
        ("battery", "%", 30),
        ("battery", None, 30),
        ("area", "m²", 30),
        ("area", "mi²", 30),
        ("distance", "m", 30),
        ("distance", "mi", 30),
        ("humidity", "%", 30),
        ("humidity", None, 30),
        ("pressure", "mPa", 30),
        ("pressure", "Pa", 30),
        ("pressure", "hPa", 30),
        ("pressure", "mbar", 30),
        ("pressure", "inHg", 30),
        ("pressure", "psi", 30),
        ("speed", "m/s", 30),
        ("speed", "mph", 30),
        ("temperature", "°C", 30),
        ("temperature", "°F", 30),
        ("volume", "m³", 30),
        ("volume", "ft³", 30),
        ("weight", "g", 30),
        ("weight", "oz", 30),
    ],
)
async def test_compile_hourly_statistics_unchanged(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    value,
) -> None:
    """Test compiling hourly statistics, with no changes during the hour."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=four)
    await async_wait_recording_done(hass)
    stats = statistics_during_period(hass, four, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(four).timestamp(),
                "end": process_timestamp(four + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(value),
                "min": pytest.approx(value),
                "max": pytest.approx(value),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_statistics_unchanged_angle(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test compiling hourly stats, no changes for measurement_angle."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass,
            freezer,
            zero,
            "sensor.test1",
            WIND_DIRECTION_ATTRIBUTES,
            seq=WIND_DIRECTION_STATES_SEQ,
        )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=four)
    await async_wait_recording_done(hass)
    stats = statistics_during_period(hass, four, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(four).timestamp(),
                "end": process_timestamp(four + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(15),
                "min": None,
                "max": None,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("attributes", "expected_mean", "expected_min", "expected_max"),
    [
        (TEMPERATURE_SENSOR_ATTRIBUTES, 21.1864406779661, 10.0, 25.0),
        (WIND_DIRECTION_ATTRIBUTES, 21.202479155239875, None, None),
    ],
)
async def test_compile_hourly_statistics_partially_unavailable(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    attributes: dict,
    expected_mean: float,
    expected_min: float | None,
    expected_max: float | None,
) -> None:
    """Test compiling hourly statistics, with the sensor being partially unavailable."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    four, states = await async_record_states_partially_unavailable(
        hass, zero, "sensor.test1", attributes
    )
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(expected_mean),
                "min": pytest.approx(expected_min),
                "max": pytest.approx(expected_max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("device_class", "state_unit", "value"),
    [
        ("battery", "%", 30),
        ("battery", None, 30),
        ("area", "m²", 30),
        ("area", "mi²", 30),
        ("distance", "m", 30),
        ("distance", "mi", 30),
        ("humidity", "%", 30),
        ("humidity", None, 30),
        ("pressure", "mPa", 30),
        ("pressure", "Pa", 30),
        ("pressure", "hPa", 30),
        ("pressure", "mbar", 30),
        ("pressure", "inHg", 30),
        ("pressure", "psi", 30),
        ("speed", "m/s", 30),
        ("speed", "mph", 30),
        ("temperature", "°C", 30),
        ("temperature", "°F", 30),
        ("volume", "m³", 30),
        ("volume", "ft³", 30),
        ("weight", "g", 30),
        ("weight", "oz", 30),
    ],
)
async def test_compile_hourly_statistics_unavailable(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    value,
) -> None:
    """Test compiling hourly statistics, with one sensor being unavailable.

    sensor.test1 is unavailable and should not have statistics generated
    sensor.test2 should have statistics generated
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    four, states = await async_record_states_partially_unavailable(
        hass, zero, "sensor.test1", attributes
    )
    with freeze_time(zero) as freezer:
        _, _states = await async_record_states(
            hass, freezer, zero, "sensor.test2", attributes
        )
    await async_wait_recording_done(hass)
    states = {**states, **_states}
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=four)
    await async_wait_recording_done(hass)
    stats = statistics_during_period(hass, four, period="5minute")
    assert stats == {
        "sensor.test2": [
            {
                "start": process_timestamp(four).timestamp(),
                "end": process_timestamp(four + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(value),
                "min": pytest.approx(value),
                "max": pytest.approx(value),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_statistics_unavailable_angle(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test hourly stats with one sensor unavailable for angle.

    sensor.test1 is unavailable and should not have statistics generated
    sensor.test2 should have statistics generated
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    four, states = await async_record_states_partially_unavailable(
        hass, zero, "sensor.test1", WIND_DIRECTION_ATTRIBUTES
    )
    with freeze_time(zero) as freezer:
        _, _states = await async_record_states(
            hass,
            freezer,
            zero,
            "sensor.test2",
            WIND_DIRECTION_ATTRIBUTES,
            seq=WIND_DIRECTION_STATES_SEQ,
        )
    await async_wait_recording_done(hass)
    states = {**states, **_states}
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=four)
    await async_wait_recording_done(hass)
    stats = statistics_during_period(hass, four, period="5minute")
    assert stats == {
        "sensor.test2": [
            {
                "start": process_timestamp(four).timestamp(),
                "end": process_timestamp(four + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(15),
                "min": None,
                "max": None,
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


async def test_compile_hourly_statistics_fails(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test compiling hourly statistics throws."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    with patch(
        "homeassistant.components.sensor.recorder.compile_statistics",
        side_effect=Exception,
    ):
        do_adhoc_statistics(hass, start=zero)
        await async_wait_recording_done(hass)
    assert "Error while processing event StatisticsTask" in caplog.text


@pytest.mark.parametrize(
    (
        "state_class",
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "statistic_type",
    ),
    [
        ("measurement", "area", "m²", "m²", "m²", "area", StatisticMeanType.ARITHMETIC),
        (
            "measurement",
            "area",
            "mi²",
            "mi²",
            "mi²",
            "area",
            StatisticMeanType.ARITHMETIC,
        ),
        ("total", "area", "m²", "m²", "m²", "area", "sum"),
        ("total", "area", "mi²", "mi²", "mi²", "area", "sum"),
        (
            "measurement",
            "battery",
            "%",
            "%",
            "%",
            "unitless",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "battery",
            None,
            None,
            None,
            "unitless",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "distance",
            "m",
            "m",
            "m",
            "distance",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "distance",
            "mi",
            "mi",
            "mi",
            "distance",
            StatisticMeanType.ARITHMETIC,
        ),
        ("total", "distance", "m", "m", "m", "distance", "sum"),
        ("total", "distance", "mi", "mi", "mi", "distance", "sum"),
        ("total", "energy", "Wh", "Wh", "Wh", "energy", "sum"),
        ("total", "energy", "kWh", "kWh", "kWh", "energy", "sum"),
        (
            "measurement",
            "energy",
            "Wh",
            "Wh",
            "Wh",
            "energy",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "energy",
            "kWh",
            "kWh",
            "kWh",
            "energy",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "humidity",
            "%",
            "%",
            "%",
            "unitless",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "humidity",
            None,
            None,
            None,
            "unitless",
            StatisticMeanType.ARITHMETIC,
        ),
        ("total", "monetary", "USD", "USD", "USD", None, "sum"),
        ("total", "monetary", "None", "None", "None", None, "sum"),
        ("total", "gas", "m³", "m³", "m³", "volume", "sum"),
        ("total", "gas", "ft³", "ft³", "ft³", "volume", "sum"),
        (
            "measurement",
            "monetary",
            "USD",
            "USD",
            "USD",
            None,
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "monetary",
            "None",
            "None",
            "None",
            None,
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "gas",
            "m³",
            "m³",
            "m³",
            "volume",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "gas",
            "ft³",
            "ft³",
            "ft³",
            "volume",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "mPa",
            "mPa",
            "mPa",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "Pa",
            "Pa",
            "Pa",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "hPa",
            "hPa",
            "hPa",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "mbar",
            "mbar",
            "mbar",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "inHg",
            "inHg",
            "inHg",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "pressure",
            "psi",
            "psi",
            "psi",
            "pressure",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "speed",
            "m/s",
            "m/s",
            "m/s",
            "speed",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "speed",
            "mph",
            "mph",
            "mph",
            "speed",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "temperature",
            "°C",
            "°C",
            "°C",
            "temperature",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "temperature",
            "°F",
            "°F",
            "°F",
            "temperature",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "temperature_delta",
            "°C",
            "°C",
            "°C",
            "temperature_delta",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "temperature_delta",
            "°F",
            "°F",
            "°F",
            "temperature_delta",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "volume",
            "m³",
            "m³",
            "m³",
            "volume",
            StatisticMeanType.ARITHMETIC,
        ),
        (
            "measurement",
            "volume",
            "ft³",
            "ft³",
            "ft³",
            "volume",
            StatisticMeanType.ARITHMETIC,
        ),
        ("total", "volume", "m³", "m³", "m³", "volume", "sum"),
        ("total", "volume", "ft³", "ft³", "ft³", "volume", "sum"),
        ("measurement", "weight", "g", "g", "g", "mass", StatisticMeanType.ARITHMETIC),
        (
            "measurement",
            "weight",
            "oz",
            "oz",
            "oz",
            "mass",
            StatisticMeanType.ARITHMETIC,
        ),
        ("total", "weight", "g", "g", "g", "mass", "sum"),
        ("total", "weight", "oz", "oz", "oz", "mass", "sum"),
        (
            SensorStateClass.MEASUREMENT_ANGLE,
            SensorDeviceClass.WIND_DIRECTION,
            DEGREE,
            DEGREE,
            DEGREE,
            None,
            StatisticMeanType.CIRCULAR,
        ),
    ],
)
async def test_list_statistic_ids(
    hass: HomeAssistant,
    state_class: str | SensorStateClass,
    device_class: str | SensorDeviceClass,
    state_unit: str,
    display_unit: str,
    statistics_unit: str,
    unit_class: str | None,
    statistic_type: str | StatisticMeanType,
) -> None:
    """Test listing future statistic ids."""
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "last_reset": 0,
        "state_class": state_class,
        "unit_of_measurement": state_unit,
    }
    hass.states.async_set("sensor.test1", 0, attributes=attributes)
    statistic_ids = await async_list_statistic_ids(hass)
    mean_type = (
        statistic_type
        if isinstance(statistic_type, StatisticMeanType)
        else StatisticMeanType.NONE
    )
    statistic_type = (
        statistic_type if not isinstance(statistic_type, StatisticMeanType) else "mean"
    )
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": mean_type is StatisticMeanType.ARITHMETIC,
            "mean_type": mean_type,
            "has_sum": statistic_type == "sum",
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        },
    ]

    for stat_type in ("mean", "sum", "dogs"):
        statistic_ids = await async_list_statistic_ids(hass, statistic_type=stat_type)
        if statistic_type == stat_type:
            assert statistic_ids == [
                {
                    "statistic_id": "sensor.test1",
                    "display_unit_of_measurement": display_unit,
                    "has_mean": mean_type is StatisticMeanType.ARITHMETIC,
                    "mean_type": mean_type,
                    "has_sum": statistic_type == "sum",
                    "name": None,
                    "source": "recorder",
                    "statistics_unit_of_measurement": statistics_unit,
                    "unit_class": unit_class,
                },
            ]
        else:
            assert statistic_ids == []


@pytest.mark.parametrize(
    "energy_attributes",
    [{**ENERGY_SENSOR_ATTRIBUTES, "last_reset": 0}, TEMPERATURE_SENSOR_ATTRIBUTES],
)
async def test_list_statistic_ids_unsupported(
    hass: HomeAssistant,
    energy_attributes: dict[str, Any],
) -> None:
    """Test listing future statistic ids for unsupported sensor."""
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = dict(energy_attributes)
    hass.states.async_set("sensor.test1", 0, attributes=attributes)
    if "last_reset" in attributes:
        attributes.pop("unit_of_measurement")
        hass.states.async_set("last_reset.test2", 0, attributes=attributes)
    attributes = dict(energy_attributes)
    if "unit_of_measurement" in attributes:
        attributes["unit_of_measurement"] = "invalid"
        hass.states.async_set("sensor.test3", 0, attributes=attributes)
        attributes.pop("unit_of_measurement")
        hass.states.async_set("sensor.test4", 0, attributes=attributes)
    attributes = dict(energy_attributes)
    attributes["state_class"] = "invalid"
    hass.states.async_set("sensor.test5", 0, attributes=attributes)
    attributes.pop("state_class")
    hass.states.async_set("sensor.test6", 0, attributes=attributes)


@pytest.mark.parametrize(
    ("device_class", "state_unit", "state_unit2", "unit_class", "mean", "min", "max"),
    [
        (None, None, "cats", "unitless", 13.050847, -10, 30),
        (None, "%", "cats", "unitless", 13.050847, -10, 30),
        ("battery", "%", "cats", "unitless", 13.050847, -10, 30),
        ("battery", None, "cats", "unitless", 13.050847, -10, 30),
        (None, "kW", "Wh", "power", 13.050847, -10, 30),
        # Can't downgrade from ft³ to ft3 or from m³ to m3
        (None, "ft³", "ft3", "volume", 13.050847, -10, 30),
        (None, "ft³/min", "ft³/m", "volume_flow_rate", 13.050847, -10, 30),
        (None, "m³", "m3", "volume", 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics_changing_units_1(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    state_unit2,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where units change from one hour to the next.

    This tests the case where the recorder cannot convert between the units.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        attributes["unit_of_measurement"] = state_unit2
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "cannot be converted to the unit of previously" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    assert (
        f"The unit of sensor.test1 ({state_unit2}) cannot be converted to the unit of "
        f"previously compiled statistics ({state_unit})" in caplog.text
    )
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        (None, "dogs", "dogs", "dogs", None, 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics_changing_units_2(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where units change during an hour.

    This tests the behaviour when the sensor units are note supported by any unit
    converter.
    """
    zero = get_start_time(dt_util.utcnow()) - timedelta(seconds=30 * 5)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        attributes["unit_of_measurement"] = "cats"
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero + timedelta(seconds=30 * 5))
    await async_wait_recording_done(hass)
    assert "The unit of sensor.test1 is changing" in caplog.text
    assert "and matches the unit of already compiled statistics" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": "cats",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "cats",
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {}

    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistics_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        (None, "dogs", "dogs", "dogs", None, 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics_changing_units_3(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistics_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where units change from one hour to the next.

    This tests the behaviour when the sensor units are note supported by any unit
    converter.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
        attributes["unit_of_measurement"] = "cats"
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "does not match the unit of already compiled" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    assert "The unit of sensor.test1 is changing" in caplog.text
    assert (
        f"matches the unit of already compiled statistics ({state_unit})" in caplog.text
    )
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistics_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit_1",
        "state_unit_2",
        "unit_class",
        "mean",
        "min",
        "max",
        "factor",
    ),
    [
        (None, None, "%", "unitless", 13.050847, -10, 30, 100),
        (None, None, "ppm", "unitless", 13.050847, -10, 30, 1000000),
        (None, "g/m³", "mg/m³", "concentration", 13.050847, -10, 30, 1000),
        (None, "mg/m³", "g/m³", "concentration", 13.050847, -10, 30, 0.001),
        (
            "carbon_monoxide",
            "ppm",
            "mg/m³",
            "carbon_monoxide",
            13.050847,
            -10,
            30,
            1.16441,
        ),
        (
            "carbon_monoxide",
            "mg/m³",
            "ppm",
            "carbon_monoxide",
            13.050847,
            -10,
            30,
            1 / 1.16441,
        ),
        (None, "%", None, "unitless", 13.050847, -10, 30, 0.01),
        (None, "W", "kW", "power", 13.050847, -10, 30, 0.001),
        (None, "kW", "W", "power", 13.050847, -10, 30, 1000),
    ],
)
async def test_compile_hourly_statistics_convert_units_1(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit_1,
    state_unit_2,
    unit_class,
    mean,
    min,
    max,
    factor,
) -> None:
    """Test compiling hourly statistics where units change from one hour to the next.

    This tests the case where the recorder can convert between the units.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit_1,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        four, _states = await async_record_states(
            hass,
            freezer,
            zero + timedelta(minutes=5),
            "sensor.test1",
            attributes,
            seq=[0, 1, None],
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "does not match the unit of already compiled" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit_1,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit_1,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    attributes["unit_of_measurement"] = state_unit_2
    with freeze_time(four) as freezer:
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    assert "The unit of sensor.test1 is changing" not in caplog.text
    assert (
        f"matches the unit of already compiled statistics ({state_unit_1})"
        not in caplog.text
    )
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit_2,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit_1,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean * factor),
                "min": pytest.approx(min * factor),
                "max": pytest.approx(max * factor),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=15)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "state_unit2",
        "unit_class",
        "unit_class2",
        "mean",
        "mean2",
        "min",
        "max",
    ),
    [
        (None, "RPM", "rpm", None, None, 13.050847, 13.333333, -10, 30),
        (None, "rpm", "RPM", None, None, 13.050847, 13.333333, -10, 30),
        (None, "ft3", "ft³", None, "volume", 13.050847, 13.333333, -10, 30),
        (
            None,
            "ft³/m",
            "ft³/min",
            None,
            "volume_flow_rate",
            13.050847,
            13.333333,
            -10,
            30,
        ),
        (None, "m3", "m³", None, "volume", 13.050847, 13.333333, -10, 30),
        (None, "\u00b5V", "\u03bcV", None, "voltage", 13.050847, 13.333333, -10, 30),
        (None, "\u00b5Sv/h", "\u03bcSv/h", None, None, 13.050847, 13.333333, -10, 30),
        (
            None,
            "\u00b5S/cm",
            "\u03bcS/cm",
            None,
            "conductivity",
            13.050847,
            13.333333,
            -10,
            30,
        ),
        (None, "\u00b5g/ft³", "\u03bcg/ft³", None, None, 13.050847, 13.333333, -10, 30),
        (
            None,
            "\u00b5g/m³",
            "\u03bcg/m³",
            None,
            "concentration",
            13.050847,
            13.333333,
            -10,
            30,
        ),
        (
            None,
            "\u00b5mol/s⋅m²",
            "\u03bcmol/s⋅m²",
            None,
            None,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        (None, "\u00b5g", "\u03bcg", None, "mass", 13.050847, 13.333333, -10, 30),
        (None, "\u00b5s", "\u03bcs", None, "duration", 13.050847, 13.333333, -10, 30),
        (None, "mVAr", "mvar", None, "reactive_power", 13.050847, 13.333333, -10, 30),
        (None, "VAr", "var", None, "reactive_power", 13.050847, 13.333333, -10, 30),
        (None, "kVAr", "kvar", None, "reactive_power", 13.050847, 13.333333, -10, 30),
        (None, "VArh", "varh", None, "reactive_energy", 13.050847, 13.333333, -10, 30),
        (
            None,
            "kVArh",
            "kvarh",
            None,
            "reactive_energy",
            13.050847,
            13.333333,
            -10,
            30,
        ),
    ],
)
async def test_compile_hourly_statistics_equivalent_units_1(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    state_unit2,
    unit_class,
    unit_class2,
    mean,
    mean2,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where units change from one hour to the next."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        attributes["unit_of_measurement"] = state_unit2
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "cannot be converted to the unit of previously" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit2,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit2,
            "unit_class": unit_class2,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=15)).timestamp(),
                "mean": pytest.approx(mean2),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("device_class", "state_unit", "state_unit2", "unit_class", "mean", "min", "max"),
    [
        (None, "RPM", "rpm", None, 13.333333, -10, 30),
        (None, "rpm", "RPM", None, 13.333333, -10, 30),
        (None, "ft3", "ft³", None, 13.333333, -10, 30),
        (None, "ft³/m", "ft³/min", None, 13.333333, -10, 30),
        (None, "m3", "m³", None, 13.333333, -10, 30),
        (None, "\u00b5V", "\u03bcV", None, 13.333333, -10, 30),
        (SensorDeviceClass.VOLTAGE, "\u00b5V", "\u03bcV", None, 13.333333, -10, 30),
        (None, "\u00b5Sv/h", "\u03bcSv/h", None, 13.333333, -10, 30),
        (None, "\u00b5S/cm", "\u03bcS/cm", None, 13.333333, -10, 30),
        (None, "\u00b5g/ft³", "\u03bcg/ft³", None, 13.333333, -10, 30),
        (None, "\u00b5g/m³", "\u03bcg/m³", None, 13.333333, -10, 30),
        (None, "\u00b5mol/s⋅m²", "\u03bcmol/s⋅m²", None, 13.333333, -10, 30),
        (None, "\u00b5g", "\u03bcg", None, 13.333333, -10, 30),
        (SensorDeviceClass.WEIGHT, "\u00b5g", "\u03bcg", None, 13.333333, -10, 30),
        (None, "\u00b5s", "\u03bcs", None, 13.333333, -10, 30),
        (SensorDeviceClass.DURATION, "\u00b5s", "\u03bcs", None, 13.333333, -10, 30),
        (None, "mVAr", "mvar", None, 13.333333, -10, 30),
        (None, "VAr", "var", None, 13.333333, -10, 30),
        (None, "kVAr", "kvar", None, 13.333333, -10, 30),
        (None, "VArh", "varh", None, 13.333333, -10, 30),
        (None, "kVArh", "kvarh", None, 13.333333, -10, 30),
        (SensorDeviceClass.REACTIVE_POWER, "mVAr", "mvar", None, 13.333333, -10, 30),
        (SensorDeviceClass.REACTIVE_POWER, "VAr", "var", None, 13.333333, -10, 30),
        (SensorDeviceClass.REACTIVE_POWER, "kVAr", "kvar", None, 13.333333, -10, 30),
        (SensorDeviceClass.REACTIVE_ENERGY, "VArh", "varh", None, 13.333333, -10, 30),
        (SensorDeviceClass.REACTIVE_ENERGY, "kVArh", "kvarh", None, 13.333333, -10, 30),
    ],
)
async def test_compile_hourly_statistics_equivalent_units_2(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    state_unit2,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where units change during an hour."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        attributes["unit_of_measurement"] = state_unit2
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero + timedelta(seconds=30 * 10))
    await async_wait_recording_done(hass)
    assert "The unit of sensor.test1 is changing" not in caplog.text
    assert "and matches the unit of already compiled statistics" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(
                    zero + timedelta(seconds=30 * 10)
                ).timestamp(),
                "end": process_timestamp(zero + timedelta(seconds=30 * 20)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }

    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    ("device_class", "state_unit", "state_unit2", "unit_class", "mean", "min", "max"),
    [
        (None, "KB/s", "kB/s", None, 13.333333, -10, 30),
        (SensorDeviceClass.DATA_RATE, "KB/s", "kB/s", None, 13.333333, -10, 30),
        # Can't downgrade from "kB/s" to "KB/s"
        (None, "kB/s", "KB/s", "data_rate", 30, 30, 30),
        (SensorDeviceClass.DATA_RATE, "kB/s", "KB/s", "data_rate", 30, 30, 30),
    ],
)
async def test_compile_hourly_statistics_custom_equivalent_units(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class: SensorDeviceClass | None,
    state_unit: str,
    state_unit2: str,
    unit_class: str | None,
    mean: float,
    min: float,
    max: float,
) -> None:
    """Test stats where units change with custom equivalent units."""
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)

    # Add a recorder platform that provides custom equivalent units
    def _mock_custom_equivalent_units(*args: Any) -> dict:
        return {"sensor.test1": {"KB/s": "kB/s"}}

    recorder_platform = Mock(
        compile_statistics=None,
        async_custom_equivalent_units=Mock(wraps=_mock_custom_equivalent_units),
        list_statistic_ids=None,
        update_statistics_issues=None,
        validate_statistics=None,
    )

    mock_platform(hass, "some_domain.recorder", recorder_platform)
    assert await async_setup_component(hass, "some_domain", {})
    # Wait for the some_domain recorder platform to be added
    await async_recorder_block_till_done(hass)

    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
        attributes["unit_of_measurement"] = state_unit2
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=zero + timedelta(seconds=30 * 10))
    await async_wait_recording_done(hass)
    assert "The unit of sensor.test1 is changing" not in caplog.text
    assert "and matches the unit of already compiled statistics" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": state_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": state_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(
                    zero + timedelta(seconds=30 * 10)
                ).timestamp(),
                "end": process_timestamp(zero + timedelta(seconds=30 * 20)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }

    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "unit_1",
        "unit_2",
        "unit_3",
        "unit_class_1",
        "unit_class_2",
        "factor_2",
        "factor_3",
        "mean1",
        "mean2",
        "min",
        "max",
    ),
    [
        (
            "power",
            "kW",
            "kW",
            "kW",
            "power",
            "power",
            1,
            1,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        (
            "carbon_monoxide",
            "ppm",
            "ppm",
            "ppm",
            "unitless",
            "carbon_monoxide",
            1,
            1,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        # Valid change of unit class from unitless to carbon_monoxide
        (
            "carbon_monoxide",
            "ppm",
            "ppm",
            "mg/m³",
            "unitless",
            "carbon_monoxide",
            1,
            1.164409,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        # Valid change of unit class from unitless to carbon_monoxide
        (
            "carbon_monoxide",
            "ppm",
            "mg/m³",
            "mg/m³",
            "unitless",
            "carbon_monoxide",
            1.164409,
            1.164409,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        # Valid change of unit class from concentration to carbon_monoxide
        (
            "carbon_monoxide",
            "mg/m³",
            "mg/m³",
            "ppm",
            "concentration",
            "carbon_monoxide",
            1,
            1 / 1.164409,
            13.050847,
            13.333333,
            -10,
            30,
        ),
        # Valid change of unit class from concentration to carbon_monoxide
        (
            "carbon_monoxide",
            "mg/m³",
            "ppm",
            "ppm",
            "concentration",
            "carbon_monoxide",
            1 / 1.164409,
            1 / 1.164409,
            13.050847,
            13.333333,
            -10,
            30,
        ),
    ],
)
async def test_compile_hourly_statistics_changing_device_class_1(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    unit_1,
    unit_2,
    unit_3,
    unit_class_1,
    unit_class_2,
    factor_2,
    factor_3,
    mean1,
    mean2,
    min,
    max,
) -> None:
    """Test stats where device class changes between hours.

    In this test, the device class is first None, then set to a
    specific device class.

    Changing device class may influence the unit class.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)

    # Record some states for an initial period, the entity has no device class
    attributes = {
        "state_class": "measurement",
        "unit_of_measurement": unit_1,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "does not match the unit of already compiled" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": unit_1,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": unit_1,
            "unit_class": unit_class_1,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean1),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    # Update device class and record additional states in a different UoM
    attributes["device_class"] = device_class
    attributes["unit_of_measurement"] = unit_2
    seq = [x * factor_2 for x in (-10, 15, 30)]
    with freeze_time(zero) as freezer:
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes, seq
        )
        states["sensor.test1"] += _states["sensor.test1"]
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    # Run statistics again, additional statistics is generated
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": unit_2,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": unit_1,
            "unit_class": unit_class_2,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean1 * factor_2),
                "min": pytest.approx(min * factor_2),
                "max": pytest.approx(max * factor_2),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=15)).timestamp(),
                "mean": pytest.approx(mean2 * factor_2),
                "min": pytest.approx(min * factor_2),
                "max": pytest.approx(max * factor_2),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }

    # Update device class and record additional states in a different UoM
    attributes["unit_of_measurement"] = unit_3
    seq = [x * factor_3 for x in (-10, 15, 30)]
    with freeze_time(zero) as freezer:
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=15), "sensor.test1", attributes, seq
        )
        states["sensor.test1"] += _states["sensor.test1"]
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=20), "sensor.test1", attributes, seq
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    # Run statistics again, additional statistics is generated
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=20))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": unit_3,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": unit_1,
            "unit_class": unit_class_2,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean1 * factor_3),
                "min": pytest.approx(min * factor_3),
                "max": pytest.approx(max * factor_3),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=15)).timestamp(),
                "mean": pytest.approx(mean2 * factor_3),
                "min": pytest.approx(min * factor_3),
                "max": pytest.approx(max * factor_3),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=20)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=25)).timestamp(),
                "mean": pytest.approx(mean2 * factor_3),
                "min": pytest.approx(min * factor_3),
                "max": pytest.approx(max * factor_3),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "display_unit",
        "statistic_unit",
        "unit_class",
        "mean",
        "mean2",
        "min",
        "max",
    ),
    [
        ("power", "kW", "kW", "kW", "power", 13.050847, 13.333333, -10, 30),
    ],
)
async def test_compile_hourly_statistics_changing_device_class_2(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    display_unit,
    statistic_unit,
    unit_class,
    mean,
    mean2,
    min,
    max,
) -> None:
    """Test stats where device class changes between hours.

    In this test, the device class is first set to a specific device
    class, then set to None.
    """
    zero = get_start_time(dt_util.utcnow())
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)

    # Record some states for an initial period, the entity has a device class
    attributes = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(zero) as freezer:
        four, states = await async_record_states(
            hass, freezer, zero, "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)

    do_adhoc_statistics(hass, start=zero)
    await async_wait_recording_done(hass)
    assert "does not match the unit of already compiled" not in caplog.text
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistic_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            }
        ]
    }

    # Remove device class and record additional states
    attributes.pop("device_class")
    with freeze_time(zero) as freezer:
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=5), "sensor.test1", attributes
        )
        states["sensor.test1"] += _states["sensor.test1"]
        four, _states = await async_record_states(
            hass, freezer, zero + timedelta(minutes=10), "sensor.test1", attributes
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, zero, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    # Run statistics again, additional statistics is generated
    do_adhoc_statistics(hass, start=zero + timedelta(minutes=10))
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": display_unit,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": statistic_unit,
            "unit_class": unit_class,
        },
    ]
    stats = statistics_during_period(hass, zero, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(zero).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=5)).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(zero + timedelta(minutes=10)).timestamp(),
                "end": process_timestamp(zero + timedelta(minutes=15)).timestamp(),
                "mean": pytest.approx(mean2),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
        ]
    }
    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.parametrize(
    (
        "device_class",
        "state_unit",
        "unit_class",
        "mean",
        "min",
        "max",
    ),
    [
        (None, None, "unitless", 13.050847, -10, 30),
    ],
)
async def test_compile_hourly_statistics_changing_state_class(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_class,
    state_unit,
    unit_class,
    mean,
    min,
    max,
) -> None:
    """Test compiling hourly statistics where state class changes."""
    period0 = get_start_time(dt_util.utcnow())
    period0_end = period1 = period0 + timedelta(minutes=5)
    period1_end = period0 + timedelta(minutes=10)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes_1 = {
        "device_class": device_class,
        "state_class": "measurement",
        "unit_of_measurement": state_unit,
    }
    attributes_2 = {
        "device_class": device_class,
        "state_class": "total_increasing",
        "unit_of_measurement": state_unit,
    }
    with freeze_time(period0) as freezer:
        four, states = await async_record_states(
            hass, freezer, period0, "sensor.test1", attributes_1
        )
    await async_wait_recording_done(hass)
    do_adhoc_statistics(hass, start=period0)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": None,
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": None,
            "unit_class": unit_class,
        },
    ]
    metadata = get_metadata(hass, statistic_ids={"sensor.test1"})
    assert metadata == {
        "sensor.test1": (
            1,
            {
                "has_mean": True,
                "mean_type": StatisticMeanType.ARITHMETIC,
                "has_sum": False,
                "name": None,
                "source": "recorder",
                "statistic_id": "sensor.test1",
                "unit_class": unit_class,
                "unit_of_measurement": None,
            },
        )
    }

    # Add more states, with changed state class
    with freeze_time(period1) as freezer:
        four, _states = await async_record_states(
            hass, freezer, period1, "sensor.test1", attributes_2
        )
    await async_wait_recording_done(hass)
    states["sensor.test1"] += _states["sensor.test1"]
    hist = history.get_significant_states(
        hass, period0, four, hass.states.async_entity_ids()
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)

    do_adhoc_statistics(hass, start=period1)
    await async_wait_recording_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": None,
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": None,
            "unit_class": unit_class,
        },
    ]
    metadata = get_metadata(hass, statistic_ids={"sensor.test1"})
    assert metadata == {
        "sensor.test1": (
            1,
            {
                "has_mean": False,
                "mean_type": StatisticMeanType.NONE,
                "has_sum": True,
                "name": None,
                "source": "recorder",
                "statistic_id": "sensor.test1",
                "unit_class": unit_class,
                "unit_of_measurement": None,
            },
        )
    }
    stats = statistics_during_period(hass, period0, period="5minute")
    assert stats == {
        "sensor.test1": [
            {
                "start": process_timestamp(period0).timestamp(),
                "end": process_timestamp(period0_end).timestamp(),
                "mean": pytest.approx(mean),
                "min": pytest.approx(min),
                "max": pytest.approx(max),
                "last_reset": None,
                "state": None,
                "sum": None,
            },
            {
                "start": process_timestamp(period1).timestamp(),
                "end": process_timestamp(period1_end).timestamp(),
                "mean": None,
                "min": None,
                "max": None,
                "last_reset": None,
                "state": pytest.approx(30.0),
                "sum": pytest.approx(30.0),
            },
        ]
    }

    assert "Error while processing event StatisticsTask" not in caplog.text


@pytest.mark.timeout(25)
@pytest.mark.parametrize("recorder_config", [{CONF_COMMIT_INTERVAL: 3600 * 4}])
@pytest.mark.freeze_time("2021-09-01 05:00")  # August 31st, 23:00 local time
async def test_compile_statistics_hourly_daily_monthly_summary(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test compiling hourly statistics + monthly and daily summary."""
    dt_util.set_default_time_zone(dt_util.get_time_zone("America/Regina"))

    zero = dt_util.utcnow()
    instance = get_instance(hass)
    await async_setup_component(hass, "sensor", {})
    # Wait for the sensor recorder platform to be added
    await async_recorder_block_till_done(hass)
    attributes = {
        "device_class": None,
        "state_class": "measurement",
        "unit_of_measurement": "%",
    }

    sum_attributes = {
        "device_class": None,
        "state_class": "total",
        "unit_of_measurement": "EUR",
    }

    durations = [50, 200, 45]

    def _weighted_average(seq, i, last_state):
        total = 0
        duration = 0
        if i > 0:
            total += last_state * 5
            duration += 5
        for j, dur in enumerate(durations):
            total += seq[j] * dur
            duration += dur
        return total / duration

    def _weighted_circular_mean(
        values: Iterable[tuple[float, float]],
    ) -> tuple[float, float]:
        sin_sum = 0
        cos_sum = 0
        for x, weight in values:
            sin_sum += math.sin(x * DEG_TO_RAD) * weight
            cos_sum += math.cos(x * DEG_TO_RAD) * weight

        return (
            (RAD_TO_DEG * math.atan2(sin_sum, cos_sum)) % 360,
            math.sqrt(sin_sum**2 + cos_sum**2),
        )

    def _min(seq, last_state):
        if last_state is None:
            return min(seq)
        return min([*seq, last_state])

    def _max(seq, last_state):
        if last_state is None:
            return max(seq)
        return max([*seq, last_state])

    def _sum(seq, last_state, last_sum):
        if last_state is None:
            return seq[-1] - seq[0]
        return last_sum[-1] + seq[-1] - last_state

    # Generate states for two hours
    states = {
        "sensor.test1": [],
        "sensor.test2": [],
        "sensor.test3": [],
        "sensor.test4": [],
        "sensor.test5": [],
    }
    expected_minima = {"sensor.test1": [], "sensor.test2": [], "sensor.test3": []}
    expected_maxima = {"sensor.test1": [], "sensor.test2": [], "sensor.test3": []}
    expected_means = {
        "sensor.test1": [],
        "sensor.test2": [],
        "sensor.test3": [],
        "sensor.test5": [],
    }
    expected_states = {"sensor.test4": []}
    expected_sums = {"sensor.test4": []}
    last_states: dict[str, float | None] = {
        "sensor.test1": None,
        "sensor.test2": None,
        "sensor.test3": None,
        "sensor.test4": None,
        "sensor.test5": None,
    }
    start = zero
    for i in range(24):
        seq = [-10, 15, 30]
        # test1 has same value in every period
        four, _states = await async_record_states(
            hass, freezer, start, "sensor.test1", attributes, seq
        )
        states["sensor.test1"] += _states["sensor.test1"]
        last_state = last_states["sensor.test1"]
        expected_minima["sensor.test1"].append(_min(seq, last_state))
        expected_maxima["sensor.test1"].append(_max(seq, last_state))
        expected_means["sensor.test1"].append(_weighted_average(seq, i, last_state))
        last_states["sensor.test1"] = seq[-1]
        # test2 values change: min/max at the last state
        seq = [-10 * (i + 1), 15 * (i + 1), 30 * (i + 1)]
        four, _states = await async_record_states(
            hass, freezer, start, "sensor.test2", attributes, seq
        )
        states["sensor.test2"] += _states["sensor.test2"]
        last_state = last_states["sensor.test2"]
        expected_minima["sensor.test2"].append(_min(seq, last_state))
        expected_maxima["sensor.test2"].append(_max(seq, last_state))
        expected_means["sensor.test2"].append(_weighted_average(seq, i, last_state))
        last_states["sensor.test2"] = seq[-1]
        # test3 values change: min/max at the first state
        seq = [-10 * (23 - i + 1), 15 * (23 - i + 1), 30 * (23 - i + 1)]
        four, _states = await async_record_states(
            hass, freezer, start, "sensor.test3", attributes, seq
        )
        states["sensor.test3"] += _states["sensor.test3"]
        last_state = last_states["sensor.test3"]
        expected_minima["sensor.test3"].append(_min(seq, last_state))
        expected_maxima["sensor.test3"].append(_max(seq, last_state))
        expected_means["sensor.test3"].append(_weighted_average(seq, i, last_state))
        last_states["sensor.test3"] = seq[-1]
        # test4 values grow
        seq = [i, i + 0.5, i + 0.75]
        start_meter = start
        for j in range(len(seq)):
            _states = await async_record_meter_state(
                hass,
                freezer,
                start_meter,
                "sensor.test4",
                sum_attributes,
                seq[j : j + 1],
            )
            start_meter += timedelta(minutes=1)
            states["sensor.test4"] += _states["sensor.test4"]
        last_state = last_states["sensor.test4"]
        expected_states["sensor.test4"].append(seq[-1])
        expected_sums["sensor.test4"].append(
            _sum(seq, last_state, expected_sums["sensor.test4"])
        )
        last_states["sensor.test4"] = seq[-1]

        # test5 circular mean
        seq = [350 - i, 0 + (i / 2.0), 15 + i]
        four, _states = await async_record_states(
            hass, freezer, start, "sensor.test5", WIND_DIRECTION_ATTRIBUTES, seq
        )
        states["sensor.test5"] += _states["sensor.test5"]
        values = [(seq, durations[j]) for j, seq in enumerate(seq)]
        if (state := last_states["sensor.test5"]) is not None:
            values.append((state, 5))
        expected_means["sensor.test5"].append(_weighted_circular_mean(values))
        last_states["sensor.test5"] = seq[-1]

        start += timedelta(minutes=5)
    await async_wait_recording_done(hass)
    hist = history.get_significant_states(
        hass,
        zero - timedelta.resolution,
        four,
        hass.states.async_entity_ids(),
        significant_changes_only=False,
    )
    assert_dict_of_states_equal_without_context_and_last_changed(states, hist)
    await async_wait_recording_done(hass)

    # Generate 5-minute statistics for two hours
    start = zero
    for _ in range(24):
        do_adhoc_statistics(hass, start=start)
        await async_wait_recording_done(hass)
        start += timedelta(minutes=5)

    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.test1",
            "display_unit_of_measurement": "%",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "%",
            "unit_class": "unitless",
        },
        {
            "statistic_id": "sensor.test2",
            "display_unit_of_measurement": "%",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "%",
            "unit_class": "unitless",
        },
        {
            "statistic_id": "sensor.test3",
            "display_unit_of_measurement": "%",
            "has_mean": True,
            "mean_type": StatisticMeanType.ARITHMETIC,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "%",
            "unit_class": "unitless",
        },
        {
            "statistic_id": "sensor.test4",
            "display_unit_of_measurement": "EUR",
            "has_mean": False,
            "mean_type": StatisticMeanType.NONE,
            "has_sum": True,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": "EUR",
            "unit_class": None,
        },
        {
            "statistic_id": "sensor.test5",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        },
    ]

    # Adjust the inserted statistics
    sum_adjustment = -10
    sum_adjustement_start = zero + timedelta(minutes=65)
    for i in range(13, 24):
        expected_sums["sensor.test4"][i] += sum_adjustment
    instance.async_adjust_statistics(
        "sensor.test4", sum_adjustement_start, sum_adjustment, "EUR"
    )
    await async_wait_recording_done(hass)

    stats = statistics_during_period(hass, zero, period="5minute")
    expected_stats = {
        "sensor.test1": [],
        "sensor.test2": [],
        "sensor.test3": [],
        "sensor.test4": [],
        "sensor.test5": [],
    }
    start = zero
    end = zero + timedelta(minutes=5)
    for i in range(24):
        for entity_id, mean_extractor in (
            ("sensor.test1", lambda x: x),
            ("sensor.test2", lambda x: x),
            ("sensor.test3", lambda x: x),
            ("sensor.test4", lambda x: x),
            ("sensor.test5", lambda x: x[0]),
        ):
            expected_average = (
                mean_extractor(expected_means[entity_id][i])
                if entity_id in expected_means
                else None
            )
            expected_minimum = (
                expected_minima[entity_id][i] if entity_id in expected_minima else None
            )
            expected_maximum = (
                expected_maxima[entity_id][i] if entity_id in expected_maxima else None
            )
            expected_state = (
                expected_states[entity_id][i] if entity_id in expected_states else None
            )
            expected_sum = (
                expected_sums[entity_id][i] if entity_id in expected_sums else None
            )
            expected_stats[entity_id].append(
                {
                    "start": process_timestamp(start).timestamp(),
                    "end": process_timestamp(end).timestamp(),
                    "mean": pytest.approx(expected_average),
                    "min": pytest.approx(expected_minimum),
                    "max": pytest.approx(expected_maximum),
                    "last_reset": None,
                    "state": expected_state,
                    "sum": expected_sum,
                }
            )
        start += timedelta(minutes=5)
        end += timedelta(minutes=5)
    assert stats == expected_stats

    def verify_stats(
        period: Literal["hour", "day", "week", "month"],
        start: datetime,
        next_datetime: Callable[[datetime], datetime],
    ) -> None:
        stats = statistics_during_period(hass, zero, period=period)
        expected_stats = {
            "sensor.test1": [],
            "sensor.test2": [],
            "sensor.test3": [],
            "sensor.test4": [],
            "sensor.test5": [],
        }
        end = next_datetime(start)
        for i in range(2):
            for entity_id, mean_fn in (
                ("sensor.test1", mean),
                ("sensor.test2", mean),
                ("sensor.test3", mean),
                ("sensor.test4", mean),
                ("sensor.test5", lambda x: _weighted_circular_mean(x)[0]),
            ):
                expected_average = (
                    mean_fn(expected_means[entity_id][i * 12 : (i + 1) * 12])
                    if entity_id in expected_means
                    else None
                )
                expected_minimum = (
                    min(expected_minima[entity_id][i * 12 : (i + 1) * 12])
                    if entity_id in expected_minima
                    else None
                )
                expected_maximum = (
                    max(expected_maxima[entity_id][i * 12 : (i + 1) * 12])
                    if entity_id in expected_maxima
                    else None
                )
                expected_state = (
                    expected_states[entity_id][(i + 1) * 12 - 1]
                    if entity_id in expected_states
                    else None
                )
                expected_sum = (
                    expected_sums[entity_id][(i + 1) * 12 - 1]
                    if entity_id in expected_sums
                    else None
                )
                expected_stats[entity_id].append(
                    {
                        "start": process_timestamp(start).timestamp(),
                        "end": process_timestamp(end).timestamp(),
                        "mean": pytest.approx(expected_average),
                        "min": pytest.approx(expected_minimum),
                        "max": pytest.approx(expected_maximum),
                        "last_reset": None,
                        "state": expected_state,
                        "sum": expected_sum,
                    }
                )
            start = next_datetime(start)
            end = next_datetime(end)
        assert stats == expected_stats

    verify_stats("hour", zero, lambda v: v + timedelta(hours=1))

    start = dt_util.parse_datetime("2021-08-31T06:00:00+00:00")
    assert start
    verify_stats("day", start, lambda v: v + timedelta(days=1))

    start = dt_util.parse_datetime("2021-08-01T06:00:00+00:00")
    assert start
    verify_stats("month", start, lambda v: (v + timedelta(days=31)).replace(day=1))

    assert "Error while processing event StatisticsTask" not in caplog.text


async def async_record_states(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    zero: datetime,
    entity_id: str,
    attributes,
    seq=None,
):
    """Record some test states.

    We inject a bunch of state updates for measurement sensors.
    """
    attributes = dict(attributes)
    if seq is None:
        seq = [-10, 15, 30]

    def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=10 * 5)
    three = two + timedelta(seconds=40 * 5)
    four = three + timedelta(seconds=9 * 5)

    states = {entity_id: []}
    freezer.move_to(one)
    states[entity_id].append(set_state(entity_id, str(seq[0]), attributes=attributes))

    freezer.move_to(two)
    states[entity_id].append(set_state(entity_id, str(seq[1]), attributes=attributes))

    freezer.move_to(three)
    states[entity_id].append(set_state(entity_id, str(seq[2]), attributes=attributes))

    return four, states


@pytest.mark.parametrize(
    ("units", "attributes", "unit_class", "unit", "unit2", "supported_unit"),
    [
        (
            US_CUSTOMARY_SYSTEM,
            POWER_SENSOR_ATTRIBUTES,
            "power",
            "W",
            "kW",
            "BTU/h, GW, MW, TW, W, kW, mW",
        ),
        (
            METRIC_SYSTEM,
            POWER_SENSOR_ATTRIBUTES,
            "power",
            "W",
            "kW",
            "BTU/h, GW, MW, TW, W, kW, mW",
        ),
        (
            US_CUSTOMARY_SYSTEM,
            TEMPERATURE_SENSOR_ATTRIBUTES,
            "temperature",
            "°F",
            "K",
            "K, °C, °F",
        ),
        (
            METRIC_SYSTEM,
            TEMPERATURE_SENSOR_ATTRIBUTES,
            "temperature",
            "°C",
            "K",
            "K, °C, °F",
        ),
        (
            US_CUSTOMARY_SYSTEM,
            PRESSURE_SENSOR_ATTRIBUTES,
            "pressure",
            "psi",
            "bar",
            "Pa, bar, cbar, hPa, inHg, inH₂O, kPa, mPa, mbar, mmHg, psi",
        ),
        (
            METRIC_SYSTEM,
            PRESSURE_SENSOR_ATTRIBUTES,
            "pressure",
            "Pa",
            "bar",
            "Pa, bar, cbar, hPa, inHg, inH₂O, kPa, mPa, mbar, mmHg, psi",
        ),
    ],
)
async def test_validate_unit_change_convertible(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit_class,
    unit,
    unit2,
    supported_unit,
) -> None:
    """Test validate_statistics.

    This tests what happens if a sensor is first recorded in a unit which supports unit
    conversion, and the unit is then changed to a unit which can and cannot be
    converted to the original unit.

    The test also asserts that the sensor's device class is ignored.
    """

    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, unit in state matching device class - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes={**attributes, "unit_of_measurement": unit},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # No statistics, unit in state not matching device class - empty response
    hass.states.async_set(
        "sensor.test",
        11,
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, incompatible unit - expect error
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    expected = {
        "sensor.test": [
            {
                "data": {
                    "metadata_unit": unit,
                    "metadata_unit_class": unit_class,
                    "state_unit": "dogs",
                    "state_unit_class": None,
                    "statistic_id": "sensor.test",
                    "supported_unit": supported_unit,
                },
                "type": UNITS_CHANGED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {UNITS_CHANGED_ISSUE})

    # Unavailable state - empty response
    hass.states.async_set(
        "sensor.test",
        "unavailable",
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Unknown state - empty response
    hass.states.async_set(
        "sensor.test",
        "unknown",
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state - empty response
    hass.states.async_set(
        "sensor.test",
        13,
        attributes={**attributes, "unit_of_measurement": unit},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state, statistic runs again - empty response
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state in compatible unit - empty response
    hass.states.async_set(
        "sensor.test",
        13,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state, statistic runs again - empty response
    do_adhoc_statistics(hass, start=now + timedelta(hours=2))
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Remove the state - expect error about missing state
    hass.states.async_remove("sensor.test")
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "no_state",
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("units", "attributes"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES),
    ],
)
async def test_validate_statistics_unit_ignore_device_class(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
) -> None:
    """Test validate_statistics.

    The test asserts that the sensor's device class is ignored.
    """
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, no device class - empty response
    initial_attributes = {"state_class": "measurement", "unit_of_measurement": "dogs"}
    hass.states.async_set(
        "sensor.test", 10, attributes=initial_attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, device class set not matching unit - empty response
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit_class", "unit", "unit2", "supported_unit"),
    [
        (
            US_CUSTOMARY_SYSTEM,
            POWER_SENSOR_ATTRIBUTES,
            "power",
            "W",
            "kW",
            "BTU/h, GW, MW, TW, W, kW, mW",
        ),
        (
            METRIC_SYSTEM,
            POWER_SENSOR_ATTRIBUTES,
            "power",
            "W",
            "kW",
            "BTU/h, GW, MW, TW, W, kW, mW",
        ),
        (
            US_CUSTOMARY_SYSTEM,
            TEMPERATURE_SENSOR_ATTRIBUTES,
            "temperature",
            "°F",
            "K",
            "K, °C, °F",
        ),
        (
            METRIC_SYSTEM,
            TEMPERATURE_SENSOR_ATTRIBUTES,
            "temperature",
            "°C",
            "K",
            "K, °C, °F",
        ),
        (
            US_CUSTOMARY_SYSTEM,
            PRESSURE_SENSOR_ATTRIBUTES,
            "pressure",
            "psi",
            "bar",
            "Pa, bar, cbar, hPa, inHg, inH₂O, kPa, mPa, mbar, mmHg, psi",
        ),
        (
            METRIC_SYSTEM,
            PRESSURE_SENSOR_ATTRIBUTES,
            "pressure",
            "Pa",
            "bar",
            "Pa, bar, cbar, hPa, inHg, inH₂O, kPa, mPa, mbar, mmHg, psi",
        ),
        (
            METRIC_SYSTEM,
            BATTERY_SENSOR_ATTRIBUTES,
            "unitless",
            "%",
            None,
            "%, <None>, ppb, ppm",
        ),
    ],
)
async def test_validate_statistics_unit_change_no_device_class(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit_class,
    unit,
    unit2,
    supported_unit,
) -> None:
    """Test validate_statistics.

    This tests what happens if a sensor is first recorded in a unit which supports unit
    conversion, and the unit is then changed to a unit which can and cannot be
    converted to the original unit.
    """
    attributes = dict(attributes)
    attributes.pop("device_class")

    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, sensor state set - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes={**attributes, "unit_of_measurement": unit},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # No statistics, sensor state set to an incompatible unit - empty response
    hass.states.async_set(
        "sensor.test",
        11,
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, incompatible unit - expect error
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    expected = {
        "sensor.test": [
            {
                "data": {
                    "metadata_unit": unit,
                    "metadata_unit_class": unit_class,
                    "state_unit": "dogs",
                    "state_unit_class": None,
                    "statistic_id": "sensor.test",
                    "supported_unit": supported_unit,
                },
                "type": UNITS_CHANGED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {UNITS_CHANGED_ISSUE})

    # Unavailable state - empty response
    hass.states.async_set(
        "sensor.test",
        "unavailable",
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Unknown state - empty response
    hass.states.async_set(
        "sensor.test",
        "unknown",
        attributes={**attributes, "unit_of_measurement": "dogs"},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state - empty response
    hass.states.async_set(
        "sensor.test",
        13,
        attributes={**attributes, "unit_of_measurement": unit},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state, statistic runs again - empty response
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state in compatible unit - empty response
    hass.states.async_set(
        "sensor.test",
        13,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state, statistic runs again - empty response
    do_adhoc_statistics(hass, start=now + timedelta(hours=2))
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Remove the state - expect error about missing state
    hass.states.async_remove("sensor.test")
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "no_state",
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
    ],
)
async def test_validate_statistics_state_class_removed(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, valid state - empty response
    hass.states.async_set(
        "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, empty response
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # State update with invalid state class, expect error
    _attributes = dict(attributes)
    _attributes.pop("state_class")
    hass.states.async_set(
        "sensor.test", 12, attributes=_attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": STATE_CLASS_REMOVED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {STATE_CLASS_REMOVED_ISSUE})

    # Unavailable state - empty response
    hass.states.async_set(
        "sensor.test", "unavailable", attributes=_attributes, timestamp=now.timestamp()
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Unknown state - empty response
    hass.states.async_set(
        "sensor.test", "unknown", attributes=_attributes, timestamp=now.timestamp()
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
    ],
)
async def test_validate_statistics_state_class_removed_issue_cleaned_up(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, valid state - empty response
    hass.states.async_set(
        "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, empty response
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # State update with invalid state class, expect error
    _attributes = dict(attributes)
    _attributes.pop("state_class")
    hass.states.async_set(
        "sensor.test", 12, attributes=_attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": STATE_CLASS_REMOVED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {STATE_CLASS_REMOVED_ISSUE})

    # Remove the statistics - empty response
    get_instance(hass).async_clear_statistics(["sensor.test"])
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
    ],
)
async def test_validate_statistics_sensor_no_longer_recorded(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, valid state - empty response
    hass.states.async_set(
        "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, empty response
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Sensor no longer recorded, expect error
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "entity_no_longer_recorded",
            }
        ],
    }
    instance = get_instance(hass)
    with patch.object(
        instance,
        "entity_filter",
        return_value=False,
    ):
        await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
    ],
)
async def test_validate_statistics_sensor_not_recorded(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # Sensor not recorded, expect error
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "entity_not_recorded",
            }
        ],
    }
    instance = get_instance(hass)
    with patch.object(
        instance,
        "entity_filter",
        return_value=False,
    ):
        hass.states.async_set(
            "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
        )
        await hass.async_block_till_done()
        await assert_validation_result(hass, client, expected, {})

        # Statistics has run, expect same error
        do_adhoc_statistics(hass, start=now)
        await async_recorder_block_till_done(hass)
        await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("units", "attributes", "unit"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
    ],
)
async def test_validate_statistics_sensor_removed(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    units,
    attributes,
    unit,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, valid state - empty response
    hass.states.async_set(
        "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    await assert_validation_result(hass, client, {}, {})

    # Statistics has run, empty response
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Sensor removed, expect error
    hass.states.async_remove("sensor.test")
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "no_state",
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("attributes", "unit1", "unit2"),
    [
        (BATTERY_SENSOR_ATTRIBUTES, "cats", "dogs"),
        (NONE_SENSOR_ATTRIBUTES, "cats", "dogs"),
    ],
)
async def test_validate_statistics_unit_change_no_conversion(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    attributes,
    unit1,
    unit2,
) -> None:
    """Test validate_statistics."""
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes={**attributes, "unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # No statistics, changed unit - empty response
    hass.states.async_set(
        "sensor.test",
        11,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics, no statistics will be generated because of conflicting units
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(hass, [])

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics one hour later, only the state with unit1 will be considered
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit1}]
    )
    await assert_validation_result(hass, client, {}, {})

    # Change unit - expect error
    hass.states.async_set(
        "sensor.test",
        13,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    expected = {
        "sensor.test": [
            {
                "data": {
                    "metadata_unit": unit1,
                    "metadata_unit_class": None,
                    "state_unit": unit2,
                    "state_unit_class": None,
                    "statistic_id": "sensor.test",
                    "supported_unit": unit1,
                },
                "type": UNITS_CHANGED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {UNITS_CHANGED_ISSUE})

    # Unavailable state - empty response
    hass.states.async_set(
        "sensor.test",
        "unavailable",
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Unknown state - empty response
    hass.states.async_set(
        "sensor.test",
        "unknown",
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Original unit - empty response
    hass.states.async_set(
        "sensor.test",
        14,
        attributes={**attributes, "unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Valid state, statistic runs again - empty response
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=2))
    await async_recorder_block_till_done(hass)
    await assert_validation_result(hass, client, {}, {})

    # Remove the state - expect error
    hass.states.async_remove("sensor.test")
    expected = {
        "sensor.test": [
            {
                "data": {"statistic_id": "sensor.test"},
                "type": "no_state",
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {})


@pytest.mark.parametrize(
    ("attributes", "unit1", "unit2"),
    [
        (NONE_SENSOR_ATTRIBUTES, "m3", "m³"),
        (NONE_SENSOR_ATTRIBUTES, "rpm", "RPM"),
        (NONE_SENSOR_ATTRIBUTES, "RPM", "rpm"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5V", "\u03bcV"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5Sv/h", "\u03bcSv/h"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5S/cm", "\u03bcS/cm"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5g/ft³", "\u03bcg/ft³"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5g/m³", "\u03bcg/m³"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5mol/s⋅m²", "\u03bcmol/s⋅m²"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5g", "\u03bcg"),
        (NONE_SENSOR_ATTRIBUTES, "\u00b5s", "\u03bcs"),
        (NONE_SENSOR_ATTRIBUTES, "mVAr", "mvar"),
        (NONE_SENSOR_ATTRIBUTES, "VAr", "var"),
        (NONE_SENSOR_ATTRIBUTES, "kVAr", "kvar"),
        (NONE_SENSOR_ATTRIBUTES, "VArh", "varh"),
        (NONE_SENSOR_ATTRIBUTES, "kVArh", "kvarh"),
    ],
)
async def test_validate_statistics_unit_change_equivalent_units(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    attributes,
    unit1,
    unit2,
) -> None:
    """Test validate_statistics.

    This tests no validation issue is created when a sensor's unit changes to an
    equivalent unit.
    """
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes={**attributes, "unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit1}]
    )

    # Units changed to an equivalent unit - empty response
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp() + 1,
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics one hour later, metadata will be updated
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit2}]
    )
    await assert_validation_result(hass, client, {}, {})


@pytest.mark.parametrize(
    ("attributes", "unit_class", "unit1", "unit2", "supported_unit"),
    [
        (
            NONE_SENSOR_ATTRIBUTES,
            "volume",
            "m³",
            "m3",
            "CCF, L, MCF, fl. oz., ft³, gal, mL, m³",
        ),
        (
            NONE_SENSOR_ATTRIBUTES,
            "voltage",
            "\u03bcV",
            "\u00b5V",
            "MV, V, kV, mV, \u03bcV",
        ),
        (
            NONE_SENSOR_ATTRIBUTES,
            "conductivity",
            "\u03bcS/cm",
            "\u00b5S/cm",
            "S/cm, mS/cm, \u03bcS/cm",
        ),
        (
            NONE_SENSOR_ATTRIBUTES,
            "mass",
            "\u03bcg",
            "\u00b5g",
            "g, kg, lb, mg, oz, st, \u03bcg",
        ),
        (
            NONE_SENSOR_ATTRIBUTES,
            "duration",
            "\u03bcs",
            "\u00b5s",
            "d, h, min, ms, s, w, \u03bcs",
        ),
        (NONE_SENSOR_ATTRIBUTES, "reactive_power", "mvar", "mVAr", "kvar, mvar, var"),
        (NONE_SENSOR_ATTRIBUTES, "reactive_power", "var", "VAr", "kvar, mvar, var"),
        (NONE_SENSOR_ATTRIBUTES, "reactive_power", "kvar", "kVAr", "kvar, mvar, var"),
        (NONE_SENSOR_ATTRIBUTES, "reactive_energy", "varh", "VArh", "kvarh, varh"),
        (NONE_SENSOR_ATTRIBUTES, "reactive_energy", "kvarh", "kVArh", "kvarh, varh"),
    ],
)
async def test_validate_statistics_unit_change_equivalent_units_2(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    attributes,
    unit_class,
    unit1,
    unit2,
    supported_unit,
) -> None:
    """Test validate_statistics.

    This tests a validation issue is created when a sensor's unit changes to an
    equivalent unit which is not known to the unit converters.
    """
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes={**attributes, "unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit1}]
    )

    # Units changed to an equivalent unit which is not known by the unit converters
    hass.states.async_set(
        "sensor.test",
        12,
        attributes={**attributes, "unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    expected = {
        "sensor.test": [
            {
                "data": {
                    "metadata_unit": unit1,
                    "metadata_unit_class": unit_class,
                    "state_unit": unit2,
                    "state_unit_class": None,
                    "statistic_id": "sensor.test",
                    "supported_unit": supported_unit,
                },
                "type": UNITS_CHANGED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {UNITS_CHANGED_ISSUE})

    # Run statistics one hour later, metadata will not be updated
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit1}]
    )
    await assert_validation_result(hass, client, expected, {UNITS_CHANGED_ISSUE})


@pytest.mark.parametrize(
    ("attributes", "unit1", "unit2"),
    [
        (NONE_SENSOR_ATTRIBUTES, "KB/s", "kB/s"),
        (NONE_SENSOR_ATTRIBUTES, "KW", "kW"),
    ],
)
async def test_validate_statistics_unit_change_custom_equivalent_units(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    attributes: dict,
    unit1: str,
    unit2: str,
) -> None:
    """Test validate_statistics.

    This tests no validation issue is created when a sensor's unit changes to a custom
    equivalent unit provided by an integration's recorder platform.
    """
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)

    # Add a recorder platform that provides custom equivalent units
    def _mock_custom_equivalent_units(*args: Any) -> dict:
        return {"sensor.test": {unit1: unit2}}

    recorder_platform = Mock(
        compile_statistics=None,
        async_custom_equivalent_units=Mock(wraps=_mock_custom_equivalent_units),
        list_statistic_ids=None,
        update_statistics_issues=None,
        validate_statistics=None,
    )

    mock_platform(hass, "some_domain.recorder", recorder_platform)
    assert await async_setup_component(hass, "some_domain", {})
    # Wait for the some_domain recorder platform to be added
    await async_recorder_block_till_done(hass)

    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.test",
        10,
        attributes=attributes | {"unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit1}]
    )

    # Units changed to a custom equivalent unit - empty response
    hass.states.async_set(
        "sensor.test",
        12,
        attributes=attributes | {"unit_of_measurement": unit2},
        timestamp=now.timestamp() + 1,
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics one hour later, metadata will be updated
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    await assert_statistic_ids(
        hass, [{"statistic_id": "sensor.test", "unit_of_measurement": unit2}]
    )
    await assert_validation_result(hass, client, {}, {})


async def test_validate_statistics_other_domain(
    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
    """Test sensor does not raise issues for statistics for other domains."""
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # Create statistics for another domain
    metadata: StatisticMetaData = {
        "has_sum": True,
        "mean_type": StatisticMeanType.ARITHMETIC,
        "name": None,
        "source": RECORDER_DOMAIN,
        "statistic_id": "number.test",
        "unit_class": None,
        "unit_of_measurement": None,
    }
    statistics: StatisticData = {
        "last_reset": None,
        "max": None,
        "mean": None,
        "min": None,
        "start": datetime(2020, 10, 6, tzinfo=dt_util.UTC),
        "state": None,
        "sum": None,
    }
    async_import_statistics(hass, metadata, (statistics,))
    await async_recorder_block_till_done(hass)

    # We should not get complains about the missing number entity
    await assert_validation_result(hass, client, {}, {})


async def _one_hour_stats(hass: HomeAssistant, start: datetime) -> datetime:
    """Generate 5-minute statistics for one hour."""
    for _ in range(12):
        do_adhoc_statistics(hass, start=start)
        await async_wait_recording_done(hass)
        start += timedelta(minutes=5)
    return start


@pytest.mark.parametrize(
    ("units", "attributes"),
    [
        (US_CUSTOMARY_SYSTEM, POWER_SENSOR_ATTRIBUTES),
    ],
)
async def test_update_statistics_issues(
    hass: HomeAssistant,
    units,
    attributes,
) -> None:
    """Test update_statistics_issues."""
    now = get_start_time(dt_util.utcnow())

    hass.config.units = units
    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)

    # No statistics, no state - no issues
    now = await _one_hour_stats(hass, now)
    assert_issues(hass, {})

    # Statistics, valid state - no issues
    hass.states.async_set(
        "sensor.test", 10, attributes=attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    now = await _one_hour_stats(hass, now)
    assert_issues(hass, {})

    # State update with invalid state class, statistics did not run again
    _attributes = dict(attributes)
    _attributes.pop("state_class")
    hass.states.async_set(
        "sensor.test", 12, attributes=_attributes, timestamp=now.timestamp()
    )
    await hass.async_block_till_done()
    assert_issues(hass, {})

    # Let statistics run for one hour, expect issue
    now = await _one_hour_stats(hass, now)
    expected = {
        "state_class_removed_sensor.test": {
            "issue_type": STATE_CLASS_REMOVED_ISSUE,
            "statistic_id": "sensor.test",
        }
    }
    assert_issues(hass, expected)


@pytest.mark.parametrize(
    ("attributes", "unit1", "unit2"),
    [
        (NONE_SENSOR_ATTRIBUTES, "KB/s", "kB/s"),
        (NONE_SENSOR_ATTRIBUTES, "KW", "kW"),
    ],
)
async def test_update_statistics_issues_with_custom_equivalent_units(
    hass: HomeAssistant,
    attributes: dict,
    unit1: str,
    unit2: str,
) -> None:
    """Test update_statistics_issues when custom equivalent units are provided."""
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)

    # Add a recorder platform that provides custom equivalent units
    def _mock_custom_equivalent_units(*args: Any) -> dict:
        return {"sensor.test": {unit1: unit2}}

    recorder_platform = Mock(
        compile_statistics=None,
        async_custom_equivalent_units=Mock(wraps=_mock_custom_equivalent_units),
        list_statistic_ids=None,
        update_statistics_issues=None,
        validate_statistics=None,
    )

    mock_platform(hass, "some_domain.recorder", recorder_platform)
    assert await async_setup_component(hass, "some_domain", {})
    # Wait for the some_domain recorder platform to be added
    await async_recorder_block_till_done(hass)

    # No statistics, no state - no issues
    now = await _one_hour_stats(hass, now)
    assert_issues(hass, {})

    # Statistics, unit not part of unit system but still a valid state - no issues
    hass.states.async_set(
        "sensor.test",
        10,
        attributes=attributes | {"unit_of_measurement": unit1},
        timestamp=now.timestamp(),
    )
    await hass.async_block_till_done()
    now = await _one_hour_stats(hass, now)
    assert_issues(hass, {})

    # State update with equivalent supported unit, statistics did not run again
    hass.states.async_set(
        "sensor.test",
        12,
        attributes=attributes | {"unit_of_measurement": unit2},
        timestamp=now.timestamp(),
    )
    await hass.async_block_till_done()
    assert_issues(hass, {})

    # Let statistics run for one hour, expect no issues
    now = await _one_hour_stats(hass, now)
    assert_issues(hass, {})


async def async_record_meter_states(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    zero: datetime,
    entity_id: str,
    _attributes,
    seq,
):
    """Record some test states.

    We inject a bunch of state updates for meter sensors.
    """

    def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=15 * 5)  # 00:01:15
    two = one + timedelta(seconds=30 * 5)  # 00:03:45
    three = two + timedelta(seconds=15 * 5)  # 00:05:00
    four = three + timedelta(seconds=15 * 5)  # 00:06:15
    five = four + timedelta(seconds=30 * 5)  # 00:08:45
    six = five + timedelta(seconds=15 * 5)  # 00:10:00
    seven = six + timedelta(seconds=15 * 5)  # 00:11:45
    eight = seven + timedelta(seconds=30 * 5)  # 00:13:45

    attributes = dict(_attributes)
    if "last_reset" in _attributes:
        attributes["last_reset"] = zero.isoformat()

    states = {entity_id: []}
    freezer.move_to(zero)

    states[entity_id].append(set_state(entity_id, seq[0], attributes=attributes))

    freezer.move_to(one)
    states[entity_id].append(set_state(entity_id, seq[1], attributes=attributes))

    freezer.move_to(two)
    states[entity_id].append(set_state(entity_id, seq[2], attributes=attributes))

    freezer.move_to(three)
    states[entity_id].append(set_state(entity_id, seq[3], attributes=attributes))

    attributes = dict(_attributes)
    if "last_reset" in _attributes:
        attributes["last_reset"] = four.isoformat()

    freezer.move_to(four)
    states[entity_id].append(set_state(entity_id, seq[4], attributes=attributes))

    freezer.move_to(five)
    states[entity_id].append(set_state(entity_id, seq[5], attributes=attributes))

    freezer.move_to(six)
    states[entity_id].append(set_state(entity_id, seq[6], attributes=attributes))

    freezer.move_to(seven)
    states[entity_id].append(set_state(entity_id, seq[7], attributes=attributes))

    freezer.move_to(eight)
    states[entity_id].append(set_state(entity_id, seq[8], attributes=attributes))

    return four, eight, states


async def async_record_meter_state(
    hass: HomeAssistant,
    freezer: FrozenDateTimeFactory,
    zero: datetime,
    entity_id: str,
    attributes,
    seq,
):
    """Record test state.

    We inject a state update for meter sensor.
    """

    def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        return hass.states.get(entity_id)

    states = {entity_id: []}
    freezer.move_to(zero)
    states[entity_id].append(set_state(entity_id, seq[0], attributes=attributes))

    return states


async def async_record_states_partially_unavailable(
    hass: HomeAssistant, zero: datetime, entity_id: str, attributes: dict[str, Any]
) -> tuple[datetime, dict[str, list[State]]]:
    """Record some test states.

    We inject a bunch of state updates temperature sensors.
    """

    def set_state(entity_id, state, **kwargs):
        """Set the state."""
        hass.states.async_set(entity_id, state, **kwargs)
        return hass.states.get(entity_id)

    one = zero + timedelta(seconds=1 * 5)
    two = one + timedelta(seconds=15 * 5)
    three = two + timedelta(seconds=30 * 5)
    four = three + timedelta(seconds=14 * 5)

    states = {entity_id: []}
    with freeze_time(one) as freezer:
        states[entity_id].append(set_state(entity_id, "10", attributes=attributes))

        freezer.move_to(two)
        states[entity_id].append(set_state(entity_id, "25", attributes=attributes))

        freezer.move_to(three)
        states[entity_id].append(
            set_state(entity_id, STATE_UNAVAILABLE, attributes=attributes)
        )

    return four, states


@pytest.mark.usefixtures("enable_custom_integrations")
async def test_exclude_attributes(hass: HomeAssistant) -> None:
    """Test sensor attributes to be excluded."""
    entity0 = MockSensor(
        has_entity_name=True,
        unique_id="test",
        name="Test",
        native_value="option1",
        device_class=SensorDeviceClass.ENUM,
        options=["option1", "option2"],
    )
    setup_test_component_platform(hass, DOMAIN, [entity0])
    assert await async_setup_component(hass, "sensor", {"sensor": {"platform": "test"}})
    await hass.async_block_till_done()
    await async_wait_recording_done(hass)

    def _fetch_states() -> list[State]:
        with session_scope(hass=hass, read_only=True) as session:
            native_states = []
            for db_state, db_state_attributes, db_states_meta in (
                session.query(States, StateAttributes, StatesMeta)
                .outerjoin(
                    StateAttributes,
                    States.attributes_id == StateAttributes.attributes_id,
                )
                .outerjoin(StatesMeta, States.metadata_id == StatesMeta.metadata_id)
            ):
                db_state.entity_id = db_states_meta.entity_id
                state = db_state_to_native(db_state)
                state.attributes = db_state_attributes_to_native(db_state_attributes)
                native_states.append(state)
            return native_states

    states: list[State] = await hass.async_add_executor_job(_fetch_states)
    assert len(states) == 1
    assert ATTR_OPTIONS not in states[0].attributes
    assert ATTR_FRIENDLY_NAME in states[0].attributes


@pytest.mark.parametrize("ignore_translations_for_mock_domains", ["test"])
@pytest.mark.parametrize(
    "ignore_missing_translations",
    [
        [
            "component.sensor.issues..title",
            "component.sensor.issues..description",
        ]
    ],
)
async def test_clean_up_repairs(
    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
    """Test cleaning up repairs."""
    await async_setup_component(hass, "sensor", {})
    issue_registry = ir.async_get(hass)
    client = await hass_ws_client()

    # Create some issues
    def create_issue(domain: str, issue_id: str, data: dict | None) -> None:
        ir.async_create_issue(
            hass,
            domain,
            issue_id,
            data=data,
            is_fixable=False,
            severity=ir.IssueSeverity.WARNING,
            translation_key="",
        )

    create_issue("test", "test_issue", None)
    create_issue(DOMAIN, "test_issue_1", None)
    create_issue(DOMAIN, "test_issue_2", {"issue_type": "another_issue"})
    create_issue(DOMAIN, "test_issue_3", {"issue_type": STATE_CLASS_REMOVED_ISSUE})
    create_issue(DOMAIN, "test_issue_4", {"issue_type": UNITS_CHANGED_ISSUE})
    create_issue(DOMAIN, "test_issue_5", {"issue_type": MEAN_TYPE_CHANGED_ISSUE})

    # Check the issues
    assert set(issue_registry.issues) == {
        ("test", "test_issue"),
        ("sensor", "test_issue_1"),
        ("sensor", "test_issue_2"),
        ("sensor", "test_issue_3"),
        ("sensor", "test_issue_4"),
        ("sensor", "test_issue_5"),
    }

    # Request update of issues
    await client.send_json_auto_id({"type": "recorder/update_statistics_issues"})
    response = await client.receive_json()
    assert response["success"]

    # Check the issues
    assert set(issue_registry.issues) == {
        ("test", "test_issue"),
        ("sensor", "test_issue_1"),
        ("sensor", "test_issue_2"),
    }


async def test_validate_statistics_mean_type_changed(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test validate_statistics.

    This tests a validation issue is created when a the mean type is changed.
    """
    now = get_start_time(dt_util.utcnow())

    await async_setup_component(hass, "sensor", {})
    await async_recorder_block_till_done(hass)
    client = await hass_ws_client()

    # No statistics, no state - empty response
    await assert_validation_result(hass, client, {}, {})

    # No statistics, original unit - empty response
    hass.states.async_set(
        "sensor.wind_direction",
        10,
        attributes=WIND_DIRECTION_ATTRIBUTES,
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.wind_direction",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        }
    ]

    expected_log_entry = (
        "homeassistant.components.sensor.recorder",
        logging.WARNING,
        (
            "The statistics mean algorithm for sensor.wind_direction have changed from"
            " CIRCULAR to ARITHMETIC. Generation of long term statistics will be "
            "suppressed unless it changes back or go to "
            "https://my.home-assistant.io/redirect/developer_statistics "
            "to delete the old statistics"
        ),
    )
    # Valid stats, no log entry
    assert expected_log_entry not in caplog.record_tuples

    # State class changed
    hass.states.async_set(
        "sensor.wind_direction",
        5,
        attributes={
            **WIND_DIRECTION_ATTRIBUTES,
            "state_class": SensorStateClass.MEASUREMENT,
        },
        timestamp=now.timestamp(),
    )
    expected = {
        "sensor.wind_direction": [
            {
                "data": {
                    "statistic_id": "sensor.wind_direction",
                    "metadata_mean_type": StatisticMeanType.CIRCULAR,
                    "state_mean_type": StatisticMeanType.ARITHMETIC,
                },
                "type": MEAN_TYPE_CHANGED_ISSUE,
            }
        ],
    }
    await assert_validation_result(hass, client, expected, {MEAN_TYPE_CHANGED_ISSUE})

    # Run statistics one hour later, metadata will not be updated
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now + timedelta(hours=1))
    await async_recorder_block_till_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.wind_direction",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        }
    ]
    await assert_validation_result(hass, client, expected, {MEAN_TYPE_CHANGED_ISSUE})
    assert expected_log_entry in caplog.record_tuples

    # State class changed back
    hass.states.async_set(
        "sensor.wind_direction",
        350,
        attributes=WIND_DIRECTION_ATTRIBUTES,
        timestamp=now.timestamp(),
    )
    await assert_validation_result(hass, client, {}, {})

    # Run statistics
    await async_recorder_block_till_done(hass)
    do_adhoc_statistics(hass, start=now)
    await async_recorder_block_till_done(hass)
    statistic_ids = await async_list_statistic_ids(hass)
    assert statistic_ids == [
        {
            "statistic_id": "sensor.wind_direction",
            "display_unit_of_measurement": DEGREE,
            "has_mean": False,
            "mean_type": StatisticMeanType.CIRCULAR,
            "has_sum": False,
            "name": None,
            "source": "recorder",
            "statistics_unit_of_measurement": DEGREE,
            "unit_class": None,
        }
    ]

    # Issue should be resolved
    await assert_validation_result(hass, client, {}, {})
