"""Models for SQLAlchemy.""" from __future__ import annotations from datetime import datetime, timedelta import json import logging from typing import TypedDict, overload from sqlalchemy import ( Boolean, Column, DateTime, Float, ForeignKey, Identity, Index, Integer, String, Text, distinct, ) from sqlalchemy.dialects import mysql, oracle, postgresql from sqlalchemy.ext.declarative import declared_attr from sqlalchemy.orm import declarative_base, relationship from sqlalchemy.orm.session import Session from homeassistant.const import ( MAX_LENGTH_EVENT_CONTEXT_ID, MAX_LENGTH_EVENT_EVENT_TYPE, MAX_LENGTH_EVENT_ORIGIN, MAX_LENGTH_STATE_DOMAIN, MAX_LENGTH_STATE_ENTITY_ID, MAX_LENGTH_STATE_STATE, ) from homeassistant.core import Context, Event, EventOrigin, State, split_entity_id from homeassistant.helpers.json import JSONEncoder import homeassistant.util.dt as dt_util # SQLAlchemy Schema # pylint: disable=invalid-name Base = declarative_base() SCHEMA_VERSION = 24 _LOGGER = logging.getLogger(__name__) DB_TIMEZONE = "+00:00" TABLE_EVENTS = "events" TABLE_STATES = "states" TABLE_RECORDER_RUNS = "recorder_runs" TABLE_SCHEMA_CHANGES = "schema_changes" TABLE_STATISTICS = "statistics" TABLE_STATISTICS_META = "statistics_meta" TABLE_STATISTICS_RUNS = "statistics_runs" TABLE_STATISTICS_SHORT_TERM = "statistics_short_term" ALL_TABLES = [ TABLE_STATES, TABLE_EVENTS, TABLE_RECORDER_RUNS, TABLE_SCHEMA_CHANGES, TABLE_STATISTICS, TABLE_STATISTICS_META, TABLE_STATISTICS_RUNS, TABLE_STATISTICS_SHORT_TERM, ] DATETIME_TYPE = DateTime(timezone=True).with_variant( mysql.DATETIME(timezone=True, fsp=6), "mysql" ) DOUBLE_TYPE = ( Float() .with_variant(mysql.DOUBLE(asdecimal=False), "mysql") .with_variant(oracle.DOUBLE_PRECISION(), "oracle") .with_variant(postgresql.DOUBLE_PRECISION(), "postgresql") ) class Events(Base): # type: ignore """Event history data.""" __table_args__ = ( # Used for fetching events at a specific time # see logbook Index("ix_events_event_type_time_fired", "event_type", "time_fired"), {"mysql_default_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}, ) __tablename__ = TABLE_EVENTS event_id = Column(Integer, Identity(), primary_key=True) event_type = Column(String(MAX_LENGTH_EVENT_EVENT_TYPE)) event_data = Column(Text().with_variant(mysql.LONGTEXT, "mysql")) origin = Column(String(MAX_LENGTH_EVENT_ORIGIN)) time_fired = Column(DATETIME_TYPE, index=True) created = Column(DATETIME_TYPE, default=dt_util.utcnow) context_id = Column(String(MAX_LENGTH_EVENT_CONTEXT_ID), index=True) context_user_id = Column(String(MAX_LENGTH_EVENT_CONTEXT_ID), index=True) context_parent_id = Column(String(MAX_LENGTH_EVENT_CONTEXT_ID), index=True) def __repr__(self) -> str: """Return string representation of instance for debugging.""" return ( f"" ) @staticmethod def from_event(event, event_data=None): """Create an event database object from a native event.""" return Events( event_type=event.event_type, event_data=event_data or json.dumps(event.data, cls=JSONEncoder, separators=(",", ":")), origin=str(event.origin.value), time_fired=event.time_fired, context_id=event.context.id, context_user_id=event.context.user_id, context_parent_id=event.context.parent_id, ) def to_native(self, validate_entity_id=True): """Convert to a native HA Event.""" context = Context( id=self.context_id, user_id=self.context_user_id, parent_id=self.context_parent_id, ) try: return Event( self.event_type, json.loads(self.event_data), EventOrigin(self.origin), process_timestamp(self.time_fired), context=context, ) except ValueError: # When json.loads fails _LOGGER.exception("Error converting to event: %s", self) return None class States(Base): # type: ignore """State change history.""" __table_args__ = ( # Used for fetching the state of entities at a specific time # (get_states in history.py) Index("ix_states_entity_id_last_updated", "entity_id", "last_updated"), {"mysql_default_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}, ) __tablename__ = TABLE_STATES state_id = Column(Integer, Identity(), primary_key=True) domain = Column(String(MAX_LENGTH_STATE_DOMAIN)) entity_id = Column(String(MAX_LENGTH_STATE_ENTITY_ID)) state = Column(String(MAX_LENGTH_STATE_STATE)) attributes = Column(Text().with_variant(mysql.LONGTEXT, "mysql")) event_id = Column( Integer, ForeignKey("events.event_id", ondelete="CASCADE"), index=True ) last_changed = Column(DATETIME_TYPE, default=dt_util.utcnow) last_updated = Column(DATETIME_TYPE, default=dt_util.utcnow, index=True) created = Column(DATETIME_TYPE, default=dt_util.utcnow) old_state_id = Column(Integer, ForeignKey("states.state_id"), index=True) event = relationship("Events", uselist=False) old_state = relationship("States", remote_side=[state_id]) def __repr__(self) -> str: """Return string representation of instance for debugging.""" return ( f"" ) @staticmethod def from_event(event): """Create object from a state_changed event.""" entity_id = event.data["entity_id"] state = event.data.get("new_state") dbstate = States(entity_id=entity_id) # State got deleted if state is None: dbstate.state = "" dbstate.domain = split_entity_id(entity_id)[0] dbstate.attributes = "{}" dbstate.last_changed = event.time_fired dbstate.last_updated = event.time_fired else: dbstate.domain = state.domain dbstate.state = state.state dbstate.attributes = json.dumps( dict(state.attributes), cls=JSONEncoder, separators=(",", ":") ) dbstate.last_changed = state.last_changed dbstate.last_updated = state.last_updated return dbstate def to_native(self, validate_entity_id=True): """Convert to an HA state object.""" try: return State( self.entity_id, self.state, json.loads(self.attributes), process_timestamp(self.last_changed), process_timestamp(self.last_updated), # Join the events table on event_id to get the context instead # as it will always be there for state_changed events context=Context(id=None), validate_entity_id=validate_entity_id, ) except ValueError: # When json.loads fails _LOGGER.exception("Error converting row to state: %s", self) return None class StatisticResult(TypedDict): """Statistic result data class. Allows multiple datapoints for the same statistic_id. """ meta: StatisticMetaData stat: StatisticData class StatisticDataBase(TypedDict): """Mandatory fields for statistic data class.""" start: datetime class StatisticData(StatisticDataBase, total=False): """Statistic data class.""" mean: float min: float max: float last_reset: datetime | None state: float sum: float class StatisticsBase: """Statistics base class.""" id = Column(Integer, Identity(), primary_key=True) created = Column(DATETIME_TYPE, default=dt_util.utcnow) @declared_attr def metadata_id(self): """Define the metadata_id column for sub classes.""" return Column( Integer, ForeignKey(f"{TABLE_STATISTICS_META}.id", ondelete="CASCADE"), index=True, ) start = Column(DATETIME_TYPE, index=True) mean = Column(DOUBLE_TYPE) min = Column(DOUBLE_TYPE) max = Column(DOUBLE_TYPE) last_reset = Column(DATETIME_TYPE) state = Column(DOUBLE_TYPE) sum = Column(DOUBLE_TYPE) @classmethod def from_stats(cls, metadata_id: int, stats: StatisticData): """Create object from a statistics.""" return cls( # type: ignore metadata_id=metadata_id, **stats, ) class Statistics(Base, StatisticsBase): # type: ignore """Long term statistics.""" duration = timedelta(hours=1) __table_args__ = ( # Used for fetching statistics for a certain entity at a specific time Index("ix_statistics_statistic_id_start", "metadata_id", "start", unique=True), ) __tablename__ = TABLE_STATISTICS class StatisticsShortTerm(Base, StatisticsBase): # type: ignore """Short term statistics.""" duration = timedelta(minutes=5) __table_args__ = ( # Used for fetching statistics for a certain entity at a specific time Index( "ix_statistics_short_term_statistic_id_start", "metadata_id", "start", unique=True, ), ) __tablename__ = TABLE_STATISTICS_SHORT_TERM class StatisticMetaData(TypedDict): """Statistic meta data class.""" has_mean: bool has_sum: bool name: str | None source: str statistic_id: str unit_of_measurement: str | None class StatisticsMeta(Base): # type: ignore """Statistics meta data.""" __table_args__ = ( {"mysql_default_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}, ) __tablename__ = TABLE_STATISTICS_META id = Column(Integer, Identity(), primary_key=True) statistic_id = Column(String(255), index=True) source = Column(String(32)) unit_of_measurement = Column(String(255)) has_mean = Column(Boolean) has_sum = Column(Boolean) name = Column(String(255)) @staticmethod def from_meta(meta: StatisticMetaData) -> StatisticsMeta: """Create object from meta data.""" return StatisticsMeta(**meta) class RecorderRuns(Base): # type: ignore """Representation of recorder run.""" __table_args__ = (Index("ix_recorder_runs_start_end", "start", "end"),) __tablename__ = TABLE_RECORDER_RUNS run_id = Column(Integer, Identity(), primary_key=True) start = Column(DateTime(timezone=True), default=dt_util.utcnow) end = Column(DateTime(timezone=True)) closed_incorrect = Column(Boolean, default=False) created = Column(DateTime(timezone=True), default=dt_util.utcnow) def __repr__(self) -> str: """Return string representation of instance for debugging.""" end = ( f"'{self.end.isoformat(sep=' ', timespec='seconds')}'" if self.end else None ) return ( f"" ) def entity_ids(self, point_in_time=None): """Return the entity ids that existed in this run. Specify point_in_time if you want to know which existed at that point in time inside the run. """ session = Session.object_session(self) assert session is not None, "RecorderRuns need to be persisted" query = session.query(distinct(States.entity_id)).filter( States.last_updated >= self.start ) if point_in_time is not None: query = query.filter(States.last_updated < point_in_time) elif self.end is not None: query = query.filter(States.last_updated < self.end) return [row[0] for row in query] def to_native(self, validate_entity_id=True): """Return self, native format is this model.""" return self class SchemaChanges(Base): # type: ignore """Representation of schema version changes.""" __tablename__ = TABLE_SCHEMA_CHANGES change_id = Column(Integer, Identity(), primary_key=True) schema_version = Column(Integer) changed = Column(DateTime(timezone=True), default=dt_util.utcnow) def __repr__(self) -> str: """Return string representation of instance for debugging.""" return ( f"" ) class StatisticsRuns(Base): # type: ignore """Representation of statistics run.""" __tablename__ = TABLE_STATISTICS_RUNS run_id = Column(Integer, Identity(), primary_key=True) start = Column(DateTime(timezone=True)) def __repr__(self) -> str: """Return string representation of instance for debugging.""" return ( f"" ) @overload def process_timestamp(ts: None) -> None: ... @overload def process_timestamp(ts: datetime) -> datetime: ... def process_timestamp(ts: datetime | None) -> datetime | None: """Process a timestamp into datetime object.""" if ts is None: return None if ts.tzinfo is None: return ts.replace(tzinfo=dt_util.UTC) return dt_util.as_utc(ts) @overload def process_timestamp_to_utc_isoformat(ts: None) -> None: ... @overload def process_timestamp_to_utc_isoformat(ts: datetime) -> str: ... def process_timestamp_to_utc_isoformat(ts: datetime | None) -> str | None: """Process a timestamp into UTC isotime.""" if ts is None: return None if ts.tzinfo == dt_util.UTC: return ts.isoformat() if ts.tzinfo is None: return f"{ts.isoformat()}{DB_TIMEZONE}" return ts.astimezone(dt_util.UTC).isoformat() class LazyState(State): """A lazy version of core State.""" __slots__ = [ "_row", "entity_id", "state", "_attributes", "_last_changed", "_last_updated", "_context", ] def __init__(self, row): # pylint: disable=super-init-not-called """Init the lazy state.""" self._row = row self.entity_id = self._row.entity_id self.state = self._row.state or "" self._attributes = None self._last_changed = None self._last_updated = None self._context = None @property # type: ignore def attributes(self): """State attributes.""" if not self._attributes: try: self._attributes = json.loads(self._row.attributes) except ValueError: # When json.loads fails _LOGGER.exception("Error converting row to state: %s", self._row) self._attributes = {} return self._attributes @attributes.setter def attributes(self, value): """Set attributes.""" self._attributes = value @property # type: ignore def context(self): """State context.""" if not self._context: self._context = Context(id=None) return self._context @context.setter def context(self, value): """Set context.""" self._context = value @property # type: ignore def last_changed(self): """Last changed datetime.""" if not self._last_changed: self._last_changed = process_timestamp(self._row.last_changed) return self._last_changed @last_changed.setter def last_changed(self, value): """Set last changed datetime.""" self._last_changed = value @property # type: ignore def last_updated(self): """Last updated datetime.""" if not self._last_updated: self._last_updated = process_timestamp(self._row.last_updated) return self._last_updated @last_updated.setter def last_updated(self, value): """Set last updated datetime.""" self._last_updated = value def as_dict(self): """Return a dict representation of the LazyState. Async friendly. To be used for JSON serialization. """ if self._last_changed: last_changed_isoformat = self._last_changed.isoformat() else: last_changed_isoformat = process_timestamp_to_utc_isoformat( self._row.last_changed ) if self._last_updated: last_updated_isoformat = self._last_updated.isoformat() else: last_updated_isoformat = process_timestamp_to_utc_isoformat( self._row.last_updated ) return { "entity_id": self.entity_id, "state": self.state, "attributes": self._attributes or self.attributes, "last_changed": last_changed_isoformat, "last_updated": last_updated_isoformat, } def __eq__(self, other): """Return the comparison.""" return ( other.__class__ in [self.__class__, State] and self.entity_id == other.entity_id and self.state == other.state and self.attributes == other.attributes )