Skip to content
Fetching contributors…
Cannot retrieve contributors at this time
166 lines (121 sloc) 4.6 KB
"""Provide a way to connect devices to one physical location."""
from asyncio import Event
from collections import OrderedDict
import logging
from typing import Iterable, MutableMapping, Optional, cast
import uuid
import attr
from homeassistant.core import callback
from homeassistant.loader import bind_hass
from .typing import HomeAssistantType
_LOGGER = logging.getLogger(__name__)
DATA_REGISTRY = "area_registry"
EVENT_AREA_REGISTRY_UPDATED = "area_registry_updated"
STORAGE_KEY = "core.area_registry"
@attr.s(slots=True, frozen=True)
class AreaEntry:
"""Area Registry Entry."""
name = attr.ib(type=str, default=None)
id = attr.ib(type=str, default=attr.Factory(lambda: uuid.uuid4().hex))
class AreaRegistry:
"""Class to hold a registry of areas."""
def __init__(self, hass: HomeAssistantType) -> None:
"""Initialize the area registry."""
self.hass = hass
self.areas: MutableMapping[str, AreaEntry] = {}
self._store =, STORAGE_KEY)
def async_get_area(self, area_id: str) -> Optional[AreaEntry]:
"""Get all areas."""
return self.areas.get(area_id)
def async_list_areas(self) -> Iterable[AreaEntry]:
"""Get all areas."""
return self.areas.values()
def async_create(self, name: str) -> AreaEntry:
"""Create a new area."""
if self._async_is_registered(name):
raise ValueError("Name is already in use")
area = AreaEntry()
self.areas[] = area
created = self._async_update(, name=name)
EVENT_AREA_REGISTRY_UPDATED, {"action": "create", "area_id":}
return created
async def async_delete(self, area_id: str) -> None:
"""Delete area."""
device_registry = await self.hass.helpers.device_registry.async_get_registry()
del self.areas[area_id]
EVENT_AREA_REGISTRY_UPDATED, {"action": "remove", "area_id": area_id}
def async_update(self, area_id: str, name: str) -> AreaEntry:
"""Update name of area."""
updated = self._async_update(area_id, name)
EVENT_AREA_REGISTRY_UPDATED, {"action": "update", "area_id": area_id}
return updated
def _async_update(self, area_id: str, name: str) -> AreaEntry:
"""Update name of area."""
old = self.areas[area_id]
changes = {}
if name ==
return old
if self._async_is_registered(name):
raise ValueError("Name is already in use")
changes["name"] = name
new = self.areas[area_id] = attr.evolve(old, **changes)
return new
def _async_is_registered(self, name: str) -> Optional[AreaEntry]:
"""Check if a name is currently registered."""
for area in self.areas.values():
if name ==
return area
return None
async def async_load(self) -> None:
"""Load the area registry."""
data = await self._store.async_load()
areas: MutableMapping[str, AreaEntry] = OrderedDict()
if data is not None:
for area in data["areas"]:
areas[area["id"]] = AreaEntry(name=area["name"], id=area["id"])
self.areas = areas
def async_schedule_save(self) -> None:
"""Schedule saving the area registry."""
self._store.async_delay_save(self._data_to_save, SAVE_DELAY)
def _data_to_save(self) -> dict:
"""Return data of area registry to store in a file."""
data = {}
data["areas"] = [
{"name":, "id":} for entry in self.areas.values()
return data
async def async_get_registry(hass: HomeAssistantType) -> AreaRegistry:
"""Return area registry instance."""
reg_or_evt =
if not reg_or_evt:
evt =[DATA_REGISTRY] = Event()
reg = AreaRegistry(hass)
await reg.async_load()[DATA_REGISTRY] = reg
return reg
if isinstance(reg_or_evt, Event):
evt = reg_or_evt
await evt.wait()
return cast(AreaRegistry,
return cast(AreaRegistry, reg_or_evt)
You can’t perform that action at this time.