from __future__ import annotations import asyncio import contextlib from datetime import UTC, datetime, timedelta import json import logging import re import sqlite3 from typing import TYPE_CHECKING, Any, NamedTuple import aiosqlite import zigpy.appdb_schemas import zigpy.backups import zigpy.device from zigpy.device import Device, Status as DeviceStatus import zigpy.endpoint from zigpy.endpoint import Endpoint, Status as EndpointStatus import zigpy.exceptions import zigpy.group import zigpy.profiles import zigpy.quirks import zigpy.state import zigpy.types as t import zigpy.typing from zigpy.typing import UNDEFINED import zigpy.util from zigpy.zcl import ( AttributeClearedEvent, AttributeReadEvent, AttributeReportedEvent, AttributeUnsupportedEvent, AttributeUpdatedEvent, AttributeWrittenEvent, ClusterType, OtaQueryCacheClearedEvent, OtaQueryCacheUpdatedEvent, ) from zigpy.zcl.clusters.general import Basic, Ota from zigpy.zcl.foundation import Status from zigpy.zdo import types as zdo_t if TYPE_CHECKING: from zigpy.application import ControllerApplication from zigpy.zcl import Cluster MIN_SQLITE_VERSION = (3, 24, 0) if sqlite3.sqlite_version_info < MIN_SQLITE_VERSION: raise RuntimeError( f"zigpy requires SQLite {'.'.join(map(str, MIN_SQLITE_VERSION))} or newer " f"(found {sqlite3.sqlite_version})." ) LOGGER = logging.getLogger(__name__) DB_VERSION = 15 DB_V = f"_v{DB_VERSION}" UNIX_EPOCH = datetime.fromtimestamp(0, tz=UTC) DB_V_REGEX = re.compile(r"(?:_v\d+)?$") MIN_UPDATE_DELTA = timedelta(seconds=30).total_seconds() # The old attribute cache was a simple `attrid: value` mapping. This works 99.9% of the # time but unfortunately some devices reuse the same attribute ID on a standard cluster # for two separate purposes, using a manufacturer code to distinguish them. We migrate # attributes safely at runtime, once a device quirk has loaded and we can tell for sure # if the device has "colliding" attributes. UNMIGRATED_MANUFACTURER_CODE = -1 class AttributeCacheRow(NamedTuple): """A row from the attribute cache table. This format is internal and will change.""" ieee: t.EUI64 endpoint_id: int cluster_type: ClusterType cluster_id: int attr_id: int manufacturer_code: int | None status: Status value: Any last_updated: float def _register_sqlite_adapters(): def adapt_ieee(eui64): return str(eui64) sqlite3.register_adapter(t.EUI64, adapt_ieee) sqlite3.register_adapter(t.ExtendedPanId, adapt_ieee) def convert_ieee(s): return t.EUI64.convert(s.decode()) sqlite3.register_converter("ieee", convert_ieee) def aiosqlite_connect( database: str, iter_chunk_size: int = 64, **kwargs ) -> aiosqlite.Connection: """Copy of the the `aiosqlite.connect` function that connects using either the built-in `sqlite3` module or the imported `pysqlite3` module. """ return aiosqlite.Connection( connector=lambda: sqlite3.connect(str(database), **kwargs), iter_chunk_size=iter_chunk_size, ) def decode_str_attribute(value: str | bytes) -> str: if isinstance(value, str): return value return value.split(b"\x00", 1)[0].decode("utf-8") class PersistingListener(zigpy.util.CatchingTaskMixin): def __init__( self, connection: aiosqlite.Connection, application: ControllerApplication, ) -> None: _register_sqlite_adapters() self._db = connection self._application = application self._callback_handlers: asyncio.Queue = asyncio.Queue() self.running = False self._worker_task = asyncio.create_task(self._worker()) async def initialize_tables(self) -> None: async with self.execute("PRAGMA integrity_check") as cursor: rows = await cursor.fetchall() status = "\n".join(row[0] for row in rows) if status != "ok": LOGGER.error( "Zigbee database is corrupted, integrity check failed!\n%s", status ) async with self.execute("PRAGMA foreign_key_check") as cursor: rows = await cursor.fetchall() if rows: LOGGER.error( "Zigbee database is corrupted, foreign key check failed!\n%s", rows ) # Truncate the SQLite journal file instead of deleting it after transactions await self._set_isolation_level(None) await self.execute("PRAGMA journal_mode = WAL") await self.execute("PRAGMA synchronous = normal") await self.execute("PRAGMA temp_store = memory") await self._set_isolation_level("DEFERRED") await self.execute("PRAGMA foreign_keys = ON") await self._run_migrations() @classmethod async def new( cls, database_file: str, app: ControllerApplication ) -> PersistingListener: """Create an instance of persisting listener.""" sqlite_conn = await aiosqlite_connect( database_file, detect_types=sqlite3.PARSE_DECLTYPES, isolation_level="DEFERRED", # The default is "", an alias for "DEFERRED" ) listener = cls(sqlite_conn, app) try: await listener.initialize_tables() except Exception: # noqa: BLE001 await listener.shutdown() raise listener.running = True return listener async def _worker(self) -> None: """Process request in the received order.""" while True: cb_name, args = await self._callback_handlers.get() handler = getattr(self, cb_name) assert handler try: await handler(*args) except sqlite3.Error as exc: LOGGER.debug( "Error handling '%s' event with %s params: %s", cb_name, args, str(exc), exc_info=True, ) except Exception: # noqa: BLE001 LOGGER.exception( "Unexpected error while processing %s(%s)", cb_name, args ) self._callback_handlers.task_done() async def shutdown(self) -> None: """Shutdown connection.""" self.running = False await self._callback_handlers.join() if not self._worker_task.done(): self._worker_task.cancel() # Delete the journal on shutdown await self._set_isolation_level(None) await self.execute("PRAGMA wal_checkpoint;") await self._set_isolation_level("DEFERRED") await self._db.close() # FIXME: aiosqlite's thread won't always be closed immediately await asyncio.get_running_loop().run_in_executor(None, self._db.join) def register_cluster_events(self, cluster) -> None: cluster.on_event(AttributeReadEvent.event_type, self.on_attribute_read) cluster.on_event(AttributeReportedEvent.event_type, self.on_attribute_reported) cluster.on_event(AttributeUpdatedEvent.event_type, self.on_attribute_updated) cluster.on_event(AttributeWrittenEvent.event_type, self.on_attribute_written) cluster.on_event( AttributeUnsupportedEvent.event_type, self.on_attribute_unsupported ) cluster.on_event(AttributeClearedEvent.event_type, self.on_attribute_cleared) cluster.on_event( OtaQueryCacheUpdatedEvent.event_type, self.on_ota_query_cache_updated, ) cluster.on_event( OtaQueryCacheClearedEvent.event_type, self.on_ota_query_cache_cleared, ) def enqueue(self, cb_name: str, *args) -> None: """Enqueue an async callback handler action.""" if not self.running: LOGGER.debug("Discarding %s event", cb_name) return self._callback_handlers.put_nowait((cb_name, args)) async def _set_isolation_level(self, level: str | None): """Set the SQLite statement isolation level in a thread-safe way.""" await self._db._execute(lambda: setattr(self._db, "isolation_level", level)) def execute(self, *args, **kwargs): return self._db.execute(*args, **kwargs) async def executescript(self, sql): """Naive replacement for `sqlite3.Cursor.executescript` that does not execute a `COMMIT` before running the script. This extra `COMMIT` breaks transactions that run scripts. """ # XXX: This will break if you use a semicolon anywhere but at the end of a line for statement in sql.split(";"): # Strip SQL comments so that pysqlite3's implicit transaction DML detection # (which doesn't skip comments) sees the actual statement keyword statement = re.sub(r"--[^\n]*", "", statement) await self.execute(statement) def device_joined(self, device: Device) -> None: self.enqueue("_update_device_nwk", device.ieee, device.nwk) async def _update_device_nwk(self, ieee: t.EUI64, nwk: t.NWK) -> None: await self.execute(f"UPDATE devices{DB_V} SET nwk=? WHERE ieee=?", (nwk, ieee)) await self._db.commit() def device_initialized(self, device: Device) -> None: pass def device_left(self, device: Device) -> None: pass def device_last_seen_updated(self, device: Device, last_seen: datetime) -> None: """Device last_seen time is updated.""" self.enqueue("_save_device_last_seen", device.ieee, last_seen) async def _save_device_last_seen(self, ieee: t.EUI64, last_seen: datetime) -> None: q = f"""UPDATE devices{DB_V} SET last_seen=:ts WHERE ieee=:ieee AND :ts - last_seen > :min_update_delta""" await self.execute( q, { "ts": last_seen.timestamp(), "ieee": ieee, "min_update_delta": MIN_UPDATE_DELTA, }, ) await self._db.commit() def device_relays_updated(self, device: Device, relays: t.Relays | None) -> None: """Device relay list is updated.""" self.enqueue("_save_device_relays", device.ieee, relays) async def _save_device_relays(self, ieee: t.EUI64, relays: t.Relays | None) -> None: if relays is None: await self.execute(f"DELETE FROM relays{DB_V} WHERE ieee = ?", (ieee,)) else: q = f"""INSERT INTO relays{DB_V} VALUES (:ieee, :relays) ON CONFLICT (ieee) DO UPDATE SET relays=excluded.relays WHERE relays != :relays""" await self.execute(q, {"ieee": ieee, "relays": relays.serialize()}) await self._db.commit() def neighbors_updated(self, ieee: t.EUI64, neighbors: list[zdo_t.Neighbor]) -> None: """Neighbor update from Mgmt_Lqi_req.""" self.enqueue("_neighbors_updated", ieee, neighbors) async def _neighbors_updated( self, ieee: t.EUI64, neighbors: list[zdo_t.Neighbor] ) -> None: await self.execute(f"DELETE FROM neighbors{DB_V} WHERE device_ieee = ?", [ieee]) rows = [(ieee, *neighbor.as_tuple()) for neighbor in neighbors] await self._db.executemany( f"INSERT INTO neighbors{DB_V} VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", rows ) await self._db.commit() def routes_updated(self, ieee: t.EUI64, routes: list[zdo_t.Route]) -> None: """Route update from Mgmt_Rtg_req.""" self.enqueue("_routes_updated", ieee, routes) async def _routes_updated(self, ieee: t.EUI64, routes: list[zdo_t.Route]) -> None: await self.execute(f"DELETE FROM routes{DB_V} WHERE device_ieee = ?", [ieee]) rows = [(ieee, *route.as_tuple()) for route in routes] await self._db.executemany( f"INSERT INTO routes{DB_V} VALUES (?,?,?,?,?,?,?,?)", rows ) await self._db.commit() def group_added(self, group: zigpy.group.Group) -> None: """Group is added.""" self.enqueue("_group_added", group) async def _group_added(self, group: zigpy.group.Group) -> None: q = f"""INSERT INTO groups{DB_V} VALUES (?, ?) ON CONFLICT (group_id) DO UPDATE SET name=excluded.name""" await self.execute(q, (group.group_id, group.name)) await self._db.commit() def group_member_added(self, group: zigpy.group.Group, ep: Endpoint) -> None: """Called when a group member is added.""" self.enqueue("_group_member_added", group, ep) async def _group_member_added(self, group: zigpy.group.Group, ep: Endpoint) -> None: q = f"""INSERT INTO group_members{DB_V} VALUES (?, ?, ?) ON CONFLICT DO NOTHING""" await self.execute(q, (group.group_id, *ep.unique_id)) await self._db.commit() def group_member_removed(self, group: zigpy.group.Group, ep: Endpoint) -> None: """Called when a group member is removed.""" self.enqueue("_group_member_removed", group, ep) async def _group_member_removed( self, group: zigpy.group.Group, ep: Endpoint ) -> None: q = f"""DELETE FROM group_members{DB_V} WHERE group_id=? AND ieee=? AND endpoint_id=?""" await self.execute(q, (group.group_id, *ep.unique_id)) await self._db.commit() def group_removed(self, group: zigpy.group.Group) -> None: """Called when a group is removed.""" self.enqueue("_group_removed", group) async def _group_removed(self, group: zigpy.group.Group) -> None: q = f"DELETE FROM groups{DB_V} WHERE group_id=?" await self.execute(q, (group.group_id,)) await self._db.commit() def device_removed(self, device: Device) -> None: self.enqueue("_remove_device", device) async def _remove_device(self, device: Device) -> None: await self.execute(f"DELETE FROM devices{DB_V} WHERE ieee = ?", (device.ieee,)) await self._db.commit() def raw_device_initialized(self, device: Device) -> None: self.enqueue("_save_device", device) async def _save_device(self, device: Device) -> None: q = f"""INSERT INTO devices{DB_V} (ieee, nwk, status, last_seen) VALUES (?, ?, ?, ?) ON CONFLICT (ieee) DO UPDATE SET nwk=excluded.nwk, status=excluded.status, last_seen=excluded.last_seen""" await self.execute( q, ( device.ieee, device.nwk, device.status, (device._last_seen or UNIX_EPOCH).timestamp(), ), ) if device.node_desc is not None: await self._save_node_descriptor(device) if isinstance(device, zigpy.quirks.BaseCustomDevice): await self._db.commit() return await self._save_endpoints(device) for ep in device.non_zdo_endpoints: await self._save_clusters(ep) await self._save_attribute_cache(ep) await self._save_unsupported_attributes(ep) await self._save_ota_query_cache(device) await self._db.commit() async def _save_endpoints(self, device: Device) -> None: rows = [ ( device.ieee, ep.endpoint_id, ep.profile_id, ep.device_type, ep.status, ) for ep in device.non_zdo_endpoints ] q = f"""INSERT INTO endpoints{DB_V} VALUES (?, ?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id) DO UPDATE SET profile_id=excluded.profile_id, device_type=excluded.device_type, status=excluded.status""" await self._db.executemany(q, rows) async def _save_node_descriptor(self, device: Device) -> None: if device.node_desc is None: return q = f"""INSERT INTO node_descriptors{DB_V} VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (ieee) DO UPDATE SET logical_type=excluded.logical_type, complex_descriptor_available=excluded.complex_descriptor_available, user_descriptor_available=excluded.user_descriptor_available, reserved=excluded.reserved, aps_flags=excluded.aps_flags, frequency_band=excluded.frequency_band, mac_capability_flags=excluded.mac_capability_flags, manufacturer_code=excluded.manufacturer_code, maximum_buffer_size=excluded.maximum_buffer_size, maximum_incoming_transfer_size=excluded.maximum_incoming_transfer_size, server_mask=excluded.server_mask, maximum_outgoing_transfer_size=excluded.maximum_outgoing_transfer_size, descriptor_capability_field=excluded.descriptor_capability_field""" await self.execute(q, (device.ieee, *device.node_desc.as_tuple())) async def _save_clusters(self, endpoint: Endpoint) -> None: clusters = [ ( endpoint.device.ieee, endpoint.endpoint_id, cluster.cluster_type, cluster.cluster_id, ) for cluster in endpoint.clusters ] q = f"""INSERT INTO clusters{DB_V} VALUES (?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id, cluster_type, cluster_id) DO NOTHING""" await self._db.executemany(q, clusters) async def _save_attribute_cache(self, ep: Endpoint) -> None: clusters = [ ( ep.device.ieee, ep.endpoint_id, cluster.cluster_type, cluster.cluster_id, attrid, manufacturer_code, Status.SUCCESS, cache_item.value, cache_item.last_updated.timestamp(), ) for cluster in ep.clusters for ( attrid, manufacturer_code, ), cache_item in cluster._attr_cache._cache.items() ] q = f"""INSERT INTO attributes_cache{DB_V} (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code_idx) DO UPDATE SET status=excluded.status, value=excluded.value, last_updated=excluded.last_updated""" await self._db.executemany(q, clusters) async def _save_unsupported_attributes(self, ep: Endpoint) -> None: clusters = [ ( ep.device.ieee, ep.endpoint_id, cluster.cluster_type, cluster.cluster_id, attrid, manufacturer_code, Status.UNSUPPORTED_ATTRIBUTE, None, datetime.now(UTC).timestamp(), ) for cluster in ep.clusters for (attrid, manufacturer_code) in cluster._attr_cache._unsupported ] q = f"""INSERT INTO attributes_cache{DB_V} (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code_idx) DO NOTHING""" await self._db.executemany(q, clusters) async def _save_ota_query_cache(self, device: Device) -> None: rows = [] for ep in device.non_zdo_endpoints: for cluster in ep.clusters: if isinstance(cluster, Ota) and cluster.last_query_cmd is not None: cmd = cluster.last_query_cmd rows.append( ( device.ieee, ep.endpoint_id, cmd.manufacturer_code, cmd.image_type, cmd.current_file_version, getattr(cmd, "hardware_version", None), datetime.now(UTC).timestamp(), ) ) if rows: q = f"""INSERT INTO ota_query_cache{DB_V} (ieee, endpoint_id, manufacturer_code, image_type, current_file_version, hardware_version, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id) DO UPDATE SET manufacturer_code=excluded.manufacturer_code, image_type=excluded.image_type, current_file_version=excluded.current_file_version, hardware_version=excluded.hardware_version, last_updated=excluded.last_updated""" await self._db.executemany(q, rows) def on_attribute_read(self, event: AttributeReadEvent) -> None: self.enqueue("_save_attribute", event) def on_attribute_reported(self, event: AttributeReportedEvent) -> None: self.enqueue("_save_attribute", event) def on_attribute_updated(self, event: AttributeUpdatedEvent) -> None: self.enqueue("_save_attribute", event) def on_attribute_written(self, event: AttributeWrittenEvent) -> None: self.enqueue("_save_attribute", event) async def _save_attribute( self, event: AttributeReadEvent | AttributeReportedEvent | AttributeUpdatedEvent | AttributeWrittenEvent, ) -> None: if isinstance(event, AttributeWrittenEvent) and event.status != Status.SUCCESS: LOGGER.debug("Ignoring failed attribute write event: %s", event) return await self.execute( f""" INSERT INTO attributes_cache{DB_V} (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (:ieee, :endpoint_id, :cluster_type, :cluster_id, :attr_id, :manufacturer_code, :status, :value, :timestamp) ON CONFLICT (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code_idx) DO UPDATE SET status=excluded.status, value=excluded.value, last_updated=excluded.last_updated WHERE value != excluded.value OR status != excluded.status OR :timestamp - last_updated > :min_update_delta """, { "ieee": event.device_ieee, "endpoint_id": event.endpoint_id, "cluster_type": event.cluster_type, "cluster_id": event.cluster_id, "attr_id": event.attribute_id, "manufacturer_code": event.manufacturer_code, "status": Status.SUCCESS, "value": event.value, "timestamp": datetime.now(UTC).timestamp(), "min_update_delta": MIN_UPDATE_DELTA, }, ) await self._db.commit() def on_attribute_cleared(self, event: AttributeClearedEvent) -> None: self.enqueue("_clear_attribute", event) async def _clear_attribute(self, event: AttributeClearedEvent) -> None: q = f""" DELETE FROM attributes_cache{DB_V} WHERE ieee = :ieee AND endpoint_id = :endpoint_id AND cluster_type = :cluster_type AND cluster_id = :cluster_id AND attr_id = :attr_id AND manufacturer_code IS NOT DISTINCT FROM :manufacturer_code """ await self.execute( q, { "ieee": event.device_ieee, "endpoint_id": event.endpoint_id, "cluster_type": event.cluster_type, "cluster_id": event.cluster_id, "attr_id": event.attribute_id, "manufacturer_code": event.manufacturer_code, }, ) await self._db.commit() def on_attribute_unsupported(self, event: AttributeUnsupportedEvent) -> None: self.enqueue("_unsupported_attribute_added", event) async def _unsupported_attribute_added( self, event: AttributeUnsupportedEvent ) -> None: q = f"""INSERT INTO attributes_cache{DB_V} (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (:ieee, :endpoint_id, :cluster_type, :cluster_id, :attr_id, :manufacturer_code, :status, :value, :timestamp) ON CONFLICT (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code_idx) DO UPDATE SET status=excluded.status, value=excluded.value, last_updated=excluded.last_updated""" await self.execute( q, { "ieee": event.device_ieee, "endpoint_id": event.endpoint_id, "cluster_type": event.cluster_type, "cluster_id": event.cluster_id, "attr_id": event.attribute_id, "manufacturer_code": event.manufacturer_code, "status": Status.UNSUPPORTED_ATTRIBUTE, "value": None, "timestamp": datetime.now(UTC).timestamp(), }, ) await self._db.commit() def on_ota_query_cache_updated(self, event: OtaQueryCacheUpdatedEvent) -> None: self.enqueue("_save_ota_query_cache_entry", event) async def _save_ota_query_cache_entry( self, event: OtaQueryCacheUpdatedEvent ) -> None: q = f"""INSERT INTO ota_query_cache{DB_V} (ieee, endpoint_id, manufacturer_code, image_type, current_file_version, hardware_version, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (ieee, endpoint_id) DO UPDATE SET manufacturer_code=excluded.manufacturer_code, image_type=excluded.image_type, current_file_version=excluded.current_file_version, hardware_version=excluded.hardware_version, last_updated=excluded.last_updated""" await self.execute( q, ( event.device_ieee, event.endpoint_id, event.manufacturer_code, event.image_type, event.current_file_version, event.hardware_version, datetime.now(UTC).timestamp(), ), ) await self._db.commit() def on_ota_query_cache_cleared(self, event: OtaQueryCacheClearedEvent) -> None: self.enqueue("_delete_ota_query_cache_entry", event) async def _delete_ota_query_cache_entry( self, event: OtaQueryCacheClearedEvent ) -> None: await self.execute( f"DELETE FROM ota_query_cache{DB_V} WHERE ieee = ? AND endpoint_id = ?", (event.device_ieee, event.endpoint_id), ) await self._db.commit() def network_backup_created(self, backup: zigpy.backups.NetworkBackup) -> None: self.enqueue("_network_backup_created", json.dumps(backup.as_dict())) async def _network_backup_created(self, backup_json: str) -> None: q = f"""INSERT INTO network_backups{DB_V} VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET backup_json=excluded.backup_json""" await self.execute(q, (None, backup_json)) await self._db.commit() def network_backup_removed(self, backup: zigpy.backups.NetworkBackup) -> None: self.enqueue("_network_backup_removed", backup.backup_time) async def _network_backup_removed(self, backup_time: datetime) -> None: q = f"""DELETE FROM network_backups{DB_V} WHERE json_extract(backup_json, '$.backup_time')=?""" await self.execute(q, (backup_time.isoformat(),)) await self._db.commit() async def _read_all_attributes( self, ) -> list[AttributeCacheRow]: """Read all attribute rows from the database.""" async with self.execute( f""" SELECT ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated FROM attributes_cache{DB_V} """ ) as cursor: return [AttributeCacheRow(*row) for row in await cursor.fetchall()] async def load(self) -> None: LOGGER.debug("Loading application state") await self._load_devices() await self._load_node_descriptors() await self._load_endpoints() await self._load_clusters() # Read all attribute rows from the database once all_attributes = await self._read_all_attributes() # First pass: populate cache on bare clusters for quirks await self._populate_attribute_cache(all_attributes) for device in self._application.devices.values(): # Populate the device signature before we apply any quirks, which can modify # the device structure (for now) device.original_signature = device.get_signature() self._application.devices[device.ieee] = zigpy.quirks.get_device(device) # Clear the attribute cache to ensure the quirked state is correct for device in self._application.devices.values(): for ep in device.non_zdo_endpoints: for cluster in ep.in_clusters.values(): cluster._attr_cache.clear() for cluster in ep.out_clusters.values(): cluster._attr_cache.clear() # Second pass: populate the attribute cache for the final device state and # migrate attributes with unknown manufacturer codes to the correct codes. Only # the migration pass modifies the database so we do it in a transaction. async with self._transaction(): await self._populate_attribute_cache(all_attributes, migrate=True) await self._load_groups() await self._load_group_members() await self._load_relays() await self._load_neighbors() await self._load_routes() await self._load_network_backups() await self._load_ota_query_cache() await self._db.commit() await self._register_device_listeners() async def _populate_attribute_cache( self, rows: list[AttributeCacheRow], *, migrate: bool = False, ) -> None: """Populate cluster attribute cache from pre-loaded rows. When `migrate` is True, unmigrated rows with ambiguous attribute IDs are resolved using the (now-quirked) cluster definitions and the database is updated to match. """ for row in rows: dev = self._application.get_device(row.ieee) LOGGER.debug( "[0x%04x:%s:0x%04x] Loading attribute %s=%r status=%r mfg_code=%r", dev.nwk, row.endpoint_id, row.cluster_id, ( row.attr_id if isinstance(row.attr_id, str) else f"0x{row.attr_id:04x}" ), row.value, row.status, row.manufacturer_code, ) if row.endpoint_id not in dev.endpoints: continue ep = dev.endpoints[row.endpoint_id] clusters = ( ep.in_clusters if row.cluster_type == ClusterType.Server else ep.out_clusters ) if row.cluster_id not in clusters: LOGGER.debug("Unknown ZCL cluster, skipping") continue cluster = clusters[row.cluster_id] # For unmigrated rows on the second pass, try to resolve the # manufacturer code using the full quirk cluster definitions manufacturer_code = row.manufacturer_code if migrate and row.manufacturer_code == UNMIGRATED_MANUFACTURER_CODE: resolved_manufacturer_code = self._resolve_unmigrated_attribute( cluster=cluster, attr_id=row.attr_id, value=row.value, dev=dev, ) if resolved_manufacturer_code is not UNDEFINED: manufacturer_code = resolved_manufacturer_code row_params = { "ieee": row.ieee, "endpoint_id": row.endpoint_id, "cluster_type": row.cluster_type, "cluster_id": row.cluster_id, "attr_id": row.attr_id, } # Delete the unmigrated row and re-insert with the resolved # manufacturer code. INSERT OR IGNORE handles the case where a row # with the resolved code already exists (e.g. the user manually # read the attribute through the UI). await self.execute( f""" DELETE FROM attributes_cache{DB_V} WHERE ieee = :ieee AND endpoint_id = :endpoint_id AND cluster_type = :cluster_type AND cluster_id = :cluster_id AND attr_id = :attr_id AND manufacturer_code = :old_manufacturer_code """, { **row_params, "old_manufacturer_code": UNMIGRATED_MANUFACTURER_CODE, }, ) async with self.execute( f""" INSERT OR IGNORE INTO attributes_cache{DB_V} (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (:ieee, :endpoint_id, :cluster_type, :cluster_id, :attr_id, :manufacturer_code, :status, :value, :last_updated) """, { **row_params, "manufacturer_code": manufacturer_code, "status": row.status, "value": row.value, "last_updated": row.last_updated, }, ) as cursor: # A resolved row already exists, it will populate the cache if cursor.rowcount == 0: continue try: attr_def = cluster.find_attribute( row.attr_id, manufacturer_code=( UNDEFINED if manufacturer_code == UNMIGRATED_MANUFACTURER_CODE else manufacturer_code ), ) except KeyError: LOGGER.debug("Unknown ZCL attribute, skipping") # Unsupported unknown attributes are dropped if row.status == Status.SUCCESS: cluster._attr_cache.set_legacy_value( row.attr_id, row.value, last_updated=datetime.fromtimestamp(row.last_updated, UTC), ) continue if row.status == Status.SUCCESS: cluster._attr_cache.set_value( attr_def, row.value, last_updated=datetime.fromtimestamp(row.last_updated, UTC), ) else: cluster._attr_cache.mark_unsupported(attr_def) continue # Populate the device's manufacturer and model attributes if ( row.cluster_id == Basic.cluster_id and attr_def == Basic.AttributeDefs.manufacturer ): dev.manufacturer = decode_str_attribute(row.value) elif ( row.cluster_id == Basic.cluster_id and attr_def == Basic.AttributeDefs.model ): dev.model = decode_str_attribute(row.value) @staticmethod def _resolve_unmigrated_attribute( cluster: Cluster, attr_id: int, value: Any, dev: Device, ) -> int | None | zigpy.typing.UndefinedType: """Try to resolve the manufacturer code for an unmigrated attribute. When multiple attribute definitions share an ID, the manufacturer-specific one is preferred if there are exactly two candidates (one standard, one manufacturer-specific). Otherwise the attribute is considered ambiguous and skipped (left unmigrated). Returns the resolved manufacturer code (including `None`), or UNDEFINED if unresolvable. """ try: attr_defs = cluster.find_attributes(attr_id) except KeyError: LOGGER.debug( "Unable to find any attributes %r=%r on cluster %r for %r" " for data migration, skipping", attr_id, value, cluster, dev, ) return UNDEFINED manuf_attrs = [ a for a in attr_defs if cluster._get_effective_manufacturer_code(a) is not None ] if len(attr_defs) == 1: attr_def = attr_defs[0] elif len(attr_defs) == 2 and len(manuf_attrs) == 1: # One standard + one manufacturer-specific: prefer manufacturer-specific attr_def = manuf_attrs[0] else: LOGGER.debug( "Unable to find unique attribute %r=%r on cluster %r for %r" " for data migration, skipping (candidates: %r)", attr_id, value, cluster, dev, attr_defs, ) return UNDEFINED return cluster._get_effective_manufacturer_code(attr_def) async def _load_devices(self) -> None: async with self.execute(f"SELECT * FROM devices{DB_V}") as cursor: async for ieee, nwk, status, last_seen in cursor: dev = self._application.add_device(ieee, nwk) dev.status = DeviceStatus(status) if last_seen > 0: dev.last_seen = last_seen async def _load_node_descriptors(self) -> None: async with self.execute(f"SELECT * FROM node_descriptors{DB_V}") as cursor: async for ieee, *fields in cursor: dev = self._application.get_device(ieee) dev.node_desc = zdo_t.NodeDescriptor(*fields) assert dev.node_desc.is_valid async def _load_endpoints(self) -> None: async with self.execute(f"SELECT * FROM endpoints{DB_V}") as cursor: async for ieee, epid, profile_id, device_type, status in cursor: dev = self._application.get_device(ieee) ep = dev.add_endpoint(epid) ep.profile_id = profile_id ep.status = EndpointStatus(status) if profile_id == zigpy.profiles.zha.PROFILE_ID: ep.device_type = zigpy.profiles.zha.DeviceType(device_type) elif profile_id == zigpy.profiles.zll.PROFILE_ID: ep.device_type = zigpy.profiles.zll.DeviceType(device_type) else: ep.device_type = device_type async def _load_clusters(self) -> None: async with self.execute(f"SELECT * FROM clusters{DB_V}") as cursor: async for ieee, endpoint_id, cluster_type, cluster_id in cursor: dev = self._application.get_device(ieee) ep = dev.endpoints[endpoint_id] if ClusterType(cluster_type) == ClusterType.Server: ep.add_input_cluster(cluster_id) else: ep.add_output_cluster(cluster_id) async def _load_groups(self) -> None: async with self.execute(f"SELECT * FROM groups{DB_V}") as cursor: async for group_id, name in cursor: self._application.groups.add_group(group_id, name, suppress_event=True) async def _load_group_members(self) -> None: async with self.execute(f"SELECT * FROM group_members{DB_V}") as cursor: async for group_id, ieee, ep_id in cursor: dev = self._application.get_device(ieee) group = self._application.groups[group_id] group.add_member(dev.endpoints[ep_id], suppress_event=True) async def _load_relays(self) -> None: async with self.execute(f"SELECT * FROM relays{DB_V}") as cursor: async for ieee, value in cursor: dev = self._application.get_device(ieee) relays, _ = t.Relays.deserialize(value) dev.relays = zigpy.util.filter_relays(relays) async def _load_neighbors(self) -> None: async with self.execute(f"SELECT * FROM neighbors{DB_V}") as cursor: async for ieee, *fields in cursor: neighbor = zdo_t.Neighbor(*fields) self._application.topology.neighbors[ieee].append(neighbor) async def _load_routes(self) -> None: async with self.execute(f"SELECT * FROM routes{DB_V}") as cursor: async for ieee, *fields in cursor: route = zdo_t.Route(*fields) self._application.topology.routes[ieee].append(route) async def _load_network_backups(self) -> None: self._application.backups.backups.clear() async with self.execute( f"SELECT * FROM network_backups{DB_V} ORDER BY id" ) as cursor: backups = [] async for _id, backup_json in cursor: backup = zigpy.backups.NetworkBackup.from_dict(json.loads(backup_json)) backups.append(backup) backups.sort(key=lambda b: b.backup_time) for backup in backups: self._application.backups.add_backup(backup, suppress_event=True) async def _load_ota_query_cache(self) -> None: async with self.execute(f"SELECT * FROM ota_query_cache{DB_V}") as cursor: async for ( ieee, endpoint_id, manufacturer_code, image_type, current_file_version, hardware_version, _last_updated, ) in cursor: try: dev = self._application.get_device(ieee) ep = dev.endpoints[endpoint_id] except KeyError: # Quirks or firmware updates can remove endpoints/clusters continue field_control = ( Ota.QueryNextImageCommand.FieldControl(0) if hardware_version is None else Ota.QueryNextImageCommand.FieldControl.HardwareVersion ) cmd = Ota.QueryNextImageCommand( field_control=field_control, manufacturer_code=manufacturer_code, image_type=image_type, current_file_version=current_file_version, ) if hardware_version is not None: cmd.hardware_version = hardware_version # Restore to the first OTA cluster found; prefer client # (out_clusters) since that's where runtime routing places it # when both cluster types exist. for clusters in (ep.out_clusters, ep.in_clusters): cluster = clusters.get(Ota.cluster_id) if isinstance(cluster, Ota): cluster.last_query_cmd = cmd break async def _register_device_listeners(self) -> None: for dev in self._application.devices.values(): dev.add_context_listener(self) @contextlib.asynccontextmanager async def _transaction(self): await self.execute("BEGIN TRANSACTION") try: yield except Exception: # noqa: BLE001 await self.execute("ROLLBACK") raise else: await self.execute("COMMIT") async def _get_table_versions(self) -> dict[str, int]: tables = {} async with self.execute( "SELECT name FROM sqlite_master WHERE type='table'" ) as cursor: async for (name,) in cursor: # Ignore tables internal to SQLite if name.startswith("sqlite_"): continue # The regex will always return a match match = DB_V_REGEX.search(name) assert match is not None tables[name] = int(match.group(0)[2:] or "0") return tables async def _table_exists(self, name: str) -> bool: return name in (await self._get_table_versions()) async def _run_migrations(self) -> bool: """Migrates the database to the newest schema, returning True if migrations ran.""" tables = await self._get_table_versions() tables_version = max(tables.values(), default=0) async with self.execute("PRAGMA user_version") as cursor: (db_version,) = await cursor.fetchone() LOGGER.debug( "Current database version is v%s (table version v%s)", db_version, tables_version, ) # Table version suffixes were introduced in v4. If the table version suffix does # not match `user_version`, either zigpy was downgraded to a *really* old # version (July 2021), or it's corrupt. Running migrations could delete existing # table data, and since we cannot guarantee the schema is intact, fail early. if tables_version >= 4 and tables_version != db_version: raise zigpy.exceptions.CorruptDatabase( f"The `zigbee.db` database version ({db_version}) does not match its" f" max table version ({tables_version}). The database is inconsistent.", ) if db_version == 0 and not tables: # If this is a brand new database, just load the current schema await self.executescript(zigpy.appdb_schemas.SCHEMAS[DB_VERSION]) return False elif db_version > DB_VERSION: LOGGER.error( "This zigpy release uses database schema v%s but the database is v%s." " Downgrading zigpy is *not* recommended and may result in data loss." " Use at your own risk.", DB_VERSION, db_version, ) return False # All migrations must succeed. If any fail, the database is not touched. async with self._transaction(): for migration, to_db_version in [ (self._migrate_to_v4, 4), (self._migrate_to_v5, 5), (self._migrate_to_v6, 6), (self._migrate_to_v7, 7), (self._migrate_to_v8, 8), (self._migrate_to_v9, 9), (self._migrate_to_v10, 10), (self._migrate_to_v11, 11), (self._migrate_to_v12, 12), (self._migrate_to_v13, 13), (self._migrate_to_v14, 14), (self._migrate_to_v15, 15), ]: if db_version >= min(to_db_version, DB_VERSION): continue LOGGER.info( "Migrating database from v%d to v%d", db_version, to_db_version ) await self.executescript(zigpy.appdb_schemas.SCHEMAS[to_db_version]) await migration() db_version = to_db_version return True async def _migrate_tables( self, table_map: dict[str, str | None], *, errors: str = "raise" ): """Copy rows from one set of tables into another.""" # Extract the "old" table version suffix tables = await self._get_table_versions() old_table_name = list(table_map.keys())[0] old_version = tables[old_table_name] # Check which tables would not be migrated old_tables = [t for t, v in tables.items() if v == old_version] unmigrated_old_tables = [t for t in old_tables if t not in table_map] if unmigrated_old_tables: raise RuntimeError( f"The following tables were not migrated: {unmigrated_old_tables}" ) # Insertion order matters for foreign key constraints but any rows that fail # to insert due to constraint violations can be discarded for old_table, new_table in table_map.items(): # Ignore tables without a migration if new_table is None: continue # Use explicit column names to skip generated columns automatically async with self.execute(f"PRAGMA table_info({old_table})") as cursor: columns = [row[1] async for row in cursor] col_list = ", ".join(columns) placeholders = ", ".join("?" * len(columns)) select_sql = f"SELECT {col_list} FROM {old_table}" insert_sql = f"INSERT INTO {new_table} ({col_list}) VALUES ({placeholders})" async with self.execute(select_sql) as cursor: async for row in cursor: try: await self.execute(insert_sql, row) except sqlite3.IntegrityError as e: if errors == "raise": raise elif errors == "warn": LOGGER.warning( "Failed to migrate row %s%s: %s", old_table, row, e ) elif errors == "ignore": pass else: raise ValueError( f"Invalid value for `errors`: {errors!r}" ) from e async def _migrate_to_v4(self): """Schema v4 expanded the node descriptor and neighbor table columns""" # The `node_descriptors` table was added in v1 if await self._table_exists("node_descriptors"): async with self.execute("SELECT * FROM node_descriptors") as cur: async for dev_ieee, value in cur: node_desc, rest = zdo_t.NodeDescriptor.deserialize(value) assert not rest await self.execute( "INSERT INTO node_descriptors_v4" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (dev_ieee, *node_desc.as_tuple()), ) # The `neighbors` table was added in v3 but the version number was not # incremented. It may not exist. if await self._table_exists("neighbors"): async with self.execute("SELECT * FROM neighbors") as cur: async for dev_ieee, epid, ieee, nwk, packed, prm, depth, lqi in cur: neighbor = zdo_t.Neighbor( extended_pan_id=epid, ieee=ieee, nwk=nwk, permit_joining=prm, depth=depth, lqi=lqi, reserved2=0b000000, **zdo_t.Neighbor._parse_packed(packed), ) await self.execute( "INSERT INTO neighbors_v4 VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (dev_ieee, *neighbor.as_tuple()), ) async def _migrate_to_v5(self): """Schema v5 introduced global table version suffixes and removed stale rows""" await self._migrate_tables( { "devices": "devices_v5", "endpoints": "endpoints_v5", "clusters": "in_clusters_v5", "output_clusters": "out_clusters_v5", "groups": "groups_v5", "group_members": "group_members_v5", "relays": "relays_v5", "attributes": "attributes_cache_v5", # These were migrated in v4 "neighbors_v4": "neighbors_v5", "node_descriptors_v4": "node_descriptors_v5", # Explicitly specify which tables will not be migrated "neighbors": None, "node_descriptors": None, }, errors="warn", ) async def _migrate_to_v6(self): """Schema v6 relaxed the `attribute_cache` table schema to ignore endpoints""" await self._migrate_tables( { "devices_v5": "devices_v6", "endpoints_v5": "endpoints_v6", "in_clusters_v5": "in_clusters_v6", "out_clusters_v5": "out_clusters_v6", "groups_v5": "groups_v6", "group_members_v5": "group_members_v6", "relays_v5": "relays_v6", "attributes_cache_v5": "attributes_cache_v6", "neighbors_v5": "neighbors_v6", "node_descriptors_v5": "node_descriptors_v6", } ) # See if we can migrate any `attributes_cache` rows skipped by the v5 migration if await self._table_exists("attributes"): async with self.execute("SELECT count(*) FROM attributes") as cur: (num_attrs_v4,) = await cur.fetchone() async with self.execute("SELECT count(*) FROM attributes_cache_v6") as cur: (num_attrs_v6,) = await cur.fetchone() if num_attrs_v6 < num_attrs_v4: LOGGER.warning( "Migrating up to %d rows skipped by v5 migration", num_attrs_v4 - num_attrs_v6, ) await self._migrate_tables( { "attributes": "attributes_cache_v6", "devices": None, "endpoints": None, "clusters": None, "neighbors": None, "node_descriptors": None, "output_clusters": None, "groups": None, "group_members": None, "relays": None, }, errors="ignore", ) async def _migrate_to_v7(self): """Schema v7 added the `unsupported_attributes` table.""" await self._migrate_tables( { "devices_v6": "devices_v7", "endpoints_v6": "endpoints_v7", "in_clusters_v6": "in_clusters_v7", "out_clusters_v6": "out_clusters_v7", "groups_v6": "groups_v7", "group_members_v6": "group_members_v7", "relays_v6": "relays_v7", "attributes_cache_v6": "attributes_cache_v7", "neighbors_v6": "neighbors_v7", "node_descriptors_v6": "node_descriptors_v7", } ) async def _migrate_to_v8(self): """Schema v8 added the `devices_v8.last_seen` column.""" async with self.execute("SELECT * FROM devices_v7") as cursor: async for ieee, nwk, status in cursor: # Set the default `last_seen` to the unix epoch await self.execute( "INSERT INTO devices_v8 VALUES (?, ?, ?, ?)", (ieee, nwk, status, 0), ) # Copy the devices table first, it should have no conflicts await self._migrate_tables( { "endpoints_v7": "endpoints_v8", "in_clusters_v7": "in_clusters_v8", "out_clusters_v7": "out_clusters_v8", "groups_v7": "groups_v8", "group_members_v7": "group_members_v8", "relays_v7": "relays_v8", "attributes_cache_v7": "attributes_cache_v8", "neighbors_v7": "neighbors_v8", "node_descriptors_v7": "node_descriptors_v8", "unsupported_attributes_v7": "unsupported_attributes_v8", "devices_v7": None, } ) async def _migrate_to_v9(self): """Schema v9 changed the data type of the `devices_v8.last_seen` column.""" await self.execute( """INSERT INTO devices_v9 (ieee, nwk, status, last_seen) SELECT ieee, nwk, status, last_seen / 1000.0 FROM devices_v8""" ) await self._migrate_tables( { "endpoints_v8": "endpoints_v9", "in_clusters_v8": "in_clusters_v9", "out_clusters_v8": "out_clusters_v9", "groups_v8": "groups_v9", "group_members_v8": "group_members_v9", "relays_v8": "relays_v9", "attributes_cache_v8": "attributes_cache_v9", "neighbors_v8": "neighbors_v9", "node_descriptors_v8": "node_descriptors_v9", "unsupported_attributes_v8": "unsupported_attributes_v9", "devices_v8": None, } ) async def _migrate_to_v10(self): """Schema v10 added a new `network_backups_v10` table.""" await self._migrate_tables( { "devices_v9": "devices_v10", "endpoints_v9": "endpoints_v10", "in_clusters_v9": "in_clusters_v10", "out_clusters_v9": "out_clusters_v10", "groups_v9": "groups_v10", "group_members_v9": "group_members_v10", "relays_v9": "relays_v10", "attributes_cache_v9": "attributes_cache_v10", "neighbors_v9": "neighbors_v10", "node_descriptors_v9": "node_descriptors_v10", "unsupported_attributes_v9": "unsupported_attributes_v10", } ) async def _migrate_to_v11(self): """Schema v11 added a new `routes_v11` table.""" await self._migrate_tables( { "devices_v10": "devices_v11", "endpoints_v10": "endpoints_v11", "in_clusters_v10": "in_clusters_v11", "out_clusters_v10": "out_clusters_v11", "groups_v10": "groups_v11", "group_members_v10": "group_members_v11", "relays_v10": "relays_v11", "attributes_cache_v10": "attributes_cache_v11", "neighbors_v10": "neighbors_v11", "node_descriptors_v10": "node_descriptors_v11", "unsupported_attributes_v10": "unsupported_attributes_v11", "network_backups_v10": "network_backups_v11", } ) async def _migrate_to_v12(self): """Schema v12 added a `timestamp` column to attribute updates.""" await self._migrate_tables( { "devices_v11": "devices_v12", "endpoints_v11": "endpoints_v12", "in_clusters_v11": "in_clusters_v12", "neighbors_v11": "neighbors_v12", "routes_v11": "routes_v12", "node_descriptors_v11": "node_descriptors_v12", "out_clusters_v11": "out_clusters_v12", "groups_v11": "groups_v12", "group_members_v11": "group_members_v12", "relays_v11": "relays_v12", "unsupported_attributes_v11": "unsupported_attributes_v12", "network_backups_v11": "network_backups_v12", "attributes_cache_v11": None, } ) async with self.execute("SELECT * FROM attributes_cache_v11") as cursor: async for ieee, endpoint_id, cluster_id, attrid, value in cursor: # Set the default `last_updated` to the unix epoch await self.execute( "INSERT INTO attributes_cache_v12 VALUES (?, ?, ?, ?, ?, ?)", (ieee, endpoint_id, cluster_id, attrid, value, 0), ) async def _migrate_to_v13(self): """Schema v13 combines both cluster types and caching for all attributes.""" await self._migrate_tables( { "devices_v12": "devices_v13", "endpoints_v12": "endpoints_v13", "neighbors_v12": "neighbors_v13", "routes_v12": "routes_v13", "node_descriptors_v12": "node_descriptors_v13", "groups_v12": "groups_v13", "group_members_v12": "group_members_v13", "relays_v12": "relays_v13", "network_backups_v12": "network_backups_v13", "in_clusters_v12": None, "out_clusters_v12": None, "unsupported_attributes_v12": None, "attributes_cache_v12": None, } ) async with self.execute("SELECT * FROM in_clusters_v12") as cursor: async for ieee, endpoint_id, cluster_id in cursor: await self.execute( "INSERT INTO clusters_v13 VALUES (?, ?, ?, ?)", (ieee, endpoint_id, ClusterType.Server, cluster_id), ) async with self.execute("SELECT * FROM out_clusters_v12") as cursor: async for ieee, endpoint_id, cluster_id in cursor: await self.execute( "INSERT INTO clusters_v13 VALUES (?, ?, ?, ?)", (ieee, endpoint_id, ClusterType.Client, cluster_id), ) async with self.execute("SELECT * FROM unsupported_attributes_v12") as cursor: async for ieee, endpoint_id, cluster_id, attrid in cursor: await self.execute( "INSERT INTO unsupported_attributes_v13 VALUES (?, ?, ?, ?, ?)", (ieee, endpoint_id, ClusterType.Server, cluster_id, attrid), ) async with self.execute("SELECT * FROM attributes_cache_v12") as cursor: async for ( ieee, endpoint_id, cluster_id, attrid, value, last_updated, ) in cursor: await self.execute( "INSERT INTO attributes_cache_v13 VALUES (?, ?, ?, ?, ?, ?, ?)", ( ieee, endpoint_id, ClusterType.Server, cluster_id, attrid, value, last_updated, ), ) async def _migrate_to_v14(self) -> None: """Schema v14 adds `manufacturer_code` and `status` to the attribute cache.""" await self._migrate_tables( { "devices_v13": "devices_v14", "endpoints_v13": "endpoints_v14", "neighbors_v13": "neighbors_v14", "routes_v13": "routes_v14", "node_descriptors_v13": "node_descriptors_v14", "groups_v13": "groups_v14", "group_members_v13": "group_members_v14", "relays_v13": "relays_v14", "network_backups_v13": "network_backups_v14", "clusters_v13": "clusters_v14", "unsupported_attributes_v13": None, "attributes_cache_v13": None, } ) # Migrate unsupported attributes into the attributes cache with status async with self.execute("SELECT * FROM unsupported_attributes_v13") as cursor: async for ( ieee, endpoint_id, cluster_type, cluster_id, attrid, ) in cursor: await self.execute( "INSERT INTO attributes_cache_v14 (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( ieee, endpoint_id, cluster_type, cluster_id, attrid, UNMIGRATED_MANUFACTURER_CODE, Status.UNSUPPORTED_ATTRIBUTE, None, datetime.fromtimestamp(0, UTC).timestamp(), ), ) async with self.execute("SELECT * FROM attributes_cache_v13") as cursor: async for ( ieee, endpoint_id, cluster_type, cluster_id, attrid, value, last_updated, ) in cursor: # Use INSERT OR IGNORE because the same attribute may exist in both # unsupported_attributes_v13 and attributes_cache_v13. The unsupported # status (inserted first) should win over old cached values. await self.execute( "INSERT OR IGNORE INTO attributes_cache_v14 (ieee, endpoint_id, cluster_type, cluster_id, attr_id, manufacturer_code, status, value, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( ieee, endpoint_id, cluster_type, cluster_id, attrid, UNMIGRATED_MANUFACTURER_CODE, Status.SUCCESS, value, last_updated, ), ) async def _migrate_to_v15(self) -> None: """Schema v15 adds `ota_query_cache` table for persisting OTA query fields.""" await self._migrate_tables( { "devices_v14": "devices_v15", "endpoints_v14": "endpoints_v15", "neighbors_v14": "neighbors_v15", "routes_v14": "routes_v15", "node_descriptors_v14": "node_descriptors_v15", "groups_v14": "groups_v15", "group_members_v14": "group_members_v15", "relays_v14": "relays_v15", "network_backups_v14": "network_backups_v15", "clusters_v14": "clusters_v15", "attributes_cache_v14": "attributes_cache_v15", } ) # ota_query_cache_v15 is new and starts empty