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,

)

from zigpy.zcl.clusters.general import Basic

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 = 14

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)

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._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)

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 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._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: