starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-28 18:24:50 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
input_button_trigger
core/homeassistant/helpers/device_registry.py
T

3377 lines
143 KiB
Python
Raw Permalink Blame History

"""Provide a way to connect entities belonging to one device."""
import asyncio
from collections import defaultdict
from collections.abc import Iterable, Mapping, Set as AbstractSet
import copy
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum
from functools import lru_cache
import logging
import os
import shutil
import time
from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypedDict, Unpack, override
import attr
from yarl import URL
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import (
Event,
HomeAssistant,
ReleaseChannel,
callback,
get_release_channel,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.loader import async_suggest_report_issue
from homeassistant.util import uuid as uuid_util
from homeassistant.util.dt import utc_from_timestamp, utcnow
from homeassistant.util.event_type import EventType
from homeassistant.util.hass_dict import HassKey
from homeassistant.util.json import format_unserializable_data
from . import storage, translation
from .debounce import Debouncer
from .deprecation import deprecated_function
from .frame import (
MissingIntegrationFrame,
ReportBehavior,
get_integration_frame,
report_usage,
)
from .json import JSON_DUMP, find_paths_unserializable_data, json_bytes, json_fragment
from .registry import BaseRegistry, BaseRegistryItems, RegistryIndexType
from .typing import UNDEFINED, UndefinedType
if TYPE_CHECKING:
# mypy cannot workout _cache Protocol with attrs
from propcache.api import cached_property as under_cached_property
from homeassistant.config_entries import ConfigEntry
from . import entity_registry
else:
from propcache.api import under_cached_property
_LOGGER = logging.getLogger(__name__)
DATA_REGISTRY: HassKey[DeviceRegistry] = HassKey("device_registry")
EVENT_DEVICE_REGISTRY_UPDATED: EventType[EventDeviceRegistryUpdatedData] = EventType(
"device_registry_updated"
)
STORAGE_KEY = "core.device_registry"
STORAGE_VERSION_MAJOR = 3
STORAGE_VERSION_MINOR = 2
CLEANUP_DELAY = 10
CONNECTION_BLUETOOTH = "bluetooth"
CONNECTION_NETWORK_MAC = "mac"
CONNECTION_UPNP = "upnp"
CONNECTION_ZIGBEE = "zigbee"
ORPHANED_DEVICE_KEEP_SECONDS = 86400 * 30
# suggested_area can be removed when suggested_area is removed from DeviceEntry.
# pending_move can be removed once add_config_entry_id and remove_config_entry_id
# are removed from the device registry API.
RUNTIME_ONLY_ATTRS = {"suggested_area", "pending_move"}
@dataclass(frozen=True, slots=True)
class _PendingMove:
"""A deferred config-entry move recorded by add_config_entry_id.
A later remove_config_entry_id from the same integration (origin_domain) completes
the move; one from a different integration cancels it. Runtime-only, never stored.
"""
config_entry_id: str
config_subentry_id: str | None
origin_domain: str | None
def _current_integration_domain() -> str | None:
"""Return the domain of the integration in the current call stack, if any."""
try:
return get_integration_frame().integration
except MissingIntegrationFrame:
return None
CONFIGURATION_URL_SCHEMES = {"http", "https", "homeassistant"}
class DeviceEntryDisabler(StrEnum):
"""What disabled a device entry."""
CONFIG_ENTRY = "config_entry"
INTEGRATION = "integration"
USER = "user"
class DeviceInfo(TypedDict, total=False):
"""Entity device information for device registry."""
configuration_url: str | URL | None
connections: set[tuple[str, str]]
created_at: str
default_manufacturer: str
default_model: str
default_name: str
entry_type: DeviceEntryType | None
identifiers: set[tuple[str, str]]
manufacturer: str | None
model: str | None
model_id: str | None
modified_at: str
name: str | None
serial_number: str | None
suggested_area: str | None
sw_version: str | None
hw_version: str | None
translation_key: str | None
translation_placeholders: Mapping[str, str] | None
via_device: tuple[str, str] # Deprecated, use via_device_id instead
via_device_id: str
DEVICE_INFO_TYPES = {
# Device info is categorized by finding the first device info type which has all
# the keys of the device info. The link device info type must be kept first
# to make it preferred over primary.
"link": {
"connections",
"identifiers",
},
"primary": {
"configuration_url",
"connections",
"entry_type",
"hw_version",
"identifiers",
"manufacturer",
"model",
"model_id",
"name",
"serial_number",
"suggested_area",
"sw_version",
"via_device",
"via_device_id",
},
"secondary": {
"connections",
"default_manufacturer",
"default_model",
"default_name",
# Used by Fritz
"via_device",
"via_device_id",
},
}
class _EventDeviceRegistryUpdatedData_Create(TypedDict):
"""EventDeviceRegistryUpdated data for action type 'create'."""
action: Literal["create"]
device_id: str
class _EventDeviceRegistryUpdatedData_Remove(TypedDict):
"""EventDeviceRegistryUpdated data for action type 'remove'."""
action: Literal["remove"]
device_id: str
device: dict[str, Any]
class _EventDeviceRegistryUpdatedData_Update(TypedDict):
"""EventDeviceRegistryUpdated data for action type 'update'."""
action: Literal["update"]
device_id: str
changes: dict[str, Any]
type EventDeviceRegistryUpdatedData = (
_EventDeviceRegistryUpdatedData_Create
| _EventDeviceRegistryUpdatedData_Remove
| _EventDeviceRegistryUpdatedData_Update
)
class DeviceEntryType(StrEnum):
"""Device entry type."""
SERVICE = "service"
class DeviceInfoError(HomeAssistantError):
"""Raised when device info is invalid."""
def __init__(self, domain: str, device_info: DeviceInfo, message: str) -> None:
"""Initialize error."""
super().__init__(
f"Invalid device info {device_info} for '{domain}' config entry: {message}",
)
self.device_info = device_info
self.domain = domain
class DeviceCollisionError(HomeAssistantError):
"""Raised when a device collision is detected."""
class DeviceIdentifierCollisionError(DeviceCollisionError):
"""Raised when a device identifier collision is detected."""
def __init__(
self, identifiers: set[tuple[str, str]], existing_device: DeviceEntry
) -> None:
"""Initialize error."""
super().__init__(
f"Identifiers {identifiers} already registered with {existing_device}"
)
class DeviceConnectionCollisionError(DeviceCollisionError):
"""Raised when a device connection collision is detected."""
def __init__(
self, normalized_connections: set[tuple[str, str]], existing_device: DeviceEntry
) -> None:
"""Initialize error."""
super().__init__(
f"Connections {normalized_connections} "
f"already registered with {existing_device}"
)
def _determine_device_info_type(
config_entry: ConfigEntry,
device_info: DeviceInfo,
) -> str:
"""Determine the type of a device info."""
keys = set(device_info)
# If no keys or not enough info to match up, abort
if not device_info.get("connections") and not device_info.get("identifiers"):
raise DeviceInfoError(
config_entry.domain,
device_info,
"device info must include at least one of identifiers or connections",
)
device_info_type: str | None = None
# Find the first device info type which has all keys in the device info
for possible_type, allowed_keys in DEVICE_INFO_TYPES.items():
if keys <= allowed_keys:
device_info_type = possible_type
break
if device_info_type is None:
raise DeviceInfoError(
config_entry.domain,
device_info,
(
"device info needs to either describe a device, "
"link to existing device or provide extra information."
),
)
return device_info_type
class _ValidatedDeviceInfoFields(TypedDict):
"""Device info fields validated on create and update."""
configuration_url: str | URL | UndefinedType | None
hw_version: str | UndefinedType | None
manufacturer: str | UndefinedType | None
model: str | UndefinedType | None
model_id: str | UndefinedType | None
serial_number: str | UndefinedType | None
sw_version: str | UndefinedType | None
_cached_parse_url = lru_cache(maxsize=512)(URL)
"""Parse a URL and cache the result."""
def _validate_str(name: str, value: Any) -> str | UndefinedType | None:
"""Validate that a device registry string field has correct type."""
if (
value is UNDEFINED
or value is None
or type(value) is str # fast path for exact str
or isinstance(value, str)
):
return value
report_usage(
f"passes a non-string value of type {type(value).__name__} "
f"as {name} to the device registry",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2026.12.0",
)
return str(value)
def _validate_device_info_fields(
**fields: Unpack[_ValidatedDeviceInfoFields],
) -> _ValidatedDeviceInfoFields:
"""Validate device-info field values."""
configuration_url = fields["configuration_url"]
url: URL | None = None
if type(configuration_url) is URL:
url = configuration_url
configuration_url = str(configuration_url)
else:
configuration_url = _validate_str("configuration_url", configuration_url)
if isinstance(configuration_url, str):
url = _cached_parse_url(configuration_url)
if url is not None and (
url.scheme not in CONFIGURATION_URL_SCHEMES or not url.host
):
raise ValueError(f"invalid configuration_url '{configuration_url}'")
return {
"configuration_url": configuration_url,
"hw_version": _validate_str("hw_version", fields["hw_version"]),
"manufacturer": _validate_str("manufacturer", fields["manufacturer"]),
"model": _validate_str("model", fields["model"]),
"model_id": _validate_str("model_id", fields["model_id"]),
"serial_number": _validate_str("serial_number", fields["serial_number"]),
"sw_version": _validate_str("sw_version", fields["sw_version"]),
}
@lru_cache(maxsize=512)
def format_mac(mac: str) -> str:
"""Format the mac address string for entry into dev reg."""
to_test = mac
if len(to_test) == 17 and to_test.count(":") == 5:
return to_test.lower()
if len(to_test) == 17 and to_test.count("-") == 5:
to_test = to_test.replace("-", "")
elif len(to_test) == 14 and to_test.count(".") == 2:
to_test = to_test.replace(".", "")
if len(to_test) == 12:
# no : included
return ":".join(to_test.lower()[i : i + 2] for i in range(0, 12, 2))
# Not sure how formatted, return original
return mac
def _normalize_connections(
connections: Iterable[tuple[str, str]],
) -> set[tuple[str, str]]:
"""Normalize connections to ensure we can match mac addresses."""
return {
(key, format_mac(value)) if key == CONNECTION_NETWORK_MAC else (key, value)
for key, value in connections
}
def _normalize_connections_validator(
instance: Any,
attribute: Any,
connections: Iterable[tuple[str, str]],
) -> None:
"""Check connections normalization used as attrs validator."""
for key, value in connections:
if key == CONNECTION_NETWORK_MAC and format_mac(value) != value:
raise ValueError(f"Invalid mac address format: {value}")
@attr.s(frozen=True, slots=True)
class DeviceEntry:
"""Device Registry Entry."""
config_entry_id: str = attr.ib()
area_id: str | None = attr.ib(default=None)
config_subentry_id: str | None = attr.ib(default=None)
configuration_url: str | None = attr.ib(default=None)
connections: set[tuple[str, str]] = attr.ib(
converter=set, factory=set, validator=_normalize_connections_validator
)
created_at: datetime = attr.ib(factory=utcnow)
disabled_by: DeviceEntryDisabler | None = attr.ib(default=None)
entry_type: DeviceEntryType | None = attr.ib(default=None)
hw_version: str | None = attr.ib(default=None)
id: str = attr.ib(factory=uuid_util.random_uuid_hex)
identifiers: set[tuple[str, str]] = attr.ib(converter=set, factory=set)
labels: set[str] = attr.ib(converter=set, factory=set)
# composite_device_id is the id of the pre-migration composite device this device was
# split from; composite_primary_config_entry is that composite's former
# primary_config_entry, so a restored composite device can report it.
# split_at records when the split happened.
composite_device_id: str | None = attr.ib(default=None)
composite_primary_config_entry: str | None = attr.ib(default=None)
split_at: datetime | None = attr.ib(default=None)
manufacturer: str | None = attr.ib(default=None)
model: str | None = attr.ib(default=None)
model_id: str | None = attr.ib(default=None)
modified_at: datetime = attr.ib(factory=utcnow)
name_by_user: str | None = attr.ib(default=None)
name: str | None = attr.ib(default=None)
# Set on devices created by splitting a pre-migration composite device: the
# identifiers and connections copied from the composite have not yet been reconciled.
# On the owning integration's first re-registration they are replaced with the ones
# it provides and this flag is cleared - a one-shot marker, unlike composite_device_id
# which is kept for the device's lifetime so old ids keep resolving; neither can be
# derived from the other. This flag and the replacement logic can be removed in HA
# Core 2027.8.
has_composite_identifiers: bool = attr.ib(default=False)
serial_number: str | None = attr.ib(default=None)
# Suggested area is deprecated and will be removed from DeviceEntry in HA Core 2026.9.
_suggested_area: str | None = attr.ib(default=None)
sw_version: str | None = attr.ib(default=None)
via_device_id: str | None = attr.ib(default=None)
# Transient pending move target (config_entry_id, config_subentry_id) initiated by
# add_config_entry_id and completed by a subsequent remove_config_entry_id. It is
# never stored and is not part of equality. Can be removed in HA Core 2027.8.
_pending_move: _PendingMove | None = attr.ib(default=None, eq=False)
# Set only on the read-only composite device that async_get synthesizes on demand
# for a pre-migration composite device id. It holds the union of the split
# devices' config entries and subentries so callers see the pre-split device. It is
# never stored and the composite is never added to the registry. Can be removed in
# HA Core 2027.8.
_composite_subentries: dict[str, set[str | None]] | None = attr.ib(
default=None, eq=False
)
_cache: dict[str, Any] = attr.ib(factory=dict, eq=False, init=False)
@property
def config_entries(self) -> set[str]:
"""Return the config entries this device belongs to.
Deprecated compatibility shim: a device now belongs to a single config
entry, available as config_entry_id.
"""
if self._composite_subentries is not None:
return set(self._composite_subentries)
return {self.config_entry_id}
@property
def config_entries_subentries(self) -> dict[str, set[str | None]]:
"""Return the config subentries this device belongs to.
Deprecated compatibility shim: a device now belongs to a single config
entry and subentry, available as config_entry_id and config_subentry_id.
"""
if self._composite_subentries is not None:
return {
entry_id: set(subentries)
for entry_id, subentries in self._composite_subentries.items()
}
return {self.config_entry_id: {self.config_subentry_id}}
@property
def primary_config_entry(self) -> str:
"""Return the primary config entry of this device.
Deprecated compatibility shim: a device now belongs to a single config
entry, available as config_entry_id, which is its primary config entry.
For a restored composite device (synthesized on the fly by async_get for a
pre-migration composite device id), this returns the composite's former
primary_config_entry, which is recorded on the split devices during migration as
composite_primary_config_entry.
"""
return self.config_entry_id
@property
def disabled(self) -> bool:
"""Return if entry is disabled."""
return self.disabled_by is not None
@property
def dict_repr(self) -> dict[str, Any]:
"""Return a dict representation of the entry."""
# Convert sets and tuples to lists
# so the JSON serializer does not have to do
# it every time
return {
"area_id": self.area_id,
"configuration_url": self.configuration_url,
# config_entries and config_entries_subentries are deprecated and kept for
# backwards compatibility, they can be removed in HA Core 2027.8. They use the
# compatibility properties so a restored composite reports its merged entries.
"config_entries": list(self.config_entries),
"config_entries_subentries": {
entry_id: list(subentries)
for entry_id, subentries in self.config_entries_subentries.items()
},
"config_entry_id": self.config_entry_id,
"config_subentry_id": self.config_subentry_id,
"connections": list(self.connections),
"created_at": self.created_at.timestamp(),
"disabled_by": self.disabled_by,
"entry_type": self.entry_type,
"hw_version": self.hw_version,
"id": self.id,
"identifiers": list(self.identifiers),
"labels": list(self.labels),
"manufacturer": self.manufacturer,
"model": self.model,
"model_id": self.model_id,
"modified_at": self.modified_at.timestamp(),
"name_by_user": self.name_by_user,
"name": self.name,
"primary_config_entry": self.primary_config_entry,
"serial_number": self.serial_number,
"sw_version": self.sw_version,
"via_device_id": self.via_device_id,
}
@under_cached_property
def json_repr(self) -> bytes | None:
"""Return a cached JSON representation of the entry."""
try:
dict_repr = self.dict_repr
return json_bytes(dict_repr)
except ValueError, TypeError:
_LOGGER.error(
"Unable to serialize entry %s to JSON. Bad data found at %s",
self.id,
format_unserializable_data(
find_paths_unserializable_data(dict_repr, dump=JSON_DUMP)
),
)
return None
@under_cached_property
def as_storage_fragment(self) -> json_fragment:
"""Return a json fragment for storage."""
return json_fragment(
json_bytes(
{
"area_id": self.area_id,
"config_entry_id": self.config_entry_id,
"config_subentry_id": self.config_subentry_id,
"configuration_url": self.configuration_url,
"connections": list(self.connections),
"created_at": self.created_at,
"disabled_by": self.disabled_by,
"entry_type": self.entry_type,
"hw_version": self.hw_version,
"id": self.id,
"identifiers": list(self.identifiers),
"labels": list(self.labels),
"composite_device_id": self.composite_device_id,
"composite_primary_config_entry": (
self.composite_primary_config_entry
),
"split_at": self.split_at,
"manufacturer": self.manufacturer,
"model": self.model,
"model_id": self.model_id,
"modified_at": self.modified_at,
"name_by_user": self.name_by_user,
"name": self.name,
"has_composite_identifiers": (self.has_composite_identifiers),
"primary_config_entry": self.primary_config_entry,
"serial_number": self.serial_number,
"sw_version": self.sw_version,
"via_device_id": self.via_device_id,
}
)
)
@property
@deprecated_function(
"code which ignores suggested_area", breaks_in_ha_version="2026.9"
)
def suggested_area(self) -> str | None:
"""Return the suggested area for this device entry."""
return self._suggested_area
# async_update_device arguments that redefine which identifiers/connections a device is
# keyed by, or move it to another config entry. They are ambiguous on a synthesized
# composite (there is no single underlying device to retarget), so the composite shim
# drops them with a warning instead of fanning them out. serial_number is intentionally
# NOT here: it describes the physical device and is consistent across a composite's
# splits, so it fans out like sw_version. Can be removed in HA Core 2027.8.
_COMPOSITE_IGNORED_UPDATE_ARGS = (
"merge_connections",
"merge_identifiers",
"new_config_entry_id",
"new_config_subentry_id",
"new_connections",
"new_identifiers",
)
@attr.s(frozen=True, slots=True)
class DeletedDeviceEntry:
"""Deleted Device Registry Entry."""
# config_entry_id is None for orphaned deleted devices, i.e. devices whose owning
# config entry has been removed
config_entry_id: str | None = attr.ib()
config_subentry_id: str | None = attr.ib()
area_id: str | None = attr.ib()
connections: set[tuple[str, str]] = attr.ib(
validator=_normalize_connections_validator
)
created_at: datetime = attr.ib()
disabled_by: DeviceEntryDisabler | UndefinedType | None = attr.ib()
id: str = attr.ib()
identifiers: set[tuple[str, str]] = attr.ib()
labels: set[str] = attr.ib()
modified_at: datetime = attr.ib()
name_by_user: str | None = attr.ib()
orphaned_timestamp: float | None = attr.ib()
# Domain of the config entry that owns (or owned) this device, recorded when the
# device is deleted so a re-added config entry only restores an orphan from the same
# integration. None for legacy stores.
domain: str | None = attr.ib(default=None)
_cache: dict[str, Any] = attr.ib(factory=dict, eq=False, init=False)
@property
def config_entries(self) -> set[str]:
"""Return the config entries this device belonged to.
Deprecated compatibility shim; empty for orphaned deleted devices.
"""
return {self.config_entry_id} if self.config_entry_id is not None else set()
@property
def config_entries_subentries(self) -> dict[str, set[str | None]]:
"""Return the config subentries this device belonged to.
Deprecated compatibility shim; empty for orphaned deleted devices.
"""
if self.config_entry_id is None:
return {}
return {self.config_entry_id: {self.config_subentry_id}}
def to_device_entry(
self,
config_entry: ConfigEntry,
config_subentry_id: str | None,
connections: set[tuple[str, str]],
identifiers: set[tuple[str, str]],
disabled_by: DeviceEntryDisabler | UndefinedType | None,
) -> DeviceEntry:
"""Create DeviceEntry from DeletedDeviceEntry."""
# Adjust disabled_by based on config entry state
if self.disabled_by is not UNDEFINED:
disabled_by = self.disabled_by
if config_entry.disabled_by:
if disabled_by is None:
disabled_by = DeviceEntryDisabler.CONFIG_ENTRY
elif disabled_by == DeviceEntryDisabler.CONFIG_ENTRY:
disabled_by = None
else:
disabled_by = disabled_by if disabled_by is not UNDEFINED else None
return DeviceEntry(
area_id=self.area_id,
config_entry_id=config_entry.entry_id,
config_subentry_id=config_subentry_id,
# type ignores: likely https://github.com/python/mypy/issues/8625
connections=connections, # type: ignore[arg-type]
created_at=self.created_at,
disabled_by=disabled_by,
identifiers=identifiers, # type: ignore[arg-type]
id=self.id,
labels=self.labels, # type: ignore[arg-type]
name_by_user=self.name_by_user,
)
@under_cached_property
def as_storage_fragment(self) -> json_fragment:
"""Return a json fragment for storage."""
return json_fragment(
json_bytes(
{
"area_id": self.area_id,
"config_entry_id": self.config_entry_id,
"config_subentry_id": self.config_subentry_id,
"connections": list(self.connections),
"created_at": self.created_at,
"disabled_by": self.disabled_by
if self.disabled_by is not UNDEFINED
else None,
"disabled_by_undefined": self.disabled_by is UNDEFINED,
"identifiers": list(self.identifiers),
"id": self.id,
"labels": list(self.labels),
"modified_at": self.modified_at,
"name_by_user": self.name_by_user,
"orphaned_timestamp": self.orphaned_timestamp,
"domain": self.domain,
}
)
)
def _copy_if_exists(source: str, destination: str) -> bool:
"""Copy source to destination when source exists (runs in the executor).
Returns whether the file was copied.
"""
if not os.path.isfile(source):
return False
shutil.copyfile(source, destination)
return True
class DeviceRegistryStore(storage.Store[dict[str, list[dict[str, Any]]]]):
"""Store entity registry data."""
@override
async def _async_migrate_func( # noqa: C901
self,
old_major_version: int,
old_minor_version: int,
old_data: dict[str, list[dict[str, Any]]],
) -> dict[str, Any]:
"""Migrate to the new version."""
# Note: There's no version 2, it was planned and supported by previous versions
# of the migrator which treated version 2 like version 1.
if old_major_version < 3:
# Copy the store before the version 3 migrator rewrites every device, so a
# user can recover the pre-migration registry if the migration misbehaves.
await self._async_backup_store()
if old_minor_version < 2:
# Version 1.2 implements migration and freezes the available keys,
# populate keys which were introduced before version 1.2
for device in old_data["devices"]:
device.setdefault("area_id", None)
device.setdefault("configuration_url", None)
device.setdefault("disabled_by", None)
try:
device["entry_type"] = DeviceEntryType(
device.get("entry_type"), # type: ignore[arg-type]
)
except ValueError:
device["entry_type"] = None
device.setdefault("name_by_user", None)
# via_device_id was originally introduced as hub_device_id
device.setdefault("via_device_id", device.get("hub_device_id"))
old_data.setdefault("deleted_devices", [])
for device in old_data["deleted_devices"]:
device.setdefault("orphaned_timestamp", None)
if old_minor_version < 3:
# Version 1.3 adds hw_version
for device in old_data["devices"]:
device["hw_version"] = None
if old_minor_version < 4:
# Introduced in 2023.11
for device in old_data["devices"]:
device["serial_number"] = None
if old_minor_version < 5:
# Introduced in 2024.3
for device in old_data["devices"]:
device["labels"] = []
if old_minor_version < 6:
# Introduced in 2024.7
for device in old_data["devices"]:
device["primary_config_entry"] = None
if old_minor_version < 7:
# Introduced in 2024.8
for device in old_data["devices"]:
device["model_id"] = None
if old_minor_version < 8:
# Introduced in 2024.8
created_at = utc_from_timestamp(0).isoformat()
for device in old_data["devices"]:
device["created_at"] = device["modified_at"] = created_at
for device in old_data["deleted_devices"]:
device["created_at"] = device["modified_at"] = created_at
if old_minor_version < 9:
# Introduced in 2025.2
for device in old_data["devices"]:
device["config_entries_subentries"] = {
config_entry_id: {None}
for config_entry_id in device["config_entries"]
}
for device in old_data["deleted_devices"]:
device["config_entries_subentries"] = {
config_entry_id: {None}
for config_entry_id in device["config_entries"]
}
if old_minor_version < 10:
# Introduced in 2025.6
for device in old_data["deleted_devices"]:
device["area_id"] = None
device["disabled_by"] = None
device["labels"] = []
device["name_by_user"] = None
if old_minor_version < 11:
# Normalization of stored CONNECTION_NETWORK_MAC, introduced in 2025.8
for device in old_data["devices"]:
device["connections"] = _normalize_connections(
device["connections"]
)
for device in old_data["deleted_devices"]:
device["connections"] = _normalize_connections(
device["connections"]
)
if old_minor_version < 12:
# Version 1.12 adds undefined flags to deleted devices, this is a bugfix
# of version 1.10
for device in old_data["deleted_devices"]:
device["disabled_by_undefined"] = old_minor_version < 10
# Version 3 restricts a device to a single config entry and subentry,
# introduced in 2026.8. Composite devices which belonged to several
# config entries (or several subentries of one entry) are split into one
# device per (config entry, subentry). Each split device keeps a copy of
# the identifiers and connections and a reference (composite_device_id) to the original
# composite device id, so that actions targeting the old id still reach
# all split devices. Entities are moved to the matching split device when
# the registries are loaded.
migrated_at = utcnow().isoformat()
devices: list[dict[str, Any]] = []
# Active splits whose copied disabled_by must be reconciled against their
# single config entry once the config entries are loaded
migrated_active_splits: list[dict[str, Any]] = []
for device in old_data["devices"]:
# One target per config entry. config_entries_subentries was a set, so
# the old model allowed a device in several subentries of one config
# entry, but the single-owner model keeps one. Multi-subentry devices
# created by core integrations all come from broken subentry migrators
# (which left a device in both None and its real subentry), so prefer
# a real subentry over the main entry (None). Collapsing rather than
# splitting avoids duplicate devices which, sharing identifiers and
# connections within one config entry, would collide in the
# per-config-entry identifier/connection index.
pairs = [
(
config_entry_id,
next((s for s in subentry_ids if s is not None), None),
)
for config_entry_id, subentry_ids in device[
"config_entries_subentries"
].items()
]
if not pairs:
# Drop devices that have no config entry / subentry pairs
continue
if len(pairs) == 1:
config_entry_id, subentry_id = pairs[0]
device["config_entry_id"] = config_entry_id
device["config_subentry_id"] = subentry_id
device["composite_device_id"] = None
device["composite_primary_config_entry"] = None
device["split_at"] = None
device["has_composite_identifiers"] = False
devices.append(device)
continue
old_id = device["id"]
composite_primary = device.get("primary_config_entry")
for config_entry_id, subentry_id in pairs:
split = copy.deepcopy(device)
split["id"] = uuid_util.random_uuid_hex()
split["config_entry_id"] = config_entry_id
split["config_subentry_id"] = subentry_id
split["primary_config_entry"] = config_entry_id
split["composite_device_id"] = old_id
split["composite_primary_config_entry"] = composite_primary
split["split_at"] = migrated_at
split["has_composite_identifiers"] = True
devices.append(split)
migrated_active_splits.append(split)
old_data["devices"] = devices
# A split inherited the composite's disabled_by, which may not match its
# single config entry (e.g. a split owned by an enabled entry must not stay
# CONFIG_ENTRY disabled). Config entries load concurrently, so wait for them
# and reconcile each split against its own entry.
if migrated_active_splits:
await self.hass.config_entries.async_wait_initialized()
for split in migrated_active_splits:
config_entry = self.hass.config_entries.async_get_entry(
split["config_entry_id"]
)
if config_entry is not None:
_migrate_device_disabled_by(
split, config_entry.disabled_by is not None
)
deleted_devices: list[dict[str, Any]] = []
for device in old_data["deleted_devices"]:
# One target per config entry. config_entries_subentries was a set, so
# the old model allowed a device in several subentries of one config
# entry, but the single-owner model keeps one. Multi-subentry devices
# created by core integrations all come from broken subentry migrators
# (which left a device in both None and its real subentry), so prefer
# a real subentry over the main entry (None). Collapsing rather than
# splitting avoids duplicate devices which, sharing identifiers and
# connections within one config entry, would collide in the
# per-config-entry identifier/connection index.
pairs = [
(
config_entry_id,
next((s for s in subentry_ids if s is not None), None),
)
for config_entry_id, subentry_ids in device[
"config_entries_subentries"
].items()
]
if len(pairs) <= 1:
# Unlike active devices, config_entry_id=None is a valid
# (orphaned) state for a deleted device, so a deleted device with
# no config entries is kept rather than dropped.
config_entry_id, subentry_id = pairs[0] if pairs else (None, None)
device["config_entry_id"] = config_entry_id
device["config_subentry_id"] = subentry_id
device["domain"] = None
deleted_devices.append(device)
continue
# A deleted device that belonged to several config entries or subentries
# is split like an active one - each split keeps a copy of the
# identifiers/connections so every config entry can still restore its
# share when a matching device is re-registered.
for config_entry_id, subentry_id in pairs:
split = copy.deepcopy(device)
split["id"] = uuid_util.random_uuid_hex()
split["config_entry_id"] = config_entry_id
split["config_subentry_id"] = subentry_id
split["domain"] = None
deleted_devices.append(split)
old_data["deleted_devices"] = deleted_devices
# config_entries and config_entries_subentries are deprecated; v3 stores only
# the singular config_entry_id / config_subentry_id (single-entry devices kept
# the old keys, splits copied them via deepcopy).
for migrated in (*devices, *deleted_devices):
migrated.pop("config_entries", None)
migrated.pop("config_entries_subentries", None)
if old_major_version < 3 or (old_major_version == 3 and old_minor_version < 2):
# Version 3.2, introduced in 2026.8, rewrites via_device_id links that do
# not reference a live device. A link to a composite parent split by the
# version 3 migration is remapped to one of the splits; any other stale
# link is detached.
device_ids = {device["id"] for device in old_data["devices"]}
# old composite id -> {config entry id -> split id}
composite_splits: dict[str, dict[str, str]] = {}
for device in old_data["devices"]:
if (composite_id := device["composite_device_id"]) is not None:
composite_splits.setdefault(composite_id, {})[
device["config_entry_id"]
] = device["id"]
def _split_for_via_device(
config_entry_id: str, splits: dict[str, str]
) -> str:
"""Pick the split for via device: same entry, same domain, any."""
if (split_id := splits.get(config_entry_id)) is not None:
return split_id
config_entries = self.hass.config_entries
self_entry = config_entries.async_get_entry(config_entry_id)
if self_entry is not None:
for split_entry_id, split_id in splits.items():
split_entry = config_entries.async_get_entry(split_entry_id)
if (
split_entry is not None
and split_entry.domain == self_entry.domain
):
return split_id
return next(iter(splits.values()))
stale_via_devices = [
device
for device in old_data["devices"]
if device["via_device_id"] is not None
and device["via_device_id"] not in device_ids
]
# The domain rung of the split resolution needs the config entries, which
# load concurrently, so wait for them only when a link must be remapped
if any(
device["via_device_id"] in composite_splits
for device in stale_via_devices
):
await self.hass.config_entries.async_wait_initialized()
for device in stale_via_devices:
if (
splits := composite_splits.get(device["via_device_id"])
) is not None:
device["via_device_id"] = _split_for_via_device(
device["config_entry_id"], splits
)
else:
device["via_device_id"] = None
if old_major_version > 3:
raise NotImplementedError
return old_data
async def _async_backup_store(self) -> None:
"""Copy the store file to a timestamped backup before migrating."""
source = self.path
backup = f"{source}.{utcnow().strftime('%Y%m%d_%H%M%S')}.migration_backup"
try:
copied = await self.hass.async_add_executor_job(
_copy_if_exists, source, backup
)
except OSError as err:
_LOGGER.warning("Could not back up %s before migration: %s", source, err)
else:
if copied:
_LOGGER.info("Backed up %s to %s before migration", source, backup)
class _CollidingKeys(NamedTuple):
"""Identifiers and connections shared with a colliding device."""
identifiers: set[tuple[str, str]]
connections: set[tuple[str, str]]
class DeviceRegistryItems[_EntryTypeT: (DeviceEntry, DeletedDeviceEntry)](
BaseRegistryItems[_EntryTypeT]
):
"""Container for device registry items, maps device id -> entry.
Maintains two additional indexes. An identifier or connection can be shared by
several devices, each belonging to a different config entry, so each index maps a
connection or identifier to the devices that have it, keyed by config entry id:
- (connection_type, connection identifier) -> {config_entry_id: entry}
- (DOMAIN, identifier) -> {config_entry_id: entry}
Registry bugs used to allow duplicate keys within a config entry, so old stores
can hold them. Only the last indexed device occupies the slot (matching historic
lookup behavior); the others are recorded as shadowed until reconciled:
- (config_entry_id, (connection_type, connection identifier)) -> {device_id}
- (config_entry_id, (DOMAIN, identifier)) -> {device_id}
"""
def __init__(self) -> None:
"""Initialize the container."""
super().__init__()
self._connections: dict[tuple[str, str], dict[str | None, _EntryTypeT]] = {}
self._identifiers: dict[tuple[str, str], dict[str | None, _EntryTypeT]] = {}
self._shadowed_connections: dict[
tuple[str | None, tuple[str, str]], set[str]
] = {}
self._shadowed_identifiers: dict[
tuple[str | None, tuple[str, str]], set[str]
] = {}
@override
def _index_entry(self, key: str, entry: _EntryTypeT) -> None:
"""Index an entry."""
for connection in entry.connections:
self._index_key(
connection,
entry,
self._connections,
self._shadowed_connections,
)
for identifier in entry.identifiers:
self._index_key(
identifier,
entry,
self._identifiers,
self._shadowed_identifiers,
)
def _index_key(
self,
key: tuple[str, str],
new_device: _EntryTypeT,
index: dict[tuple[str, str], dict[str | None, _EntryTypeT]],
shadowed_index: dict[tuple[str | None, tuple[str, str]], set[str]],
) -> None:
"""Index one key, recording a displaced device as shadowed."""
by_config_entry = index.setdefault(key, {})
config_entry_id = new_device.config_entry_id
if (
existing := by_config_entry.get(config_entry_id)
) is not None and existing.id != new_device.id:
shadowed_index.setdefault((config_entry_id, key), set()).add(existing.id)
by_config_entry[config_entry_id] = new_device
@override
def _unindex_entry(
self, key: str, replacement_entry: _EntryTypeT | None = None
) -> None:
"""Unindex an entry."""
old_device = self.data[key]
for connection in old_device.connections:
self._unindex_key(
connection,
old_device,
self._connections,
self._shadowed_connections,
)
for identifier in old_device.identifiers:
self._unindex_key(
identifier,
old_device,
self._identifiers,
self._shadowed_identifiers,
)
def _unindex_key(
self,
key: tuple[str, str],
old_device: _EntryTypeT,
index: dict[tuple[str, str], dict[str | None, _EntryTypeT]],
shadowed_index: dict[tuple[str | None, tuple[str, str]], set[str]],
) -> None:
"""Unindex one key, promoting a shadowed device into the slot."""
by_config_entry = index[key]
config_entry_id = old_device.config_entry_id
shadow_key = (config_entry_id, key)
shadowed_ids = shadowed_index.get(shadow_key)
if by_config_entry[config_entry_id] is old_device:
if shadowed_ids:
by_config_entry[config_entry_id] = self.data[shadowed_ids.pop()]
else:
del by_config_entry[config_entry_id]
if not by_config_entry:
del index[key]
else:
# Not the slot holder, so it must be shadowed
assert shadowed_ids is not None
shadowed_ids.remove(old_device.id)
if shadowed_ids is not None and not shadowed_ids:
del shadowed_index[shadow_key]
def get_entry(
self,
identifiers: set[tuple[str, str]] | None = None,
connections: set[tuple[str, str]] | None = None,
*,
config_entry_id: str | UndefinedType | None = UNDEFINED,
) -> _EntryTypeT | None:
"""Get the first entry matching identifiers or connections.
If config_entry_id is given, only an entry belonging to that config entry is
returned. Otherwise the first matching entry from any config entry is returned.
"""
if identifiers:
for identifier in identifiers:
if (by_config_entry := self._identifiers.get(identifier)) is not None:
if config_entry_id is UNDEFINED:
return next(iter(by_config_entry.values()))
if config_entry_id in by_config_entry:
return by_config_entry[config_entry_id]
if not connections:
return None
for connection in _normalize_connections(connections):
if (by_config_entry := self._connections.get(connection)) is not None:
if config_entry_id is UNDEFINED:
return next(iter(by_config_entry.values()))
if config_entry_id in by_config_entry:
return by_config_entry[config_entry_id]
return None
def get_entries(
self,
identifiers: AbstractSet[tuple[str, str]] | None = None,
connections: AbstractSet[tuple[str, str]] | None = None,
*,
config_entry_id: str | None = None,
) -> list[_EntryTypeT]:
"""Get all entries matching identifiers or connections.
Matches across all config entries, or only within one if config_entry_id
is given.
"""
entries: dict[str, _EntryTypeT] = {}
if identifiers:
for identifier in identifiers:
if (by_config_entry := self._identifiers.get(identifier)) is not None:
if config_entry_id is None:
for entry in by_config_entry.values():
entries[entry.id] = entry
elif (scoped := by_config_entry.get(config_entry_id)) is not None:
entries[scoped.id] = scoped
if connections:
for connection in _normalize_connections(connections):
if (by_config_entry := self._connections.get(connection)) is not None:
if config_entry_id is None:
for entry in by_config_entry.values():
entries[entry.id] = entry
elif (scoped := by_config_entry.get(config_entry_id)) is not None:
entries[scoped.id] = scoped
return list(entries.values())
def get_colliding_device_ids(
self,
identifiers: set[tuple[str, str]],
connections: set[tuple[str, str]],
*,
config_entry_id: str,
exclude_device_id: str | None,
) -> dict[str, _CollidingKeys]:
"""Get the ids of other same-config-entry devices holding the given keys.
Returns a map from the id of each colliding device to the identifiers and
connections it shares with the given ones. Includes devices shadowed in the
index. connections must be normalized.
"""
colliding: dict[str, _CollidingKeys] = {}
for identifier in identifiers:
for holder_id in self._holder_device_ids(
identifier,
config_entry_id,
self._identifiers,
self._shadowed_identifiers,
):
if holder_id != exclude_device_id:
colliding.setdefault(
holder_id, _CollidingKeys(set(), set())
).identifiers.add(identifier)
for connection in connections:
for holder_id in self._holder_device_ids(
connection,
config_entry_id,
self._connections,
self._shadowed_connections,
):
if holder_id != exclude_device_id:
colliding.setdefault(
holder_id, _CollidingKeys(set(), set())
).connections.add(connection)
return colliding
def _holder_device_ids(
self,
key: tuple[str, str],
config_entry_id: str,
index: dict[tuple[str, str], dict[str | None, _EntryTypeT]],
shadowed_index: dict[tuple[str | None, tuple[str, str]], set[str]],
) -> list[str]:
"""Get a list of ids of the config entry's devices holding a key."""
holder_device_ids: list[str] = []
if (by_config_entry := index.get(key)) is not None and (
slot_holder := by_config_entry.get(config_entry_id)
) is not None:
holder_device_ids.append(slot_holder.id)
holder_device_ids.extend(shadowed_index.get((config_entry_id, key), ()))
return holder_device_ids
def count_shadowed_keys(self) -> int:
"""Count keys registered to multiple devices of one config entry."""
return sum(
len(device_ids)
for shadowed_index in (
self._shadowed_connections,
self._shadowed_identifiers,
)
for device_ids in shadowed_index.values()
)
class ActiveDeviceRegistryItems(DeviceRegistryItems[DeviceEntry]):
"""Container for active (non-deleted) device registry entries."""
def __init__(self) -> None:
"""Initialize the container.
Maintains four additional indexes:
- area_id -> dict[key, True]
- config_entry_id -> dict[key, True]
- label -> dict[key, True]
- composite_device_id -> dict[key, True]
"""
super().__init__()
self._area_id_index: RegistryIndexType = defaultdict(dict)
self._config_entry_id_index: RegistryIndexType = defaultdict(dict)
self._labels_index: RegistryIndexType = defaultdict(dict)
self._composite_device_id_index: RegistryIndexType = defaultdict(dict)
@override
def _index_entry(self, key: str, entry: DeviceEntry) -> None:
"""Index an entry."""
super()._index_entry(key, entry)
if (area_id := entry.area_id) is not None:
self._area_id_index[area_id][key] = True
for label in entry.labels:
self._labels_index[label][key] = True
self._config_entry_id_index[entry.config_entry_id][key] = True
if entry.composite_device_id is not None:
self._composite_device_id_index[entry.composite_device_id][key] = True
@override
def _unindex_entry(
self, key: str, replacement_entry: DeviceEntry | None = None
) -> None:
"""Unindex an entry."""
entry = self.data[key]
if area_id := entry.area_id:
self._unindex_entry_value(key, area_id, self._area_id_index)
if labels := entry.labels:
for label in labels:
self._unindex_entry_value(key, label, self._labels_index)
self._unindex_entry_value(
key, entry.config_entry_id, self._config_entry_id_index
)
if entry.composite_device_id is not None:
self._unindex_entry_value(
key, entry.composite_device_id, self._composite_device_id_index
)
super()._unindex_entry(key, replacement_entry)
def get_devices_for_area_id(self, area_id: str) -> list[DeviceEntry]:
"""Get devices for area."""
data = self.data
return [data[key] for key in self._area_id_index.get(area_id, ())]
def get_devices_for_label(self, label: str) -> list[DeviceEntry]:
"""Get devices for label."""
data = self.data
return [data[key] for key in self._labels_index.get(label, ())]
def get_devices_for_config_entry_id(
self, config_entry_id: str
) -> list[DeviceEntry]:
"""Get devices for config entry."""
data = self.data
return [
data[key] for key in self._config_entry_id_index.get(config_entry_id, ())
]
def get_devices_for_composite_device_id(
self, composite_device_id: str
) -> list[DeviceEntry]:
"""Get the devices a pre-migration composite device was split into."""
data = self.data
return [
data[key]
for key in self._composite_device_id_index.get(composite_device_id, ())
]
def get_composite_splits(self) -> dict[str, list[DeviceEntry]]:
"""Get the pre-migration composite device ids and the devices split from them."""
data = self.data
return {
composite_device_id: [data[key] for key in keys]
for composite_device_id, keys in self._composite_device_id_index.items()
}
class DeletedDeviceRegistryItems(DeviceRegistryItems[DeletedDeviceEntry]):
"""Container for deleted device registry entries.
A deleted device that still belongs to a config entry is indexed by config entry id in
the base class, like an active device. An orphaned deleted device (its config entry
removed) has no config entry id and would collide with every other orphan in the base
config_entry_id=None slot, so orphans are kept out of the base index and tracked in a
separate index keyed by device id, which is unique so orphans never shadow each other.
Orphans are matched on restore by get_orphaned_entry.
"""
def __init__(self) -> None:
"""Initialize the container."""
super().__init__()
self._orphaned_connections: dict[
tuple[str, str], dict[str, DeletedDeviceEntry]
] = {}
self._orphaned_identifiers: dict[
tuple[str, str], dict[str, DeletedDeviceEntry]
] = {}
@override
def _index_entry(self, key: str, entry: DeletedDeviceEntry) -> None:
"""Index an entry, keeping orphans in the separate id-keyed index."""
if entry.config_entry_id is not None:
super()._index_entry(key, entry)
return
for connection in entry.connections:
self._orphaned_connections.setdefault(connection, {})[entry.id] = entry
for identifier in entry.identifiers:
self._orphaned_identifiers.setdefault(identifier, {})[entry.id] = entry
@override
def _unindex_entry(
self, key: str, replacement_entry: DeletedDeviceEntry | None = None
) -> None:
"""Unindex an entry from the base or the orphan index."""
entry = self.data[key]
if entry.config_entry_id is not None:
super()._unindex_entry(key, replacement_entry)
return
for connection in entry.connections:
if connection in self._orphaned_connections:
del self._orphaned_connections[connection][entry.id]
if not self._orphaned_connections[connection]:
del self._orphaned_connections[connection]
for identifier in entry.identifiers:
if identifier in self._orphaned_identifiers:
del self._orphaned_identifiers[identifier][entry.id]
if not self._orphaned_identifiers[identifier]:
del self._orphaned_identifiers[identifier]
def get_orphaned_entry(
self,
identifiers: set[tuple[str, str]] | None,
connections: set[tuple[str, str]] | None,
domain: str,
) -> DeletedDeviceEntry | None:
"""Return an orphan of the given domain to restore.
Orphans are matched on their recorded domain so a chance identifier or connection
collision doesn't restore another integration's device. A domain-less orphan
(carried over by the migration with no recoverable domain) is left for the
periodic purge rather than restored.
"""
orphans: dict[str, DeletedDeviceEntry] = {}
for identifier in identifiers or ():
orphans.update(self._orphaned_identifiers.get(identifier, {}))
for connection in _normalize_connections(connections or set()):
orphans.update(self._orphaned_connections.get(connection, {}))
for entry in orphans.values():
if entry.domain == domain:
return entry
return None
class DeviceRegistry(BaseRegistry[dict[str, list[dict[str, Any]]]]):
"""Class to hold a registry of devices."""
devices: ActiveDeviceRegistryItems
deleted_devices: DeletedDeviceRegistryItems
_device_data: dict[str, DeviceEntry]
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the device registry."""
self.hass = hass
# Devices registered through async_get_or_create in the current setup session
# of their config entry, keyed by config entry id. A key collision with one of
# these raises; one with a not yet registered device is reconciled.
self._live_device_ids: dict[str, set[str]] = {}
self._loaded_event = asyncio.Event()
self._store = DeviceRegistryStore(
hass,
STORAGE_VERSION_MAJOR,
STORAGE_KEY,
atomic_writes=True,
minor_version=STORAGE_VERSION_MINOR,
serialize_in_event_loop=False,
)
@callback
def async_get(self, device_id: str) -> DeviceEntry | None:
"""Get device.
We retrieve the DeviceEntry from the underlying dict to avoid
the overhead of the UserDict __getitem__.
For a pre-migration composite device id, a read-only composite device
merged from the split devices is returned, so integration code that resolves a
device by id (e.g. in a service handler) keeps working. The composite is
synthesized on demand and never stored, so it stays invisible to enumeration,
identifier search and the frontend device list.
"""
if (device := self._device_data.get(device_id)) is not None:
return device
if split_devices := self.devices.get_devices_for_composite_device_id(device_id):
return self._restore_composite_device(device_id, split_devices)
return None
@callback
def _restore_composite_device(
self, device_id: str, split_devices: list[DeviceEntry]
) -> DeviceEntry:
"""Synthesize a read-only composite device from its split devices."""
composite_subentries: dict[str, set[str | None]] = {}
identifiers: set[tuple[str, str]] = set()
connections: set[tuple[str, str]] = set()
for split_device in split_devices:
composite_subentries.setdefault(split_device.config_entry_id, set()).add(
split_device.config_subentry_id
)
identifiers |= split_device.identifiers
connections |= split_device.connections
# Functional identity (identifiers, connections, serial_number) is consistent
# across splits of the same physical device. Use the split owning the composite's
# former primary config entry as the base, so config_entry_id - and thus
# primary_config_entry - reports the composite's former primary.
primary_config_entry = split_devices[0].composite_primary_config_entry
base = next(
(
split_device
for split_device in split_devices
if split_device.config_entry_id == primary_config_entry
),
split_devices[0],
)
return attr.evolve(
base,
composite_subentries=composite_subentries,
connections=connections, # type: ignore[arg-type]
id=device_id,
identifiers=identifiers, # type: ignore[arg-type]
)
@callback
def async_get_device(
self,
identifiers: set[tuple[str, str]] | None = None,
connections: set[tuple[str, str]] | None = None,
) -> DeviceEntry | None:
"""Check if a device is registered.
Identifiers and connections are unique per config entry. If several config
entries share the looked-up identifier or connection, the match is resolved to a
single device when possible - preferring the device whose config entry domain
matches the looked-up identifier. If the remaining matches are the splits of one
pre-migration composite device, a read-only composite spanning them is returned
(async_update_device and async_remove_device fan it out to the underlying
devices). Otherwise, for independent devices sharing an identifier or connection,
one owned by the calling integration is preferred, falling back to the first
match.
"""
matches = self._async_matching_devices(identifiers, connections)
if len(matches) <= 1:
return matches[0] if matches else None
# If the matches are the splits of one pre-migration composite device, return a
# read-only composite over them, reusing the composite's id so stored references
# (an automation, a fired event, or an entity holding the old device id) keep
# resolving to it as before the split.
composite_device_ids = {match.composite_device_id for match in matches}
if (
len(composite_device_ids) == 1
and (pre_migration_id := next(iter(composite_device_ids))) is not None
):
return self._restore_composite_device(pre_migration_id, matches)
# Otherwise they are independent devices sharing an identifier or connection.
# Prefer one owned by the calling integration so the caller resolves to its own
# device rather than an insertion-order-dependent one; fall back to the first.
if (domain := _current_integration_domain()) is not None and (
device := self._first_device_in_domain(matches, domain)
) is not None:
return device
return matches[0]
@callback
def async_get_device_by_identifier(
self, identifier: tuple[str, str], config_entry_id: str
) -> DeviceEntry | None:
"""Get the device with the identifier, owned by the config entry.
Identifiers are unique within a config entry, so unlike async_get_device
the lookup cannot be ambiguous.
"""
return self.devices.get_entry(
identifiers={identifier}, config_entry_id=config_entry_id
)
@callback
def async_get_device_by_connection(
self, connection: tuple[str, str], config_entry_id: str
) -> DeviceEntry | None:
"""Get the device with the connection, owned by the config entry.
Connections are unique within a config entry, so unlike async_get_device
the lookup cannot be ambiguous.
"""
return self.devices.get_entry(
connections={connection}, config_entry_id=config_entry_id
)
@callback
def async_get_devices(
self,
*,
identifiers: set[tuple[str, str]] | None = None,
connections: set[tuple[str, str]] | None = None,
config_entry_id: str | None = None,
) -> list[DeviceEntry]:
"""Get all devices matching any of the identifiers or connections.
If config_entry_id is given, only devices owned by that config entry are
returned.
"""
return self.devices.get_entries(
identifiers, connections, config_entry_id=config_entry_id
)
def _first_device_in_domain(
self, devices: Iterable[DeviceEntry], domain: str
) -> DeviceEntry | None:
"""Return the first device whose config entry belongs to domain."""
for device in devices:
entry = self.hass.config_entries.async_get_entry(device.config_entry_id)
if entry is not None and entry.domain == domain:
return device
return None
@callback
def _async_matching_devices(
self,
identifiers: AbstractSet[tuple[str, str]] | None,
connections: AbstractSet[tuple[str, str]] | None,
) -> list[DeviceEntry]:
"""Return devices matching the lookup, narrowed by identifier-domain priority."""
matches = self.devices.get_entries(identifiers, connections)
if len(matches) > 1 and identifiers:
domains = {identifier[0] for identifier in identifiers}
preferred = [
device
for device in matches
if (
entry := self.hass.config_entries.async_get_entry(
device.config_entry_id
)
)
and entry.domain in domains
]
if preferred:
return preferred
return matches
@callback
def _async_device_ids_for_composite_device_id(
self, device_id: str
) -> list[str] | None:
"""Return the underlying real device ids if device_id is a composite."""
if device_id in self.devices:
return None
if split_devices := self.devices.get_devices_for_composite_device_id(device_id):
return [split_device.id for split_device in split_devices]
return None
@callback
def async_get_devices_for_composite_device_id(
self, composite_device_id: str
) -> list[DeviceEntry]:
"""Return the devices a composite device id represents.
A composite device id is a pre-migration composite id - a device that belonged to
several config entries, split into one device per config entry, each keeping the
original id as composite_device_id. The underlying live devices are returned so
that actions and entity lookups targeting the composite id still reach all of
them; unmodified integrations keep the pre-rewrite behaviour, where a shared
identifier/connection resolved to a single multi-config-entry device. Returns an
empty list for a device id which is not a composite device id.
"""
return self.devices.get_devices_for_composite_device_id(composite_device_id)
@callback
def async_is_composite_device_id(self, device_id: str) -> bool | None:
"""Return True if device_id is a pre-migration composite device id.
A composite device was split into one device per config entry; the
composite device id no longer refers to a registered device. Returns
False for a registered device id, and None for an unknown id.
"""
if device_id in self.devices:
return False
if self.devices.get_devices_for_composite_device_id(device_id):
return True
return None
@callback
def _resolve_via_device_id(
self, via_device_id: str, config_entry_id: str
) -> str | None:
"""Resolve a via_device_id to the id of a registered device.
The id of a pre-migration composite device is resolved to one of the devices
it was split into - preferring the split owned by config_entry_id, then one
owned by the same domain, then any of them. Returns None for an unknown id.
"""
if via_device_id in self.devices:
return via_device_id
if splits := self.devices.get_devices_for_composite_device_id(via_device_id):
# The composite resolution can be removed in HA Core 2027.8
report_usage(
f"passes the id of a pre-migration composite device {via_device_id} "
"as `via_device_id`; pass the id of a single device instead, e.g. "
"one returned by async_get_device_by_identifier",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2027.8",
)
for split in splits:
if split.config_entry_id == config_entry_id:
return split.id
if (
config_entry := self.hass.config_entries.async_get_entry(
config_entry_id
)
) is not None and (
split_in_domain := self._first_device_in_domain(
splits, config_entry.domain
)
) is not None:
return split_in_domain.id
return splits[0].id
return None
def _substitute_name_placeholders(
self,
domain: str,
name: str,
translation_placeholders: Mapping[str, str],
) -> str:
"""Substitute placeholders in entity name."""
try:
return name.format(**translation_placeholders)
except KeyError as err:
if get_release_channel() is not ReleaseChannel.STABLE:
raise HomeAssistantError(f"Missing placeholder {err}") from err
report_issue = async_suggest_report_issue(
self.hass, integration_domain=domain
)
_LOGGER.warning(
(
"Device from integration %s has translation placeholders '%s' "
"which do not match the name '%s', please %s"
),
domain,
translation_placeholders,
name,
report_issue,
)
return name
@callback
def async_get_or_create(
self,
*,
config_entry_id: str,
config_subentry_id: str | UndefinedType | None = UNDEFINED,
configuration_url: str | URL | UndefinedType | None = UNDEFINED,
connections: set[tuple[str, str]] | UndefinedType | None = UNDEFINED,
created_at: str | datetime | UndefinedType = UNDEFINED, # will be ignored
default_manufacturer: str | UndefinedType | None = UNDEFINED,
default_model: str | UndefinedType | None = UNDEFINED,
default_name: str | UndefinedType | None = UNDEFINED,
# To disable a device if it gets created, does not affect existing devices
disabled_by: DeviceEntryDisabler | UndefinedType | None = UNDEFINED,
entry_type: DeviceEntryType | UndefinedType | None = UNDEFINED,
hw_version: str | UndefinedType | None = UNDEFINED,
identifiers: set[tuple[str, str]] | UndefinedType | None = UNDEFINED,
manufacturer: str | UndefinedType | None = UNDEFINED,
model: str | UndefinedType | None = UNDEFINED,
model_id: str | UndefinedType | None = UNDEFINED,
modified_at: str | datetime | UndefinedType = UNDEFINED, # will be ignored
name: str | UndefinedType | None = UNDEFINED,
serial_number: str | UndefinedType | None = UNDEFINED,
suggested_area: str | UndefinedType | None = UNDEFINED,
sw_version: str | UndefinedType | None = UNDEFINED,
translation_key: str | None = None,
translation_placeholders: Mapping[str, str] | None = None,
# via_device is deprecated and will be removed in HA Core 2027.8, use
# via_device_id instead
via_device: tuple[str, str] | UndefinedType | None = UNDEFINED,
via_device_id: str | UndefinedType | None = UNDEFINED,
) -> DeviceEntry:
"""Get device. Create if it doesn't exist."""
default_manufacturer = _validate_str(
"default_manufacturer", default_manufacturer
)
default_model = _validate_str("default_model", default_model)
validated_fields = _validate_device_info_fields(
configuration_url=configuration_url,
hw_version=hw_version,
manufacturer=manufacturer,
model=model,
model_id=model_id,
serial_number=serial_number,
sw_version=sw_version,
)
config_entry = self.hass.config_entries.async_get_entry(config_entry_id)
if config_entry is None:
raise HomeAssistantError(
f"Can't link device to unknown config entry {config_entry_id}"
)
# Validate before mutating the registry below. `via_device=None` (an explicit
# "no via device") alongside a via_device_id is contradictory, so reject it too.
if via_device is not UNDEFINED and via_device_id is not UNDEFINED:
raise HomeAssistantError(
"Passing both `via_device` and `via_device_id` is not allowed; "
"`via_device` is deprecated, pass `via_device_id` only"
)
if (
config_subentry_id is not UNDEFINED
and config_subentry_id is not None
and config_subentry_id not in config_entry.subentries
):
raise HomeAssistantError(
f"Config entry {config_entry_id} has no subentry {config_subentry_id}"
)
if translation_key:
full_translation_key = (
f"component.{config_entry.domain}.device.{translation_key}.name"
)
translations = translation.async_get_cached_translations(
self.hass, self.hass.config.language, "device", config_entry.domain
)
translated_name = translations.get(full_translation_key, translation_key)
name = self._substitute_name_placeholders(
config_entry.domain, translated_name, translation_placeholders or {}
)
# Reconstruct a DeviceInfo dict from the arguments.
# When we upgrade to Python 3.12, we can change this method to instead
# accept kwargs typed as a DeviceInfo dict (PEP 692)
device_info: DeviceInfo = { # type: ignore[assignment]
key: val
for key, val in (
("connections", connections),
("default_manufacturer", default_manufacturer),
("default_model", default_model),
("default_name", default_name),
("entry_type", entry_type),
("identifiers", identifiers),
("name", name),
("suggested_area", suggested_area),
("via_device", via_device),
("via_device_id", via_device_id),
*validated_fields.items(),
)
if val is not UNDEFINED
}
device_info_type = _determine_device_info_type(config_entry, device_info)
if identifiers is None or identifiers is UNDEFINED:
identifiers = set()
if connections is None or connections is UNDEFINED:
connections = set()
else:
connections = _normalize_connections(connections)
device = self.devices.get_entry(
connections=connections,
identifiers=identifiers,
config_entry_id=config_entry_id,
)
self._async_reconcile_collisions(
device, config_entry, device_info, identifiers, connections
)
if device is not None:
# Reconciliation can update the matched device (e.g. detach its via link)
device = self.devices[device.id]
# Resolved after reconciliation so a removed stale duplicate can't be linked
if via_device_id is not UNDEFINED and via_device_id is not None:
resolved_via_device_id = self._resolve_via_device_id(
via_device_id, config_entry_id
)
if resolved_via_device_id is None:
raise DeviceInfoError(
config_entry.domain,
device_info,
f"via_device_id {via_device_id} is not a registered device id",
)
via_device_id = resolved_via_device_id
is_new = False
if device is None:
is_new = True
deleted_device = self.deleted_devices.get_entry(
connections=connections,
identifiers=identifiers,
config_entry_id=config_entry_id,
)
if deleted_device is None:
# Fall back to an orphan (its owning config entry was removed)
# so re-adding an integration restores the device id, area, labels and name
# rather than create a fresh device. Matching on the recorded domain keeps
# a chance identifier/connection collision from restoring another
# integration's device.
deleted_device = self.deleted_devices.get_orphaned_entry(
identifiers, connections, config_entry.domain
)
if deleted_device is None:
area_id: str | None = None
if (
suggested_area is not None
and suggested_area is not UNDEFINED
and suggested_area != ""
):
# Circular dep
from . import area_registry as ar # noqa: PLC0415
area = ar.async_get(self.hass).async_get_or_create(suggested_area)
area_id = area.id
device = DeviceEntry(
area_id=area_id,
config_entry_id=config_entry_id,
# Interpret not specifying a subentry as None
config_subentry_id=(
config_subentry_id
if config_subentry_id is not UNDEFINED
else None
),
)
else:
self.deleted_devices.pop(deleted_device.id)
device = deleted_device.to_device_entry(
config_entry,
# Interpret not specifying a subentry as None
config_subentry_id if config_subentry_id is not UNDEFINED else None,
connections,
identifiers,
disabled_by,
)
disabled_by = UNDEFINED
self.devices[device.id] = device
# If creating a new device, default to the config entry name
if device_info_type == "primary" and (not name or name is UNDEFINED):
name = config_entry.title
elif (
config_subentry_id is not UNDEFINED
and device.config_subentry_id != config_subentry_id
):
# A device belongs to a single config subentry. Re-registering an existing
# device under a different subentry of the same config entry (e.g. entities
# from several subentries sharing one device_info identity) silently moves
# it. This is deprecated since moves should be explicit via
# async_update_device(new_config_subentry_id=...).
# For now warn and fall through to the move below, but it will raise in HA
# Core 2027.8.
report_usage(
"assigns an existing device to a different config subentry, by calling "
"`async_get_or_create` or by adding entities from several subentries that "
"share a device; this silently moves the device. A device belongs to one "
"subentry - keep a shared device in a single subentry, or move it with "
"`async_update_device`",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2027.8.0",
)
self._async_purge_colliding_deleted_devices(device, identifiers, connections)
if default_manufacturer is not UNDEFINED and device.manufacturer is None:
validated_fields["manufacturer"] = default_manufacturer
if default_model is not UNDEFINED and device.model is None:
validated_fields["model"] = default_model
if default_name is not UNDEFINED and device.name is None:
name = default_name
if via_device is not None and via_device is not UNDEFINED:
# Resolve the deprecated via_device to a device id. The identifier is not
# unique across config entries, so prefer a via device in the same config
# entry, then one from the same integration (domain), falling back to any
# config entry (a via device may legitimately belong to a different config
# entry). This ambiguity is why via_device is deprecated.
via = (
self.devices.get_entry(
identifiers={via_device}, config_entry_id=config_entry_id
)
or self._first_device_in_domain(
self.devices.get_entries(identifiers={via_device}),
config_entry.domain,
)
or self.devices.get_entry(identifiers={via_device})
)
if via is None:
report_usage(
"calls `device_registry.async_get_or_create` referencing a "
f"non existing `via_device` {via_device}, "
f"with device info: {device_info}",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2025.12.0",
)
via_device_id = via.id if via else UNDEFINED
elif via_device is None:
# An explicit `via_device=None` means "no via device" (a via_device_id
# alongside it is rejected above).
via_device_id = None
# On the owning integration's first re-registration of a device created by
# splitting a pre-migration composite device, replace the identifiers and
# connections copied from the composite with the ones the integration provides,
# instead of merging. This block and the has_composite_identifiers flag
# can be removed in HA Core 2027.8.
identifiers_connections: dict[str, Any]
has_composite_identifiers: bool | UndefinedType = UNDEFINED
if device.has_composite_identifiers:
identifiers_connections = {
"new_connections": connections,
"new_identifiers": identifiers,
}
has_composite_identifiers = False
else:
identifiers_connections = {
"merge_connections": connections or UNDEFINED,
"merge_identifiers": identifiers or UNDEFINED,
}
device = self._async_update_device(
device.id,
disabled_by=disabled_by,
entry_type=entry_type,
is_new=is_new,
name=name,
has_composite_identifiers=has_composite_identifiers,
new_config_subentry_id=config_subentry_id,
suggested_area=suggested_area,
via_device_id=via_device_id,
**identifiers_connections,
**validated_fields,
)
# This is safe because _async_update_device will always return a device
# in this use case.
assert device
self._live_device_ids.setdefault(device.config_entry_id, set()).add(device.id)
return device
@callback
def _async_update_device( # noqa: C901
self,
device_id: str,
*,
add_config_entry_id: str | UndefinedType = UNDEFINED,
add_config_subentry_id: str | UndefinedType | None = UNDEFINED,
# Only set when stripping colliding keys from a stale device: its retained
# keys can still be duplicated in other stale devices and must not validate.
allow_collisions: bool = False,
area_id: str | UndefinedType | None = UNDEFINED,
configuration_url: str | URL | UndefinedType | None = UNDEFINED,
disabled_by: DeviceEntryDisabler | UndefinedType | None = UNDEFINED,
entry_type: DeviceEntryType | UndefinedType | None = UNDEFINED,
hw_version: str | UndefinedType | None = UNDEFINED,
is_new: bool = False,
labels: set[str] | UndefinedType = UNDEFINED,
manufacturer: str | UndefinedType | None = UNDEFINED,
merge_connections: set[tuple[str, str]] | UndefinedType = UNDEFINED,
merge_identifiers: set[tuple[str, str]] | UndefinedType = UNDEFINED,
model: str | UndefinedType | None = UNDEFINED,
model_id: str | UndefinedType | None = UNDEFINED,
name_by_user: str | UndefinedType | None = UNDEFINED,
name: str | UndefinedType | None = UNDEFINED,
# has_composite_identifiers can be removed in HA Core 2027.8
has_composite_identifiers: bool | UndefinedType = UNDEFINED,
new_config_entry_id: str | UndefinedType = UNDEFINED,
new_config_subentry_id: str | UndefinedType | None = UNDEFINED,
new_connections: set[tuple[str, str]] | UndefinedType = UNDEFINED,
new_identifiers: set[tuple[str, str]] | UndefinedType = UNDEFINED,
remove_config_entry_id: str | UndefinedType = UNDEFINED,
remove_config_subentry_id: str | UndefinedType | None = UNDEFINED,
serial_number: str | UndefinedType | None = UNDEFINED,
# Can be removed when suggested_area is removed from DeviceEntry
suggested_area: str | UndefinedType | None = UNDEFINED,
sw_version: str | UndefinedType | None = UNDEFINED,
via_device_id: str | UndefinedType | None = UNDEFINED,
) -> DeviceEntry | None:
"""Private update device attributes.
:param add_config_subentry_id: Add the device to a specific
subentry of add_config_entry_id
:param remove_config_subentry_id: Remove the device from a
specific subentry of remove_config_entry_id
"""
old = self.devices[device_id]
new_values: dict[str, Any] = {} # Dict with new key/value pairs
old_values: dict[str, Any] = {} # Dict with old key/value pairs
if add_config_entry_id is not UNDEFINED:
if (
add_config_entry := self.hass.config_entries.async_get_entry(
add_config_entry_id
)
) is None:
raise HomeAssistantError(
f"Can't link device to unknown config entry {add_config_entry_id}"
)
if add_config_subentry_id is not UNDEFINED:
if add_config_entry_id is UNDEFINED:
raise HomeAssistantError(
"Can't add config subentry without specifying config entry"
)
if (
add_config_subentry_id
# mypy says add_config_entry can be None. That's impossible, because we
# raise above if that happens
and add_config_subentry_id not in add_config_entry.subentries # type: ignore[union-attr]
):
raise HomeAssistantError(
f"Config entry {add_config_entry_id} has no"
f" subentry {add_config_subentry_id}"
)
if (
remove_config_subentry_id is not UNDEFINED
and remove_config_entry_id is UNDEFINED
):
raise HomeAssistantError(
"Can't remove config subentry without specifying config entry"
)
if (
new_config_entry_id is not UNDEFINED
and self.hass.config_entries.async_get_entry(new_config_entry_id) is None
):
raise HomeAssistantError(
f"Can't move device to unknown config entry {new_config_entry_id}"
)
if (
new_config_entry_id is not UNDEFINED
or new_config_subentry_id is not UNDEFINED
) and (
add_config_entry_id is not UNDEFINED
or remove_config_entry_id is not UNDEFINED
):
raise HomeAssistantError(
"Can't combine new_config_entry_id or new_config_subentry_id with "
"add_config_entry_id or remove_config_entry_id"
)
if not new_connections and not new_identifiers:
raise HomeAssistantError(
"A device must have at least one of identifiers or connections"
)
if merge_connections is not UNDEFINED and new_connections is not UNDEFINED:
raise HomeAssistantError(
"Cannot define both merge_connections and new_connections"
)
if merge_identifiers is not UNDEFINED and new_identifiers is not UNDEFINED:
raise HomeAssistantError(
"Cannot define both merge_identifiers and new_identifiers"
)
if (
via_device_id is not UNDEFINED
and via_device_id is not None
and via_device_id not in self.devices
and not self.devices.get_devices_for_composite_device_id(via_device_id)
):
raise HomeAssistantError(
f"Can't link device to unknown via device {via_device_id}"
)
# A device belongs to exactly one config entry and subentry:
# - add_config_entry_id (with an optional add_config_subentry_id) records a
# transient pending move to that config entry and subentry; on its own it does
# not move the device. Integrations move a device by adding the new config
# entry and then removing the current one, often in separate calls; the removal
# of the current config entry performs the pending move.
# - remove_config_entry_id on the owning entry performs a pending move if there
# is one, otherwise it removes the device, since it has no other config entry.
# - new_config_entry_id / new_config_subentry_id move the device immediately.
target_config_entry_id: str | UndefinedType = UNDEFINED
target_config_subentry_id: str | UndefinedType | None = UNDEFINED
pending_move: _PendingMove | UndefinedType | None = UNDEFINED
if new_config_entry_id is not UNDEFINED:
target_config_entry_id = new_config_entry_id
target_config_subentry_id = (
new_config_subentry_id
if new_config_subentry_id is not UNDEFINED
else None
)
# An immediate move to a new config entry supersedes a deferred move from an
# earlier add_config_entry_id; clear it so a later removal of the new owner
# deletes the device instead of performing the stale move.
pending_move = None
elif new_config_subentry_id is not UNDEFINED:
target_config_subentry_id = new_config_subentry_id
else:
if add_config_entry_id is not UNDEFINED:
# Adding the config entry (and subentry) the device already belongs to is a
# no-op; recording it as a pending move would make a later removal of that
# sole owner move the device to itself instead of deleting it.
already_owner = add_config_entry_id == old.config_entry_id and (
add_config_subentry_id is UNDEFINED
or add_config_subentry_id == old.config_subentry_id
)
if not already_owner:
pending_move = _PendingMove(
add_config_entry_id,
add_config_subentry_id
if add_config_subentry_id is not UNDEFINED
else None,
_current_integration_domain(),
)
if remove_config_entry_id == old.config_entry_id and (
remove_config_subentry_id is UNDEFINED
or remove_config_subentry_id == old.config_subentry_id
):
move_from_prior_call = pending_move is UNDEFINED
move_target = (
pending_move if pending_move is not UNDEFINED else old._pending_move # noqa: SLF001
)
# A deferred move armed by an earlier add_config_entry_id only completes
# if the integration now removing the owning entry is the one that armed
# it. A removal from a different integration (e.g. device_tracker
# attaching a shared MAC) is unrelated, so cancel the move and delete the
# device instead of silently transferring it. Origins from core/tests are
# undetermined (None) and never cancel.
if (
move_target is not None
and move_from_prior_call
and move_target.origin_domain is not None
and (current_domain := _current_integration_domain()) is not None
and current_domain != move_target.origin_domain
):
move_target = None
if move_target is None:
self.async_remove_device(device_id)
return None
target_config_entry_id = move_target.config_entry_id
target_config_subentry_id = move_target.config_subentry_id
pending_move = None
# A pre-migration composite's splits share identity, so once one split
# completes the move to the target entry the others must not also move
# there and collide; clear their pending moves.
if old.composite_device_id is not None:
for sibling in self.devices.get_devices_for_composite_device_id(
old.composite_device_id
):
if (
sibling.id != device_id
and sibling._pending_move is not None # noqa: SLF001
):
self.devices[sibling.id] = attr.evolve(
sibling, pending_move=None
)
if target_config_subentry_id not in (UNDEFINED, None):
resolved_config_entry_id = (
target_config_entry_id
if target_config_entry_id is not UNDEFINED
else old.config_entry_id
)
resolved_config_entry = self.hass.config_entries.async_get_entry(
resolved_config_entry_id
)
if (
resolved_config_entry is None
or target_config_subentry_id not in resolved_config_entry.subentries
):
raise HomeAssistantError(
f"Config entry {resolved_config_entry_id} has no"
f" subentry {target_config_subentry_id}"
)
if (
target_config_entry_id is not UNDEFINED
and target_config_entry_id != old.config_entry_id
):
new_values["config_entry_id"] = target_config_entry_id
old_values["config_entry_id"] = old.config_entry_id
if (
target_config_subentry_id is not UNDEFINED
and target_config_subentry_id != old.config_subentry_id
):
new_values["config_subentry_id"] = target_config_subentry_id
old_values["config_subentry_id"] = old.config_subentry_id
# pending_move is a transient runtime-only attribute; it is not reported in the
# update event (not added to old_values) and never stored
if pending_move is not UNDEFINED and pending_move != old._pending_move: # noqa: SLF001
new_values["pending_move"] = pending_move
# The config entry owning the device after the update. Identifiers and
# connections are unique per config entry, so they are validated against the
# owning entry, as is the disabled state.
effective_config_entry_id = (
target_config_entry_id
if target_config_entry_id is not UNDEFINED
else old.config_entry_id
)
is_move = effective_config_entry_id != old.config_entry_id
if via_device_id is not UNDEFINED and via_device_id is not None:
# Existence was already validated, so this cannot be None
via_device_id = self._resolve_via_device_id(
via_device_id, effective_config_entry_id
)
added_connections: set[tuple[str, str]] | None = None
added_identifiers: set[tuple[str, str]] | None = None
if merge_connections is not UNDEFINED:
normalized_connections = self._validate_connections(
device_id,
effective_config_entry_id,
merge_connections,
allow_collisions,
)
old_connections = old.connections
if not normalized_connections.issubset(old_connections):
added_connections = normalized_connections
new_values["connections"] = old_connections | normalized_connections
old_values["connections"] = old_connections
if merge_identifiers is not UNDEFINED:
merge_identifiers = self._validate_identifiers(
device_id,
effective_config_entry_id,
merge_identifiers,
allow_collisions,
)
old_identifiers = old.identifiers
if not merge_identifiers.issubset(old_identifiers):
added_identifiers = merge_identifiers
new_values["identifiers"] = old_identifiers | merge_identifiers
old_values["identifiers"] = old_identifiers
if new_connections is not UNDEFINED:
added_connections = new_values["connections"] = self._validate_connections(
device_id, effective_config_entry_id, new_connections, allow_collisions
)
old_values["connections"] = old.connections
if new_identifiers is not UNDEFINED:
added_identifiers = new_values["identifiers"] = self._validate_identifiers(
device_id, effective_config_entry_id, new_identifiers, allow_collisions
)
old_values["identifiers"] = old.identifiers
# On a move to another config entry, validate the identifiers and connections
# retained from the old entry against the new one, so the move can't silently
# overwrite the index slot of a device that already has the same identity there.
# A full new_identifiers / new_connections replacement is validated above;
# merge_* only adds, so the retained old values still need checking here.
if is_move:
if new_identifiers is UNDEFINED:
self._validate_identifiers(
device_id, effective_config_entry_id, old.identifiers, False
)
if new_connections is UNDEFINED:
self._validate_connections(
device_id, effective_config_entry_id, old.connections, False
)
# An explicit disabled_by must be consistent with the disabled state of the
# config entry owning the device after the update: a device can't be enabled
# when the owning config entry is disabled, and can't be disabled by
# CONFIG_ENTRY when the owning config entry is enabled. An inconsistent
# disabled_by is ignored; this will raise in HA Core 2027.8.
# On a move, reflect the new owning config entry's disabled state (as restoring
# a deleted device does) unless a consistent disabled_by was passed explicitly:
# disable an enabled device moved onto a disabled entry, and clear a
# CONFIG_ENTRY disable when moved onto an enabled entry. A USER disable is
# preserved. A new device is reconciled the same way, so a create can't
# leave the device's disabled state contradicting the owning entry's.
if (disabled_by is not UNDEFINED or is_move or is_new) and (
owning_entry := self.hass.config_entries.async_get_entry(
effective_config_entry_id
)
) is not None:
if is_move:
context = "when moving a device to"
elif is_new:
context = "when creating a device attached to"
else:
context = "on a device belonging to"
if disabled_by is None and owning_entry.disabled_by:
report_usage(
f"sets disabled_by to None {context} the disabled "
f"config entry {effective_config_entry_id}",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2027.8",
)
disabled_by = UNDEFINED
elif (
disabled_by is DeviceEntryDisabler.CONFIG_ENTRY
and not owning_entry.disabled_by
):
report_usage(
f"sets disabled_by to DeviceEntryDisabler.CONFIG_ENTRY {context} "
f"the enabled config entry {effective_config_entry_id}",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2027.8",
)
disabled_by = UNDEFINED
if (is_move or is_new) and disabled_by is UNDEFINED:
if owning_entry.disabled_by:
if old.disabled_by is None:
disabled_by = DeviceEntryDisabler.CONFIG_ENTRY
elif old.disabled_by is DeviceEntryDisabler.CONFIG_ENTRY:
disabled_by = None
for attr_name, value in (
("area_id", area_id),
("configuration_url", configuration_url),
("disabled_by", disabled_by),
("entry_type", entry_type),
("hw_version", hw_version),
("labels", labels),
("manufacturer", manufacturer),
("model", model),
("model_id", model_id),
("name", name),
("name_by_user", name_by_user),
("has_composite_identifiers", has_composite_identifiers),
("serial_number", serial_number),
("sw_version", sw_version),
("via_device_id", via_device_id),
):
if value is not UNDEFINED and value != getattr(old, attr_name):
new_values[attr_name] = value
old_values[attr_name] = getattr(old, attr_name)
# Can be removed when suggested_area is removed from DeviceEntry
if suggested_area is not UNDEFINED and suggested_area != old._suggested_area: # noqa: SLF001
new_values["suggested_area"] = suggested_area
old_values["suggested_area"] = old._suggested_area # noqa: SLF001
if not new_values and not is_new:
return old
# This condition can be removed when suggested_area is removed from DeviceEntry
if not RUNTIME_ONLY_ATTRS.issuperset(new_values):
# Change modified_at if we are changing something that we store
new_values["modified_at"] = utcnow()
self.hass.verify_event_loop_thread("device_registry._async_update_device")
new = attr.evolve(old, **new_values)
self.devices[device_id] = new
# On a move, the device's whole retained identity newly appears in the target
# config entry; added_identifiers/added_connections are empty on a retained-
# identity move, so match the target entry's deleted device by the full identity.
match_identifiers: set[tuple[str, str]] | None
match_connections: set[tuple[str, str]] | None
if is_move:
match_identifiers = new.identifiers
match_connections = new.connections
else:
match_identifiers = added_identifiers
match_connections = added_connections
# A deleted device holding an identity the device now owns can never restore
for deleted_device_id in self.deleted_devices.get_colliding_device_ids(
match_identifiers or set(),
match_connections or set(),
config_entry_id=effective_config_entry_id,
exclude_device_id=None,
):
del self.deleted_devices[deleted_device_id]
# If its only run time attributes (suggested_area)
# that do not get saved we do not want to write
# to disk or fire an event as we would end up
# firing events for data we have nothing to compare
# against since its never saved on disk
if RUNTIME_ONLY_ATTRS.issuperset(new_values):
# This can be removed when suggested_area is removed from DeviceEntry
return new
self.async_schedule_save()
data: EventDeviceRegistryUpdatedData
if is_new:
data = {"action": "create", "device_id": new.id}
else:
data = {"action": "update", "device_id": new.id, "changes": old_values}
self.hass.bus.async_fire_internal(EVENT_DEVICE_REGISTRY_UPDATED, data)
return new
@callback
def async_update_device(
self,
device_id: str,
*,
add_config_entry_id: str | UndefinedType = UNDEFINED,
add_config_subentry_id: str | UndefinedType | None = UNDEFINED,
area_id: str | UndefinedType | None = UNDEFINED,
configuration_url: str | URL | UndefinedType | None = UNDEFINED,
disabled_by: DeviceEntryDisabler | UndefinedType | None = UNDEFINED,
entry_type: DeviceEntryType | UndefinedType | None = UNDEFINED,
hw_version: str | UndefinedType | None = UNDEFINED,
labels: set[str] | UndefinedType = UNDEFINED,
manufacturer: str | UndefinedType | None = UNDEFINED,
merge_connections: set[tuple[str, str]] | UndefinedType = UNDEFINED,
merge_identifiers: set[tuple[str, str]] | UndefinedType = UNDEFINED,
model: str | UndefinedType | None = UNDEFINED,
model_id: str | UndefinedType | None = UNDEFINED,
name_by_user: str | UndefinedType | None = UNDEFINED,
name: str | UndefinedType | None = UNDEFINED,
new_config_entry_id: str | UndefinedType = UNDEFINED,
new_config_subentry_id: str | UndefinedType | None = UNDEFINED,
new_connections: set[tuple[str, str]] | UndefinedType = UNDEFINED,
new_identifiers: set[tuple[str, str]] | UndefinedType = UNDEFINED,
remove_config_entry_id: str | UndefinedType = UNDEFINED,
remove_config_subentry_id: str | UndefinedType | None = UNDEFINED,
serial_number: str | UndefinedType | None = UNDEFINED,
# suggested_area is deprecated and will be removed in 2026.9
suggested_area: str | UndefinedType | None = UNDEFINED,
sw_version: str | UndefinedType | None = UNDEFINED,
via_device_id: str | UndefinedType | None = UNDEFINED,
) -> DeviceEntry | None:
"""Update device attributes.
A device belongs to a single config entry and subentry. To move a device to
another config entry or subentry, pass new_config_entry_id and/or
new_config_subentry_id. To remove a device, pass remove_config_entry_id with the
device's config entry.
:param add_config_entry_id: Deprecated. Combined with remove_config_entry_id it
moves the device; on its own it does nothing.
:param add_config_subentry_id: Deprecated. Combined with remove_config_subentry_id
it moves the device to another subentry; on its own it does nothing.
:param disabled_by: Disable or enable the device. Must be consistent with the
disabled state of the config entry owning the device after the update:
a device can't be enabled when the owning config entry is disabled, and
can't be disabled by CONFIG_ENTRY when the owning config entry is enabled.
An inconsistent disabled_by is deprecated and ignored; this will raise in
HA Core 2027.8.
:param new_config_entry_id: Move the device to this config entry. Unless a
disabled_by consistent with the new config entry's disabled state is
passed explicitly, the device's disabled state is updated to reflect the
new config entry's disabled state.
:param new_config_subentry_id: Move the device to this subentry.
:param remove_config_entry_id: Remove the device if it is the device's config
entry, unless combined with add_config_entry_id to move the device.
:param remove_config_subentry_id: Remove the device from a specific subentry of
remove_config_entry_id.
"""
if (
underlying_ids := self._async_device_ids_for_composite_device_id(device_id)
) is not None:
# Fan the update out to each underlying device; keep in sync with the
# update parameters above.
update_args = {
"add_config_entry_id": add_config_entry_id,
"add_config_subentry_id": add_config_subentry_id,
"area_id": area_id,
"configuration_url": configuration_url,
"disabled_by": disabled_by,
"entry_type": entry_type,
"hw_version": hw_version,
"labels": labels,
"manufacturer": manufacturer,
"merge_connections": merge_connections,
"merge_identifiers": merge_identifiers,
"model": model,
"model_id": model_id,
"name_by_user": name_by_user,
"name": name,
"new_config_entry_id": new_config_entry_id,
"new_config_subentry_id": new_config_subentry_id,
"new_connections": new_connections,
"new_identifiers": new_identifiers,
"remove_config_entry_id": remove_config_entry_id,
"remove_config_subentry_id": remove_config_subentry_id,
"serial_number": serial_number,
"suggested_area": suggested_area,
"sw_version": sw_version,
"via_device_id": via_device_id,
}
return self._async_update_composite_device(
device_id, underlying_ids, update_args
)
if suggested_area is not UNDEFINED:
report_usage(
"passes a suggested_area to device_registry.async_update device",
core_behavior=ReportBehavior.LOG,
breaks_in_ha_version="2026.9.0",
)
validated_fields = _validate_device_info_fields(
configuration_url=configuration_url,
hw_version=hw_version,
manufacturer=manufacturer,
model=model,
model_id=model_id,
serial_number=serial_number,
sw_version=sw_version,
)
return self._async_update_device(
device_id,
add_config_entry_id=add_config_entry_id,
add_config_subentry_id=add_config_subentry_id,
area_id=area_id,
disabled_by=disabled_by,
entry_type=entry_type,
labels=labels,
merge_connections=merge_connections,
merge_identifiers=merge_identifiers,
name_by_user=name_by_user,
name=name,
new_config_entry_id=new_config_entry_id,
new_config_subentry_id=new_config_subentry_id,
new_connections=new_connections,
new_identifiers=new_identifiers,
remove_config_entry_id=remove_config_entry_id,
remove_config_subentry_id=remove_config_subentry_id,
suggested_area=suggested_area,
via_device_id=via_device_id,
**validated_fields,
)
@callback
def _async_reconcile_collisions(
self,
matched_device: DeviceEntry | None,
config_entry: ConfigEntry,
device_info: DeviceInfo,
identifiers: set[tuple[str, str]],
connections: set[tuple[str, str]],
) -> None:
"""Resolve device key collisions with the registering device.
Shared keys are stripped from stale duplicates (devices not registered this
setup session); a duplicate left without any keys is removed. A collision
with a device registered this setup session raises.
"""
matched_device_id: str | None = None
if matched_device is not None:
matched_device_id = matched_device.id
if not matched_device.has_composite_identifiers:
identifiers = matched_device.identifiers | identifiers
connections = matched_device.connections | connections
colliding = self.devices.get_colliding_device_ids(
identifiers,
connections,
config_entry_id=config_entry.entry_id,
exclude_device_id=matched_device_id,
)
live_device_ids = self._live_device_ids.get(config_entry.entry_id, ())
for holder_id, (shared_identifiers, shared_connections) in colliding.items():
if holder_id not in live_device_ids:
continue
raise DeviceInfoError(
config_entry.domain,
device_info,
f"identifiers or connections "
f"{sorted(shared_identifiers | shared_connections)} are already "
f"registered for device {holder_id} of the same config entry",
)
for holder_id, (shared_identifiers, shared_connections) in colliding.items():
holder = self.devices[holder_id]
remaining_identifiers = holder.identifiers - shared_identifiers
remaining_connections = holder.connections - shared_connections
if not remaining_identifiers and not remaining_connections:
_LOGGER.debug(
"Removing device %s, its identifiers and connections are all "
"registered by another device of the same config entry",
holder_id,
)
self.async_remove_device(holder_id)
continue
_LOGGER.debug(
"Stripping %s from device %s, registered by another device of the "
"same config entry",
sorted(shared_identifiers | shared_connections),
holder_id,
)
strip_values: dict[str, Any] = {}
if shared_identifiers:
strip_values["new_identifiers"] = remaining_identifiers
if shared_connections:
strip_values["new_connections"] = remaining_connections
self._async_update_device(holder_id, allow_collisions=True, **strip_values)
@callback
def _async_purge_colliding_deleted_devices(
self,
device: DeviceEntry,
identifiers: set[tuple[str, str]],
connections: set[tuple[str, str]],
) -> None:
"""Purge deleted devices with key collisions."""
if not device.has_composite_identifiers:
identifiers = device.identifiers | identifiers
connections = device.connections | connections
colliding = self.deleted_devices.get_colliding_device_ids(
identifiers,
connections,
config_entry_id=device.config_entry_id,
exclude_device_id=None,
)
if not colliding:
return
for deleted_device_id in colliding:
_LOGGER.debug(
"Removing deleted device %s, its identifiers or connections are "
"registered by device %s of the same config entry",
deleted_device_id,
device.id,
)
del self.deleted_devices[deleted_device_id]
self.async_schedule_save()
@callback
def _validate_connections(
self,
device_id: str,
config_entry_id: str,
connections: set[tuple[str, str]],
allow_collisions: bool,
) -> set[tuple[str, str]]:
"""Normalize and validate connections, raise on collision with other devices.
Connections are unique per config entry, so only collisions with other devices
of the same config entry are considered.
"""
normalized_connections = _normalize_connections(connections)
if allow_collisions:
return normalized_connections
for connection in normalized_connections:
# We need to iterate over each connection because if there is a
# conflict, the index will only see the last one and we will not
# be able to tell which one caused the conflict
if (
existing_device := self.devices.get_entry(
connections={connection}, config_entry_id=config_entry_id
)
) and existing_device.id != device_id:
raise DeviceConnectionCollisionError(
normalized_connections, existing_device
)
return normalized_connections
@callback
def _validate_identifiers(
self,
device_id: str,
config_entry_id: str,
identifiers: set[tuple[str, str]],
allow_collisions: bool,
) -> set[tuple[str, str]]:
"""Validate identifiers, raise on collision with other devices.
Identifiers are unique per config entry, so only collisions with other devices
of the same config entry are considered.
"""
if allow_collisions:
return identifiers
for identifier in identifiers:
# We need to iterate over each identifier because if there is a
# conflict, the index will only see the last one and we will not
# be able to tell which one caused the conflict
if (
existing_device := self.devices.get_entry(
identifiers={identifier}, config_entry_id=config_entry_id
)
) and existing_device.id != device_id:
raise DeviceIdentifierCollisionError(identifiers, existing_device)
return identifiers
@callback
def _async_update_composite_device(
self,
composite_id: str,
underlying_ids: list[str],
update_args: dict[str, Any],
) -> DeviceEntry | None:
"""Fan an async_update_device call on a composite out to its real devices."""
forward = {
name: value for name, value in update_args.items() if value is not UNDEFINED
}
if ignored := [
name for name in _COMPOSITE_IGNORED_UPDATE_ARGS if name in forward
]:
# These rewrite a device's functional identity or move it, which is ambiguous
# across the composite's underlying devices; drop them rather than corrupt or
# collide, and report the offending integration.
report_usage(
f"passed {', '.join(ignored)} to device_registry.async_update_device "
"for a composite device that spans several config entries (returned for "
"an ambiguous async_get_device lookup, or "
"resolved from a stored device id of a pre-migration composite); the "
"argument cannot be applied to the merged device and was ignored - "
"target a single device, e.g. one returned by "
"async_entries_for_config_entry",
core_behavior=ReportBehavior.LOG,
)
for name in ignored:
del forward[name]
for underlying_id in underlying_ids:
self.async_update_device(underlying_id, **forward)
remaining = [
self.devices[underlying_id]
for underlying_id in underlying_ids
if underlying_id in self.devices
]
if not remaining:
return None
return self._restore_composite_device(composite_id, remaining)
@callback
def async_remove_device(self, device_id: str) -> None:
"""Remove a device from the device registry."""
if (
underlying_ids := self._async_device_ids_for_composite_device_id(device_id)
) is not None:
for underlying_id in underlying_ids:
self.async_remove_device(underlying_id)
return
self.hass.verify_event_loop_thread("device_registry.async_remove_device")
device = self.devices.pop(device_id)
config_entry = self.hass.config_entries.async_get_entry(device.config_entry_id)
self.deleted_devices[device_id] = DeletedDeviceEntry(
area_id=device.area_id,
config_entry_id=device.config_entry_id,
config_subentry_id=device.config_subentry_id,
connections=device.connections,
created_at=device.created_at,
disabled_by=device.disabled_by,
identifiers=device.identifiers,
id=device.id,
labels=device.labels,
modified_at=utcnow(),
name_by_user=device.name_by_user,
orphaned_timestamp=None,
domain=config_entry.domain if config_entry is not None else None,
)
for other_device in list(self.devices.values()):
if other_device.via_device_id == device_id:
self._async_update_device(other_device.id, via_device_id=None)
self.hass.bus.async_fire_internal(
EVENT_DEVICE_REGISTRY_UPDATED,
_EventDeviceRegistryUpdatedData_Remove(
action="remove", device_id=device_id, device=device.dict_repr
),
)
self.async_schedule_save()
@override
async def _async_load(self) -> None:
"""Load the device registry."""
if self._loaded_event.is_set():
raise RuntimeError("Device registry is already loaded")
async_setup_cleanup(self.hass, self)
data = await self._store.async_load()
devices = ActiveDeviceRegistryItems()
deleted_devices = DeletedDeviceRegistryItems()
if data is not None:
for device in data["devices"]:
devices[device["id"]] = DeviceEntry(
area_id=device["area_id"],
config_entry_id=device["config_entry_id"],
config_subentry_id=device["config_subentry_id"],
configuration_url=device["configuration_url"],
# type ignores (if tuple arg was cast): likely https://github.com/python/mypy/issues/8625
connections={
tuple(conn) # type: ignore[misc]
for conn in device["connections"]
},
created_at=datetime.fromisoformat(device["created_at"]),
disabled_by=(
DeviceEntryDisabler(device["disabled_by"])
if device["disabled_by"]
else None
),
entry_type=(
DeviceEntryType(device["entry_type"])
if device["entry_type"]
else None
),
hw_version=device["hw_version"],
id=device["id"],
identifiers={
tuple(iden) # type: ignore[misc]
for iden in device["identifiers"]
},
labels=set(device["labels"]),
composite_device_id=device["composite_device_id"],
composite_primary_config_entry=device[
"composite_primary_config_entry"
],
split_at=(
datetime.fromisoformat(device["split_at"])
if device["split_at"]
else None
),
manufacturer=device["manufacturer"],
model=device["model"],
model_id=device["model_id"],
modified_at=datetime.fromisoformat(device["modified_at"]),
name_by_user=device["name_by_user"],
name=device["name"],
has_composite_identifiers=device["has_composite_identifiers"],
serial_number=device["serial_number"],
sw_version=device["sw_version"],
via_device_id=device["via_device_id"],
)
# Introduced in 0.111
def get_optional_enum[_EnumT: StrEnum](
cls: type[_EnumT], value: str | None, undefined: bool
) -> _EnumT | UndefinedType | None:
"""Convert string to the passed enum, UNDEFINED or None."""
if undefined:
return UNDEFINED
if value is None:
return None
try:
return cls(value)
except ValueError:
return None
for device in data["deleted_devices"]:
deleted_devices[device["id"]] = DeletedDeviceEntry(
area_id=device["area_id"],
config_entry_id=device["config_entry_id"],
config_subentry_id=device["config_subentry_id"],
connections={tuple(conn) for conn in device["connections"]},
created_at=datetime.fromisoformat(device["created_at"]),
disabled_by=get_optional_enum(
DeviceEntryDisabler,
device["disabled_by"],
device["disabled_by_undefined"],
),
identifiers={tuple(iden) for iden in device["identifiers"]},
id=device["id"],
labels=set(device["labels"]),
modified_at=datetime.fromisoformat(device["modified_at"]),
name_by_user=device["name_by_user"],
orphaned_timestamp=device["orphaned_timestamp"],
domain=device["domain"],
)
if (
shadowed_count := devices.count_shadowed_keys()
+ deleted_devices.count_shadowed_keys()
):
_LOGGER.info(
"Loaded %d identifiers/connections registered to multiple devices of "
"one config entry; they will be reconciled as integrations register "
"their devices",
shadowed_count,
)
self.devices = devices
self.deleted_devices = deleted_devices
self._device_data = devices.data
self._loaded_event.set()
async def async_wait_loaded(self) -> None:
"""Wait until the device registry is fully loaded."""
await self._loaded_event.wait()
@callback
@override
def _data_to_save(self) -> dict[str, Any]:
"""Return data of device registry to store in a file."""
# Create intermediate lists to allow this method to be called from a thread
# other than the event loop.
return {
"devices": [
entry.as_storage_fragment for entry in list(self.devices.values())
],
"deleted_devices": [
entry.as_storage_fragment
for entry in list(self.deleted_devices.values())
],
}
@callback
def _resolve_orphan_domain(
self, config_entry_id: str, domain: str | None
) -> str | None:
"""Return the domain to record on devices orphaned from a config entry."""
if domain is not None:
return domain
if (
entry := self.hass.config_entries.async_get_entry(config_entry_id)
) is not None:
return entry.domain
return None
@callback
def _async_orphan_deleted_device(
self, deleted_device: DeletedDeviceEntry, domain: str | None, now_time: float
) -> None:
"""Mark a deleted device as orphaned, remembering its former domain."""
if domain is not None:
# Orphans are indexed by their recorded domain, so two orphans of the
# same domain sharing an identifier or connection would collide. When a
# device from the same integration is orphaned, drop any existing orphan
# it overlaps so the newest one wins deterministically instead of shadowing
# it.
for existing in list(self.deleted_devices.values()):
if (
existing.config_entry_id is None
and existing.domain == domain
and (
existing.connections & deleted_device.connections
or existing.identifiers & deleted_device.identifiers
)
):
del self.deleted_devices[existing.id]
self.deleted_devices[deleted_device.id] = attr.evolve(
deleted_device,
config_entry_id=None,
config_subentry_id=None,
orphaned_timestamp=now_time,
domain=domain,
)
self.async_schedule_save()
@callback
def async_config_entry_unloaded(self, config_entry_id: str) -> None:
"""Forget the live devices of a config entry that unloaded or failed setup."""
self._live_device_ids.pop(config_entry_id, None)
@callback
def async_clear_config_entry(
self, config_entry_id: str, domain: str | None = None
) -> None:
"""Clear config entry from registry entries."""
self._live_device_ids.pop(config_entry_id, None)
domain = self._resolve_orphan_domain(config_entry_id, domain)
now_time = time.time()
for device in self.devices.get_devices_for_config_entry_id(config_entry_id):
self.async_remove_device(device.id)
# A split device records the composite's former primary config entry; when that
# config entry is removed, clear the now-dangling reference so a restored
# composite no longer points at a config entry that no longer exists.
for device in list(self.devices.values()):
if device.composite_primary_config_entry == config_entry_id:
self.devices[device.id] = attr.evolve(
device, composite_primary_config_entry=None
)
self.async_schedule_save()
# A device owned by another config entry may hold a transient pending move
# targeting the entry being removed; clear it so a later completion deletes the
# device instead of moving it onto the removed entry.
for device in list(self.devices.values()):
pending_move = device._pending_move # noqa: SLF001
if (
pending_move is not None
and pending_move.config_entry_id == config_entry_id
):
self.devices[device.id] = attr.evolve(device, pending_move=None)
for deleted_device in list(self.deleted_devices.values()):
if deleted_device.config_entry_id != config_entry_id:
continue
self._async_orphan_deleted_device(deleted_device, domain, now_time)
@callback
def async_clear_config_subentry(
self, config_entry_id: str, config_subentry_id: str, domain: str | None = None
) -> None:
"""Clear config subentry from registry entries."""
domain = self._resolve_orphan_domain(config_entry_id, domain)
now_time = time.time()
for device in self.devices.get_devices_for_config_entry_id(config_entry_id):
if device.config_subentry_id != config_subentry_id:
continue
self.async_remove_device(device.id)
# A device may hold a transient pending move targeting the subentry being removed;
# clear it so a later completion deletes the device instead of validating against
# the removed subentry.
for device in list(self.devices.values()):
pending_move = device._pending_move # noqa: SLF001
if (
pending_move is not None
and pending_move.config_entry_id == config_entry_id
and pending_move.config_subentry_id == config_subentry_id
):
self.devices[device.id] = attr.evolve(device, pending_move=None)
for deleted_device in list(self.deleted_devices.values()):
if (
deleted_device.config_entry_id != config_entry_id
or deleted_device.config_subentry_id != config_subentry_id
):
continue
self._async_orphan_deleted_device(deleted_device, domain, now_time)
@callback
def async_purge_expired_orphaned_devices(self) -> None:
"""Purge expired orphaned devices from the registry.
We need to purge these periodically to avoid the database
growing without bound.
"""
now_time = time.time()
for deleted_device in list(self.deleted_devices.values()):
if deleted_device.orphaned_timestamp is None:
continue
if (
deleted_device.orphaned_timestamp + ORPHANED_DEVICE_KEEP_SECONDS
< now_time
):
del self.deleted_devices[deleted_device.id]
@callback
def async_clear_area_id(self, area_id: str) -> None:
"""Clear area id from registry entries."""
for device in self.devices.get_devices_for_area_id(area_id):
self._async_update_device(device.id, area_id=None)
for deleted_device in list(self.deleted_devices.values()):
if deleted_device.area_id != area_id:
continue
self.deleted_devices[deleted_device.id] = attr.evolve(
deleted_device, area_id=None
)
self.async_schedule_save()
@callback
def async_clear_label_id(self, label_id: str) -> None:
"""Clear label from registry entries."""
for device in self.devices.get_devices_for_label(label_id):
self._async_update_device(device.id, labels=device.labels - {label_id})
for deleted_device in list(self.deleted_devices.values()):
if label_id not in deleted_device.labels:
continue
self.deleted_devices[deleted_device.id] = attr.evolve(
deleted_device, labels=deleted_device.labels - {label_id}
)
self.async_schedule_save()
@callback
def async_get(hass: HomeAssistant) -> DeviceRegistry:
"""Get device registry."""
try:
return hass.data[DATA_REGISTRY]
except KeyError as ex:
raise RuntimeError("Device registry not set up") from ex
@callback
def async_get_device_id_by_identifier(
hass: HomeAssistant, identifier: tuple[str, str], *, config_entry_id: str
) -> str:
"""Get the id of the device with the identifier, owned by the config entry.
Convenience wrapper for linking a device to its via device through
via_device_id. Identifiers are unique within a config entry, so the lookup
cannot be ambiguous.
Raises ValueError if no such device exists.
"""
device = async_get(hass).async_get_device_by_identifier(identifier, config_entry_id)
if device is None:
raise ValueError(
f"There is no device with identifier {identifier} in config entry "
f"{config_entry_id}"
)
return device.id
def async_setup(hass: HomeAssistant) -> None:
"""Set up device registry."""
if DATA_REGISTRY in hass.data:
raise RuntimeError("Device registry is already set up")
hass.data[DATA_REGISTRY] = DeviceRegistry(hass)
async def async_load(hass: HomeAssistant, *, load_empty: bool = False) -> None:
"""Load device registry."""
await async_get(hass).async_load(load_empty=load_empty)
@callback
def async_entries_for_area(registry: DeviceRegistry, area_id: str) -> list[DeviceEntry]:
"""Return entries that match an area."""
return registry.devices.get_devices_for_area_id(area_id)
@callback
def async_entries_for_label(
registry: DeviceRegistry, label_id: str
) -> list[DeviceEntry]:
"""Return entries that match a label."""
return registry.devices.get_devices_for_label(label_id)
@callback
def async_entries_for_config_entry(
registry: DeviceRegistry, config_entry_id: str
) -> list[DeviceEntry]:
"""Return entries that match a config entry."""
return registry.devices.get_devices_for_config_entry_id(config_entry_id)
@callback
def async_config_entry_disabled_by_changed(
registry: DeviceRegistry, config_entry: ConfigEntry
) -> None:
"""Handle a config entry being disabled or enabled.
Disable devices in the registry that are associated with a config entry when
the config entry is disabled, enable devices in the registry that are associated
with a config entry when the config entry is enabled and the devices are marked
DeviceEntryDisabler.CONFIG_ENTRY.
"""
devices = async_entries_for_config_entry(registry, config_entry.entry_id)
if not config_entry.disabled_by:
for device in devices:
if device.disabled_by is not DeviceEntryDisabler.CONFIG_ENTRY:
continue
registry._async_update_device(device.id, disabled_by=None) # noqa: SLF001
return
for device in devices:
if device.disabled:
# Device already disabled, do not overwrite
continue
registry._async_update_device( # noqa: SLF001
device.id, disabled_by=DeviceEntryDisabler.CONFIG_ENTRY
)
@callback
def _migrate_device_disabled_by(
device: dict[str, Any], config_entry_disabled: bool
) -> None:
"""Reconcile a stored device's disabled_by with its config entry's disabled state.
Reimplements async_config_entry_disabled_by_changed on stored data so the 1.13
migration can fix a split device that inherited the composite's disabled_by. Kept in
lockstep with that function by test_migrate_device_disabled_by_matches_runtime; can be
removed in HA Core 2027.8.
"""
disabled_by = device["disabled_by"]
if not config_entry_disabled:
# Config entry enabled: drop a config-entry disable, keep a user/integration one
if disabled_by == DeviceEntryDisabler.CONFIG_ENTRY:
device["disabled_by"] = None
return
# Config entry disabled: disable the device unless it is already disabled
if disabled_by is None:
device["disabled_by"] = DeviceEntryDisabler.CONFIG_ENTRY
@callback
def async_cleanup(
hass: HomeAssistant,
dev_reg: DeviceRegistry,
ent_reg: entity_registry.EntityRegistry,
) -> None:
"""Clean up device registry."""
# Find all devices that are referenced by a config_entry.
config_entry_ids = set(hass.config_entries.async_entry_ids())
references_config_entries = {
device.id
for device in dev_reg.devices.values()
if device.config_entry_id in config_entry_ids
}
# Find all devices that are referenced in the entity registry.
device_ids_referenced_by_entities = set(ent_reg.entities.get_device_ids())
orphan = (
set(dev_reg.devices)
- device_ids_referenced_by_entities
- references_config_entries
)
for dev_id in orphan:
dev_reg.async_remove_device(dev_id)
# Find all referenced config entries that no longer exist
# This shouldn't happen but have not been able to track down the bug :(
for device in list(dev_reg.devices.values()):
if device.config_entry_id not in config_entry_ids:
dev_reg._async_update_device( # noqa: SLF001
device.id, remove_config_entry_id=device.config_entry_id
)
# Periodic purge of orphaned devices to avoid the registry
# growing without bounds when there are lots of deleted devices
dev_reg.async_purge_expired_orphaned_devices()
@callback
def async_setup_cleanup(hass: HomeAssistant, dev_reg: DeviceRegistry) -> None:
"""Clean up device registry when entities removed."""
from . import entity_registry, label_registry as lr # noqa: PLC0415
@callback
def _label_removed_from_registry_filter(
event_data: lr.EventLabelRegistryUpdatedData,
) -> bool:
"""Filter all except for the remove action from label registry events."""
return event_data["action"] == "remove"
@callback
def _handle_label_registry_update(event: lr.EventLabelRegistryUpdated) -> None:
"""Update devices that have a label that has been removed."""
dev_reg.async_clear_label_id(event.data["label_id"])
hass.bus.async_listen(
event_type=lr.EVENT_LABEL_REGISTRY_UPDATED,
event_filter=_label_removed_from_registry_filter,
listener=_handle_label_registry_update,
)
@callback
def _async_cleanup() -> None:
"""Cleanup."""
ent_reg = entity_registry.async_get(hass)
async_cleanup(hass, dev_reg, ent_reg)
debounced_cleanup: Debouncer[None] = Debouncer(
hass, _LOGGER, cooldown=CLEANUP_DELAY, immediate=False, function=_async_cleanup
)
@callback
def _async_entity_registry_changed(
event: Event[entity_registry.EventEntityRegistryUpdatedData],
) -> None:
"""Handle entity updated or removed dispatch."""
debounced_cleanup.async_schedule_call()
@callback
def entity_registry_changed_filter(
event_data: entity_registry.EventEntityRegistryUpdatedData,
) -> bool:
"""Handle entity updated or removed filter."""
if (
event_data["action"] == "update"
and "device_id" not in event_data["changes"]
) or event_data["action"] == "create":
return False
return True
def _async_listen_for_cleanup() -> None:
"""Listen for entity registry changes."""
hass.bus.async_listen(
entity_registry.EVENT_ENTITY_REGISTRY_UPDATED,
_async_entity_registry_changed,
event_filter=entity_registry_changed_filter,
)
if hass.is_running:
_async_listen_for_cleanup()
return
async def startup_clean(event: Event) -> None:
"""Clean up on startup."""
_async_listen_for_cleanup()
await debounced_cleanup.async_call()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, startup_clean)
@callback
def _on_homeassistant_stop(event: Event) -> None:
"""Cancel debounced cleanup."""
debounced_cleanup.async_cancel()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _on_homeassistant_stop)
Reference in New Issue View Git Blame Copy Permalink