diff --git a/homeassistant/components/history/websocket_api.py b/homeassistant/components/history/websocket_api.py index 03bb7efc561..465416607a2 100644 --- a/homeassistant/components/history/websocket_api.py +++ b/homeassistant/components/history/websocket_api.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import Callable, Iterable, MutableMapping +from collections.abc import Callable, Iterable from dataclasses import dataclass from datetime import datetime as dt, timedelta import logging @@ -173,7 +173,7 @@ async def ws_get_history_during_period( def _generate_stream_message( - states: MutableMapping[str, list[dict[str, Any]]], + states: dict[str, list[dict[str, Any]]], start_day: dt, end_day: dt, ) -> dict[str, Any]: @@ -201,7 +201,7 @@ def _generate_websocket_response( msg_id: int, start_time: dt, end_time: dt, - states: MutableMapping[str, list[dict[str, Any]]], + states: dict[str, list[dict[str, Any]]], ) -> bytes: """Generate a websocket response.""" return json_bytes( @@ -225,7 +225,7 @@ def _generate_historical_response( ) -> tuple[float, dt | None, bytes | None]: """Generate a historical response.""" states = cast( - MutableMapping[str, list[dict[str, Any]]], + dict[str, list[dict[str, Any]]], history.get_significant_states( hass, start_time, @@ -311,7 +311,7 @@ def _history_compressed_state(state: State, no_attributes: bool) -> dict[str, An def _events_to_compressed_states( events: Iterable[Event], no_attributes: bool -) -> MutableMapping[str, list[dict[str, Any]]]: +) -> dict[str, list[dict[str, Any]]]: """Convert events to a compressed states.""" states_by_entity_ids: dict[str, list[dict[str, Any]]] = {} for event in events: diff --git a/homeassistant/components/recorder/history/__init__.py b/homeassistant/components/recorder/history/__init__.py index 05452fdac47..de7002eb6a4 100644 --- a/homeassistant/components/recorder/history/__init__.py +++ b/homeassistant/components/recorder/history/__init__.py @@ -2,7 +2,6 @@ from __future__ import annotations -from collections.abc import MutableMapping from datetime import datetime from typing import Any @@ -43,7 +42,7 @@ def get_full_significant_states_with_session( include_start_time_state: bool = True, significant_changes_only: bool = True, no_attributes: bool = False, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return a dict of significant states during a time period.""" if not recorder.get_instance(hass).states_meta_manager.active: from .legacy import ( # pylint: disable=import-outside-toplevel @@ -68,7 +67,7 @@ def get_full_significant_states_with_session( def get_last_state_changes( hass: HomeAssistant, number_of_states: int, entity_id: str -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return the last number_of_states.""" if not recorder.get_instance(hass).states_meta_manager.active: from .legacy import ( # pylint: disable=import-outside-toplevel @@ -92,7 +91,7 @@ def get_significant_states( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Return a dict of significant states during a time period.""" if not recorder.get_instance(hass).states_meta_manager.active: from .legacy import ( # pylint: disable=import-outside-toplevel @@ -128,7 +127,7 @@ def get_significant_states_with_session( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Return a dict of significant states during a time period.""" if not recorder.get_instance(hass).states_meta_manager.active: from .legacy import ( # pylint: disable=import-outside-toplevel @@ -162,7 +161,7 @@ def state_changes_during_period( descending: bool = False, limit: int | None = None, include_start_time_state: bool = True, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return a list of states that changed during a time period.""" if not recorder.get_instance(hass).states_meta_manager.active: from .legacy import ( # pylint: disable=import-outside-toplevel diff --git a/homeassistant/components/recorder/history/legacy.py b/homeassistant/components/recorder/history/legacy.py index ad9505e1af2..8ee3cd30316 100644 --- a/homeassistant/components/recorder/history/legacy.py +++ b/homeassistant/components/recorder/history/legacy.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections import defaultdict -from collections.abc import Callable, Iterable, Iterator, MutableMapping +from collections.abc import Callable, Iterable, Iterator from datetime import datetime from itertools import groupby from operator import attrgetter @@ -209,7 +209,7 @@ def get_significant_states( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Wrap get_significant_states_with_session with an sql session.""" with session_scope(hass=hass, read_only=True) as session: return get_significant_states_with_session( @@ -317,7 +317,7 @@ def get_significant_states_with_session( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Return states changes during UTC period start_time - end_time. entity_ids is an optional iterable of entities to include in the results. @@ -365,14 +365,14 @@ def get_full_significant_states_with_session( include_start_time_state: bool = True, significant_changes_only: bool = True, no_attributes: bool = False, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Variant of get_significant_states_with_session. Difference with get_significant_states_with_session is that it does not return minimal responses. """ return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], get_significant_states_with_session( hass=hass, session=session, @@ -454,7 +454,7 @@ def state_changes_during_period( descending: bool = False, limit: int | None = None, include_start_time_state: bool = True, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return states changes during UTC period start_time - end_time.""" if not entity_id: raise ValueError("entity_id must be provided") @@ -471,7 +471,7 @@ def state_changes_during_period( ) states = execute_stmt_lambda_element(session, stmt, None, end_time) return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], _sorted_states_to_dict( hass, session, @@ -522,7 +522,7 @@ def _get_last_state_changes_stmt( def get_last_state_changes( hass: HomeAssistant, number_of_states: int, entity_id: str -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return the last number_of_states.""" entity_id_lower = entity_id.lower() entity_ids = [entity_id_lower] @@ -533,7 +533,7 @@ def get_last_state_changes( ) states = list(execute_stmt_lambda_element(session, stmt)) return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], _sorted_states_to_dict( hass, session, @@ -693,7 +693,7 @@ def _sorted_states_to_dict( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Convert SQL results into JSON friendly data structure. This takes our state list and turns it into a JSON friendly data diff --git a/homeassistant/components/recorder/history/modern.py b/homeassistant/components/recorder/history/modern.py index 675bb1b8cf8..96347a1f57b 100644 --- a/homeassistant/components/recorder/history/modern.py +++ b/homeassistant/components/recorder/history/modern.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable, Iterable, Iterator, MutableMapping +from collections.abc import Callable, Iterable, Iterator from datetime import datetime from itertools import groupby from operator import itemgetter @@ -117,7 +117,7 @@ def get_significant_states( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Wrap get_significant_states_with_session with an sql session.""" with session_scope(hass=hass, read_only=True) as session: return get_significant_states_with_session( @@ -213,7 +213,7 @@ def get_significant_states_with_session( minimal_response: bool = False, no_attributes: bool = False, compressed_state_format: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Return states changes during UTC period start_time - end_time. entity_ids is an optional iterable of entities to include in the results. @@ -296,14 +296,14 @@ def get_full_significant_states_with_session( include_start_time_state: bool = True, significant_changes_only: bool = True, no_attributes: bool = False, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Variant of get_significant_states_with_session. Difference with get_significant_states_with_session is that it does not return minimal responses. """ return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], get_significant_states_with_session( hass=hass, session=session, @@ -390,7 +390,7 @@ def state_changes_during_period( descending: bool = False, limit: int | None = None, include_start_time_state: bool = True, -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return states changes during UTC period start_time - end_time.""" has_last_reported = ( recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION @@ -438,7 +438,7 @@ def state_changes_during_period( ], ) return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], _sorted_states_to_dict( execute_stmt_lambda_element( session, stmt, None, end_time, orm_rows=False @@ -504,7 +504,7 @@ def _get_last_state_changes_multiple_stmt( def get_last_state_changes( hass: HomeAssistant, number_of_states: int, entity_id: str -) -> MutableMapping[str, list[State]]: +) -> dict[str, list[State]]: """Return the last number_of_states.""" has_last_reported = ( recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION @@ -539,7 +539,7 @@ def get_last_state_changes( ) states = list(execute_stmt_lambda_element(session, stmt, orm_rows=False)) return cast( - MutableMapping[str, list[State]], + dict[str, list[State]], _sorted_states_to_dict( reversed(states), None, @@ -680,7 +680,7 @@ def _sorted_states_to_dict( compressed_state_format: bool = False, descending: bool = False, no_attributes: bool = False, -) -> MutableMapping[str, list[State | dict[str, Any]]]: +) -> dict[str, list[State | dict[str, Any]]]: """Convert SQL results into JSON friendly data structure. This takes our state list and turns it into a JSON friendly data diff --git a/homeassistant/components/sensor/recorder.py b/homeassistant/components/sensor/recorder.py index 97ad49fb937..1db811599ad 100644 --- a/homeassistant/components/sensor/recorder.py +++ b/homeassistant/components/sensor/recorder.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections import defaultdict -from collections.abc import Callable, Iterable, MutableMapping +from collections.abc import Callable, Iterable import datetime import itertools import logging @@ -402,7 +402,7 @@ def compile_statistics( # noqa: C901 entities_full_history = [ i.entity_id for i in sensor_states if "sum" in wanted_statistics[i.entity_id] ] - history_list: MutableMapping[str, list[State]] = {} + history_list: dict[str, list[State]] = {} if entities_full_history: history_list = history.get_full_significant_states_with_session( hass,