Verisure: Remove JSONPath, unique IDs, small cleanups (#47870)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/47899/head
Franck Nijhof 2021-03-14 10:38:09 +01:00 committed by GitHub
parent 984f02882b
commit 60838cf7ed
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 287 additions and 290 deletions

View File

@ -1,30 +1,27 @@
"""Support for Verisure devices.""" """Support for Verisure devices."""
from __future__ import annotations from __future__ import annotations
from datetime import timedelta from verisure import Error as VerisureError
from typing import Any, Literal
from jsonpath import jsonpath
from verisure import (
Error as VerisureError,
ResponseError as VerisureResponseError,
Session as Verisure,
)
import voluptuous as vol import voluptuous as vol
from homeassistant.components.alarm_control_panel import (
DOMAIN as ALARM_CONTROL_PANEL_DOMAIN,
)
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import ( from homeassistant.const import (
CONF_PASSWORD, CONF_PASSWORD,
CONF_SCAN_INTERVAL, CONF_SCAN_INTERVAL,
CONF_USERNAME, CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP, EVENT_HOMEASSISTANT_STOP,
HTTP_SERVICE_UNAVAILABLE,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util import Throttle
from .const import ( from .const import (
ATTR_DEVICE_SERIAL, ATTR_DEVICE_SERIAL,
@ -47,14 +44,15 @@ from .const import (
SERVICE_DISABLE_AUTOLOCK, SERVICE_DISABLE_AUTOLOCK,
SERVICE_ENABLE_AUTOLOCK, SERVICE_ENABLE_AUTOLOCK,
) )
from .coordinator import VerisureDataUpdateCoordinator
PLATFORMS = [ PLATFORMS = [
"sensor", ALARM_CONTROL_PANEL_DOMAIN,
"switch", BINARY_SENSOR_DOMAIN,
"alarm_control_panel", CAMERA_DOMAIN,
"lock", LOCK_DOMAIN,
"camera", SENSOR_DOMAIN,
"binary_sensor", SWITCH_DOMAIN,
] ]
CONFIG_SCHEMA = vol.Schema( CONFIG_SCHEMA = vol.Schema(
@ -88,18 +86,13 @@ DEVICE_SERIAL_SCHEMA = vol.Schema({vol.Required(ATTR_DEVICE_SERIAL): cv.string})
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Verisure integration.""" """Set up the Verisure integration."""
verisure = Verisure(config[DOMAIN][CONF_USERNAME], config[DOMAIN][CONF_PASSWORD]) coordinator = VerisureDataUpdateCoordinator(hass, config=config[DOMAIN])
coordinator = VerisureDataUpdateCoordinator(
hass, session=verisure, domain_config=config[DOMAIN]
)
if not await hass.async_add_executor_job(coordinator.login): if not await coordinator.async_login():
LOGGER.error("Login failed") LOGGER.error("Login failed")
return False return False
hass.bus.async_listen_once( hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, coordinator.async_logout)
EVENT_HOMEASSISTANT_STOP, lambda event: coordinator.logout()
)
await coordinator.async_refresh() await coordinator.async_refresh()
if not coordinator.last_update_success: if not coordinator.last_update_success:
@ -152,95 +145,3 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
DOMAIN, SERVICE_ENABLE_AUTOLOCK, enable_autolock, schema=DEVICE_SERIAL_SCHEMA DOMAIN, SERVICE_ENABLE_AUTOLOCK, enable_autolock, schema=DEVICE_SERIAL_SCHEMA
) )
return True return True
class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
"""A Verisure Data Update Coordinator."""
def __init__(
self, hass: HomeAssistant, domain_config: ConfigType, session: Verisure
) -> None:
"""Initialize the Verisure hub."""
self.imageseries = {}
self.config = domain_config
self.giid = domain_config.get(CONF_GIID)
self.session = session
super().__init__(
hass, LOGGER, name=DOMAIN, update_interval=domain_config[CONF_SCAN_INTERVAL]
)
def login(self) -> bool:
"""Login to Verisure."""
try:
self.session.login()
except VerisureError as ex:
LOGGER.error("Could not log in to verisure, %s", ex)
return False
if self.giid:
return self.set_giid()
return True
def logout(self) -> bool:
"""Logout from Verisure."""
try:
self.session.logout()
except VerisureError as ex:
LOGGER.error("Could not log out from verisure, %s", ex)
return False
return True
def set_giid(self) -> bool:
"""Set installation GIID."""
try:
self.session.set_giid(self.giid)
except VerisureError as ex:
LOGGER.error("Could not set installation GIID, %s", ex)
return False
return True
async def _async_update_data(self) -> dict:
"""Fetch data from Verisure."""
try:
return await self.hass.async_add_executor_job(self.session.get_overview)
except VerisureResponseError as ex:
LOGGER.error("Could not read overview, %s", ex)
if ex.status_code == HTTP_SERVICE_UNAVAILABLE: # Service unavailable
LOGGER.info("Trying to log in again")
await self.hass.async_add_executor_job(self.login)
return {}
raise
@Throttle(timedelta(seconds=60))
def update_smartcam_imageseries(self) -> None:
"""Update the image series."""
self.imageseries = self.session.get_camera_imageseries()
@Throttle(timedelta(seconds=30))
def smartcam_capture(self, device_id: str) -> None:
"""Capture a new image from a smartcam."""
self.session.capture_image(device_id)
def disable_autolock(self, device_id: str) -> None:
"""Disable autolock."""
self.session.set_lock_config(device_id, auto_lock_enabled=False)
def enable_autolock(self, device_id: str) -> None:
"""Enable autolock."""
self.session.set_lock_config(device_id, auto_lock_enabled=True)
def get(self, jpath: str, *args) -> list[Any] | Literal[False]:
"""Get values from the overview that matches the jsonpath."""
res = jsonpath(self.data, jpath % args)
return res or []
def get_first(self, jpath: str, *args) -> Any | None:
"""Get first value from the overview that matches the jsonpath."""
res = self.get(jpath, *args)
return res[0] if res else None
def get_image_info(self, jpath: str, *args) -> list[Any] | Literal[False]:
"""Get values from the imageseries that matches the jsonpath."""
res = jsonpath(self.imageseries, jpath % args)
return res or []

View File

@ -22,8 +22,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VerisureDataUpdateCoordinator
from .const import CONF_ALARM, CONF_GIID, DOMAIN, LOGGER from .const import CONF_ALARM, CONF_GIID, DOMAIN, LOGGER
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
@ -56,19 +56,19 @@ class VerisureAlarm(CoordinatorEntity, AlarmControlPanelEntity):
giid = self.coordinator.config.get(CONF_GIID) giid = self.coordinator.config.get(CONF_GIID)
if giid is not None: if giid is not None:
aliass = { aliass = {
i["giid"]: i["alias"] for i in self.coordinator.session.installations i["giid"]: i["alias"] for i in self.coordinator.verisure.installations
} }
if giid in aliass: if giid in aliass:
return "{} alarm".format(aliass[giid]) return "{} alarm".format(aliass[giid])
LOGGER.error("Verisure installation giid not found: %s", giid) LOGGER.error("Verisure installation giid not found: %s", giid)
return "{} alarm".format(self.coordinator.session.installations[0]["alias"]) return "{} alarm".format(self.coordinator.verisure.installations[0]["alias"])
@property @property
def state(self) -> str | None: def state(self) -> str | None:
"""Return the state of the device.""" """Return the state of the device."""
status = self.coordinator.get_first("$.armState.statusType") status = self.coordinator.data["alarm"]["statusType"]
if status == "DISARMED": if status == "DISARMED":
self._state = STATE_ALARM_DISARMED self._state = STATE_ALARM_DISARMED
elif status == "ARMED_HOME": elif status == "ARMED_HOME":
@ -95,19 +95,19 @@ class VerisureAlarm(CoordinatorEntity, AlarmControlPanelEntity):
@property @property
def changed_by(self) -> str | None: def changed_by(self) -> str | None:
"""Return the last change triggered by.""" """Return the last change triggered by."""
return self.coordinator.get_first("$.armState.name") return self.coordinator.data["alarm"]["name"]
async def _async_set_arm_state(self, state: str, code: str | None = None) -> None: async def _async_set_arm_state(self, state: str, code: str | None = None) -> None:
"""Send set arm state command.""" """Send set arm state command."""
arm_state = await self.hass.async_add_executor_job( arm_state = await self.hass.async_add_executor_job(
self.coordinator.session.set_arm_state, code, state self.coordinator.verisure.set_arm_state, code, state
) )
LOGGER.debug("Verisure set arm state %s", state) LOGGER.debug("Verisure set arm state %s", state)
transaction = {} transaction = {}
while "result" not in transaction: while "result" not in transaction:
await asyncio.sleep(0.5) await asyncio.sleep(0.5)
transaction = await self.hass.async_add_executor_job( transaction = await self.hass.async_add_executor_job(
self.coordinator.session.get_arm_state_transaction, self.coordinator.verisure.get_arm_state_transaction,
arm_state["armStateChangeTransactionId"], arm_state["armStateChangeTransactionId"],
) )

View File

@ -5,33 +5,32 @@ from typing import Any, Callable
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
DEVICE_CLASS_CONNECTIVITY, DEVICE_CLASS_CONNECTIVITY,
DEVICE_CLASS_OPENING,
BinarySensorEntity, BinarySensorEntity,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CONF_DOOR_WINDOW, DOMAIN, VerisureDataUpdateCoordinator from . import CONF_DOOR_WINDOW, DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any], config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None], add_entities: Callable[[list[CoordinatorEntity]], None],
discovery_info: dict[str, Any] | None = None, discovery_info: dict[str, Any] | None = None,
) -> None: ) -> None:
"""Set up the Verisure binary sensors.""" """Set up the Verisure binary sensors."""
coordinator = hass.data[DOMAIN] coordinator = hass.data[DOMAIN]
sensors = [VerisureEthernetStatus(coordinator)] sensors: list[CoordinatorEntity] = [VerisureEthernetStatus(coordinator)]
if int(coordinator.config.get(CONF_DOOR_WINDOW, 1)): if int(coordinator.config.get(CONF_DOOR_WINDOW, 1)):
sensors.extend( sensors.extend(
[ [
VerisureDoorWindowSensor(coordinator, device_label) VerisureDoorWindowSensor(coordinator, serial_number)
for device_label in coordinator.get( for serial_number in coordinator.data["door_window"]
"$.doorWindow.doorWindowDevice[*].deviceLabel"
)
] ]
) )
@ -44,40 +43,40 @@ class VerisureDoorWindowSensor(CoordinatorEntity, BinarySensorEntity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_label: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the Verisure door window sensor.""" """Initialize the Verisure door window sensor."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the binary sensor.""" """Return the name of the binary sensor."""
return self.coordinator.get_first( return self.coordinator.data["door_window"][self.serial_number]["area"]
"$.doorWindow.doorWindowDevice[?(@.deviceLabel=='%s')].area",
self._device_label, @property
) def unique_id(self) -> str:
"""Return the unique ID for this alarm control panel."""
return f"{self.serial_number}_door_window"
@property
def device_class(self) -> str:
"""Return the class of this device, from component DEVICE_CLASSES."""
return DEVICE_CLASS_OPENING
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return the state of the sensor.""" """Return the state of the sensor."""
return ( return (
self.coordinator.get_first( self.coordinator.data["door_window"][self.serial_number]["state"] == "OPEN"
"$.doorWindow.doorWindowDevice[?(@.deviceLabel=='%s')].state",
self._device_label,
)
== "OPEN"
) )
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available
"$.doorWindow.doorWindowDevice[?(@.deviceLabel=='%s')]", and self.serial_number in self.coordinator.data["door_window"]
self._device_label,
)
is not None
) )
@ -94,12 +93,12 @@ class VerisureEthernetStatus(CoordinatorEntity, BinarySensorEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self.coordinator.get_first("$.ethernetConnectedNow") return self.coordinator.data["ethernet"]
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self.coordinator.get_first("$.ethernetConnectedNow") is not None return super().available and self.coordinator.data["ethernet"] is not None
@property @property
def device_class(self) -> str: def device_class(self) -> str:

View File

@ -8,33 +8,28 @@ from typing import Any, Callable
from homeassistant.components.camera import Camera from homeassistant.components.camera import Camera
from homeassistant.const import EVENT_HOMEASSISTANT_STOP from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VerisureDataUpdateCoordinator
from .const import CONF_SMARTCAM, DOMAIN, LOGGER from .const import CONF_SMARTCAM, DOMAIN, LOGGER
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any], config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None], add_entities: Callable[[list[VerisureSmartcam]], None],
discovery_info: dict[str, Any] | None = None, discovery_info: dict[str, Any] | None = None,
) -> None: ) -> None:
"""Set up the Verisure Camera.""" """Set up the Verisure Camera."""
coordinator = hass.data[DOMAIN] coordinator: VerisureDataUpdateCoordinator = hass.data[DOMAIN]
if not int(coordinator.config.get(CONF_SMARTCAM, 1)): if not int(coordinator.config.get(CONF_SMARTCAM, 1)):
return return
directory_path = hass.config.config_dir assert hass.config.config_dir
if not os.access(directory_path, os.R_OK):
LOGGER.error("file path %s is not readable", directory_path)
return
add_entities( add_entities(
[ [
VerisureSmartcam(hass, coordinator, device_label, directory_path) VerisureSmartcam(hass, coordinator, serial_number, hass.config.config_dir)
for device_label in coordinator.get("$.customerImageCameras[*].deviceLabel") for serial_number in coordinator.data["cameras"]
] ]
) )
@ -48,13 +43,13 @@ class VerisureSmartcam(CoordinatorEntity, Camera):
self, self,
hass: HomeAssistant, hass: HomeAssistant,
coordinator: VerisureDataUpdateCoordinator, coordinator: VerisureDataUpdateCoordinator,
device_label: str, serial_number: str,
directory_path: str, directory_path: str,
): ):
"""Initialize Verisure File Camera component.""" """Initialize Verisure File Camera component."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
self._directory_path = directory_path self._directory_path = directory_path
self._image = None self._image = None
self._image_id = None self._image_id = None
@ -73,21 +68,27 @@ class VerisureSmartcam(CoordinatorEntity, Camera):
def check_imagelist(self) -> None: def check_imagelist(self) -> None:
"""Check the contents of the image list.""" """Check the contents of the image list."""
self.coordinator.update_smartcam_imageseries() self.coordinator.update_smartcam_imageseries()
image_ids = self.coordinator.get_image_info(
"$.imageSeries[?(@.deviceLabel=='%s')].image[0].imageId", self._device_label images = self.coordinator.imageseries.get("imageSeries", [])
) new_image_id = None
if not image_ids: for image in images:
if image["deviceLabel"] == self.serial_number:
new_image_id = image["image"][0]["imageId"]
break
if not new_image_id:
return return
new_image_id = image_ids[0]
if new_image_id in ("-1", self._image_id): if new_image_id in ("-1", self._image_id):
LOGGER.debug("The image is the same, or loading image_id") LOGGER.debug("The image is the same, or loading image_id")
return return
LOGGER.debug("Download new image %s", new_image_id) LOGGER.debug("Download new image %s", new_image_id)
new_image_path = os.path.join( new_image_path = os.path.join(
self._directory_path, "{}{}".format(new_image_id, ".jpg") self._directory_path, "{}{}".format(new_image_id, ".jpg")
) )
self.coordinator.session.download_image( self.coordinator.verisure.download_image(
self._device_label, new_image_id, new_image_path self.serial_number, new_image_id, new_image_path
) )
LOGGER.debug("Old image_id=%s", self._image_id) LOGGER.debug("Old image_id=%s", self._image_id)
self.delete_image() self.delete_image()
@ -110,6 +111,9 @@ class VerisureSmartcam(CoordinatorEntity, Camera):
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of this camera.""" """Return the name of this camera."""
return self.coordinator.get_first( return self.coordinator.data["cameras"][self.serial_number]["area"]
"$.customerImageCameras[?(@.deviceLabel=='%s')].area", self._device_label
) @property
def unique_id(self) -> str:
"""Return the unique ID for this camera."""
return self.serial_number

View File

@ -0,0 +1,126 @@
"""DataUpdateCoordinator for the Verisure integration."""
from __future__ import annotations
from datetime import timedelta
from verisure import (
Error as VerisureError,
ResponseError as VerisureResponseError,
Session as Verisure,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, HTTP_SERVICE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util import Throttle
from .const import CONF_GIID, DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
"""A Verisure Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
"""Initialize the Verisure hub."""
self.imageseries = {}
self.config = config
self.giid = config.get(CONF_GIID)
self.verisure = Verisure(
username=config[CONF_USERNAME], password=config[CONF_PASSWORD]
)
super().__init__(
hass, LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
)
async def async_login(self) -> bool:
"""Login to Verisure."""
try:
await self.hass.async_add_executor_job(self.verisure.login)
except VerisureError as ex:
LOGGER.error("Could not log in to verisure, %s", ex)
return False
if self.giid:
return await self.async_set_giid()
return True
async def async_logout(self) -> bool:
"""Logout from Verisure."""
try:
await self.hass.async_add_executor_job(self.verisure.logout)
except VerisureError as ex:
LOGGER.error("Could not log out from verisure, %s", ex)
return False
return True
async def async_set_giid(self) -> bool:
"""Set installation GIID."""
try:
await self.hass.async_add_executor_job(self.verisure.set_giid, self.giid)
except VerisureError as ex:
LOGGER.error("Could not set installation GIID, %s", ex)
return False
return True
async def _async_update_data(self) -> dict:
"""Fetch data from Verisure."""
try:
overview = await self.hass.async_add_executor_job(
self.verisure.get_overview
)
except VerisureResponseError as ex:
LOGGER.error("Could not read overview, %s", ex)
if ex.status_code == HTTP_SERVICE_UNAVAILABLE: # Service unavailable
LOGGER.info("Trying to log in again")
await self.async_login()
return {}
raise
# Store data in a way Home Assistant can easily consume it
return {
"alarm": overview["armState"],
"ethernet": overview.get("ethernetConnectedNow"),
"cameras": {
device["deviceLabel"]: device
for device in overview["customerImageCameras"]
},
"climate": {
device["deviceLabel"]: device for device in overview["climateValues"]
},
"door_window": {
device["deviceLabel"]: device
for device in overview["doorWindow"]["doorWindowDevice"]
},
"locks": {
device["deviceLabel"]: device
for device in overview["doorLockStatusList"]
},
"mice": {
device["deviceLabel"]: device
for device in overview["eventCounts"]
if device["deviceType"] == "MOUSE1"
},
"smart_plugs": {
device["deviceLabel"]: device for device in overview["smartPlugs"]
},
}
@Throttle(timedelta(seconds=60))
def update_smartcam_imageseries(self) -> None:
"""Update the image series."""
self.imageseries = self.verisure.get_camera_imageseries()
@Throttle(timedelta(seconds=30))
def smartcam_capture(self, device_id: str) -> None:
"""Capture a new image from a smartcam."""
self.verisure.capture_image(device_id)
def disable_autolock(self, device_id: str) -> None:
"""Disable autolock."""
self.verisure.set_lock_config(device_id, auto_lock_enabled=False)
def enable_autolock(self, device_id: str) -> None:
"""Enable autolock."""
self.verisure.set_lock_config(device_id, auto_lock_enabled=True)

View File

@ -7,17 +7,16 @@ from typing import Any, Callable
from homeassistant.components.lock import LockEntity from homeassistant.components.lock import LockEntity
from homeassistant.const import ATTR_CODE, STATE_LOCKED, STATE_UNLOCKED from homeassistant.const import ATTR_CODE, STATE_LOCKED, STATE_UNLOCKED
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VerisureDataUpdateCoordinator
from .const import CONF_CODE_DIGITS, CONF_DEFAULT_LOCK_CODE, CONF_LOCKS, DOMAIN, LOGGER from .const import CONF_CODE_DIGITS, CONF_DEFAULT_LOCK_CODE, CONF_LOCKS, DOMAIN, LOGGER
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any], config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None], add_entities: Callable[[list[VerisureDoorlock]], None],
discovery_info: dict[str, Any] | None = None, discovery_info: dict[str, Any] | None = None,
) -> None: ) -> None:
"""Set up the Verisure lock platform.""" """Set up the Verisure lock platform."""
@ -26,10 +25,8 @@ def setup_platform(
if int(coordinator.config.get(CONF_LOCKS, 1)): if int(coordinator.config.get(CONF_LOCKS, 1)):
locks.extend( locks.extend(
[ [
VerisureDoorlock(coordinator, device_label) VerisureDoorlock(coordinator, serial_number)
for device_label in coordinator.get( for serial_number in coordinator.data["locks"]
"$.doorLockStatusList[*].deviceLabel"
)
] ]
) )
@ -42,11 +39,11 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_label: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the Verisure lock.""" """Initialize the Verisure lock."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
self._state = None self._state = None
self._digits = coordinator.config.get(CONF_CODE_DIGITS) self._digits = coordinator.config.get(CONF_CODE_DIGITS)
self._default_lock_code = coordinator.config.get(CONF_DEFAULT_LOCK_CODE) self._default_lock_code = coordinator.config.get(CONF_DEFAULT_LOCK_CODE)
@ -54,27 +51,19 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the lock.""" """Return the name of the lock."""
return self.coordinator.get_first( return self.coordinator.data["locks"][self.serial_number]["area"]
"$.doorLockStatusList[?(@.deviceLabel=='%s')].area", self._device_label
)
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available and self.serial_number in self.coordinator.data["locks"]
"$.doorLockStatusList[?(@.deviceLabel=='%s')]", self._device_label
)
is not None
) )
@property @property
def changed_by(self) -> str | None: def changed_by(self) -> str | None:
"""Last change triggered by.""" """Last change triggered by."""
return self.coordinator.get_first( return self.coordinator.data["locks"][self.serial_number].get("userString")
"$.doorLockStatusList[?(@.deviceLabel=='%s')].userString",
self._device_label,
)
@property @property
def code_format(self) -> str: def code_format(self) -> str:
@ -84,11 +73,10 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
@property @property
def is_locked(self) -> bool: def is_locked(self) -> bool:
"""Return true if lock is locked.""" """Return true if lock is locked."""
status = self.coordinator.get_first( return (
"$.doorLockStatusList[?(@.deviceLabel=='%s')].lockedState", self.coordinator.data["locks"][self.serial_number]["lockedState"]
self._device_label, == "LOCKED"
) )
return status == "LOCKED"
async def async_unlock(self, **kwargs) -> None: async def async_unlock(self, **kwargs) -> None:
"""Send unlock command.""" """Send unlock command."""
@ -112,9 +100,9 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
"""Send set lock state command.""" """Send set lock state command."""
target_state = "lock" if state == STATE_LOCKED else "unlock" target_state = "lock" if state == STATE_LOCKED else "unlock"
lock_state = await self.hass.async_add_executor_job( lock_state = await self.hass.async_add_executor_job(
self.coordinator.session.set_lock_state, self.coordinator.verisure.set_lock_state,
code, code,
self._device_label, self.serial_number,
target_state, target_state,
) )
@ -123,7 +111,7 @@ class VerisureDoorlock(CoordinatorEntity, LockEntity):
attempts = 0 attempts = 0
while "result" not in transaction: while "result" not in transaction:
transaction = await self.hass.async_add_executor_job( transaction = await self.hass.async_add_executor_job(
self.coordinator.session.get_lock_state_transaction, self.coordinator.verisure.get_lock_state_transaction,
lock_state["doorLockStateChangeTransactionId"], lock_state["doorLockStateChangeTransactionId"],
) )
attempts += 1 attempts += 1

View File

@ -2,6 +2,6 @@
"domain": "verisure", "domain": "verisure",
"name": "Verisure", "name": "Verisure",
"documentation": "https://www.home-assistant.io/integrations/verisure", "documentation": "https://www.home-assistant.io/integrations/verisure",
"requirements": ["jsonpath==0.82", "vsure==1.7.2"], "requirements": ["vsure==1.7.2"],
"codeowners": ["@frenck"] "codeowners": ["@frenck"]
} }

View File

@ -8,47 +8,43 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VerisureDataUpdateCoordinator
from .const import CONF_HYDROMETERS, CONF_MOUSE, CONF_THERMOMETERS, DOMAIN from .const import CONF_HYDROMETERS, CONF_MOUSE, CONF_THERMOMETERS, DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any], config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None], add_entities: Callable[[list[CoordinatorEntity], bool], None],
discovery_info: dict[str, Any] | None = None, discovery_info: dict[str, Any] | None = None,
) -> None: ) -> None:
"""Set up the Verisure platform.""" """Set up the Verisure platform."""
coordinator = hass.data[DOMAIN] coordinator = hass.data[DOMAIN]
sensors = [] sensors: list[CoordinatorEntity] = []
if int(coordinator.config.get(CONF_THERMOMETERS, 1)): if int(coordinator.config.get(CONF_THERMOMETERS, 1)):
sensors.extend( sensors.extend(
[ [
VerisureThermometer(coordinator, device_label) VerisureThermometer(coordinator, serial_number)
for device_label in coordinator.get( for serial_number, values in coordinator.data["climate"].items()
"$.climateValues[?(@.temperature)].deviceLabel" if "temperature" in values
)
] ]
) )
if int(coordinator.config.get(CONF_HYDROMETERS, 1)): if int(coordinator.config.get(CONF_HYDROMETERS, 1)):
sensors.extend( sensors.extend(
[ [
VerisureHygrometer(coordinator, device_label) VerisureHygrometer(coordinator, serial_number)
for device_label in coordinator.get( for serial_number, values in coordinator.data["climate"].items()
"$.climateValues[?(@.humidity)].deviceLabel" if "humidity" in values
)
] ]
) )
if int(coordinator.config.get(CONF_MOUSE, 1)): if int(coordinator.config.get(CONF_MOUSE, 1)):
sensors.extend( sensors.extend(
[ [
VerisureMouseDetection(coordinator, device_label) VerisureMouseDetection(coordinator, serial_number)
for device_label in coordinator.get( for serial_number in coordinator.data["mice"]
"$.eventCounts[?(@.deviceType=='MOUSE1')].deviceLabel"
)
] ]
) )
@ -61,38 +57,35 @@ class VerisureThermometer(CoordinatorEntity, Entity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_label: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the device.""" """Return the name of the entity."""
return ( name = self.coordinator.data["climate"][self.serial_number]["deviceArea"]
self.coordinator.get_first( return f"{name} Temperature"
"$.climateValues[?(@.deviceLabel=='%s')].deviceArea", self._device_label
) @property
+ " temperature" def unique_id(self) -> str:
) """Return the unique ID for this entity."""
return f"{self.serial_number}_temperature"
@property @property
def state(self) -> str | None: def state(self) -> str | None:
"""Return the state of the device.""" """Return the state of the entity."""
return self.coordinator.get_first( return self.coordinator.data["climate"][self.serial_number]["temperature"]
"$.climateValues[?(@.deviceLabel=='%s')].temperature", self._device_label
)
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available
"$.climateValues[?(@.deviceLabel=='%s')].temperature", and self.serial_number in self.coordinator.data["climate"]
self._device_label, and "temperature" in self.coordinator.data["climate"][self.serial_number]
)
is not None
) )
@property @property
@ -107,37 +100,35 @@ class VerisureHygrometer(CoordinatorEntity, Entity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_label: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the device.""" """Return the name of the entity."""
return ( name = self.coordinator.data["climate"][self.serial_number]["deviceArea"]
self.coordinator.get_first( return f"{name} Humidity"
"$.climateValues[?(@.deviceLabel=='%s')].deviceArea", self._device_label
) @property
+ " humidity" def unique_id(self) -> str:
) """Return the unique ID for this entity."""
return f"{self.serial_number}_humidity"
@property @property
def state(self) -> str | None: def state(self) -> str | None:
"""Return the state of the device.""" """Return the state of the entity."""
return self.coordinator.get_first( return self.coordinator.data["climate"][self.serial_number]["humidity"]
"$.climateValues[?(@.deviceLabel=='%s')].humidity", self._device_label
)
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available
"$.climateValues[?(@.deviceLabel=='%s')].humidity", self._device_label and self.serial_number in self.coordinator.data["climate"]
) and "humidity" in self.coordinator.data["climate"][self.serial_number]
is not None
) )
@property @property
@ -152,37 +143,35 @@ class VerisureMouseDetection(CoordinatorEntity, Entity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_label: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_label self.serial_number = serial_number
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the device.""" """Return the name of the entity."""
return ( name = self.coordinator.data["mice"][self.serial_number]["area"]
self.coordinator.get_first( return f"{name} Mouse"
"$.eventCounts[?(@.deviceLabel=='%s')].area", self._device_label
) @property
+ " mouse" def unique_id(self) -> str:
) """Return the unique ID for this entity."""
return f"{self.serial_number}_mice"
@property @property
def state(self) -> str | None: def state(self) -> str | None:
"""Return the state of the device.""" """Return the state of the device."""
return self.coordinator.get_first( return self.coordinator.data["mice"][self.serial_number]["detections"]
"$.eventCounts[?(@.deviceLabel=='%s')].detections", self._device_label
)
@property @property
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available
"$.eventCounts[?(@.deviceLabel=='%s')]", self._device_label and self.serial_number in self.coordinator.data["mice"]
) and "detections" in self.coordinator.data["mice"][self.serial_number]
is not None
) )
@property @property

View File

@ -6,17 +6,16 @@ from typing import Any, Callable
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import VerisureDataUpdateCoordinator
from .const import CONF_SMARTPLUGS, DOMAIN from .const import CONF_SMARTPLUGS, DOMAIN
from .coordinator import VerisureDataUpdateCoordinator
def setup_platform( def setup_platform(
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any], config: dict[str, Any],
add_entities: Callable[[list[Entity], bool], None], add_entities: Callable[[list[CoordinatorEntity]], None],
discovery_info: dict[str, Any] | None = None, discovery_info: dict[str, Any] | None = None,
) -> None: ) -> None:
"""Set up the Verisure switch platform.""" """Set up the Verisure switch platform."""
@ -27,8 +26,8 @@ def setup_platform(
add_entities( add_entities(
[ [
VerisureSmartplug(coordinator, device_label) VerisureSmartplug(coordinator, serial_number)
for device_label in coordinator.get("$.smartPlugs[*].deviceLabel") for serial_number in coordinator.data["smart_plugs"]
] ]
) )
@ -39,20 +38,18 @@ class VerisureSmartplug(CoordinatorEntity, SwitchEntity):
coordinator: VerisureDataUpdateCoordinator coordinator: VerisureDataUpdateCoordinator
def __init__( def __init__(
self, coordinator: VerisureDataUpdateCoordinator, device_id: str self, coordinator: VerisureDataUpdateCoordinator, serial_number: str
) -> None: ) -> None:
"""Initialize the Verisure device.""" """Initialize the Verisure device."""
super().__init__(coordinator) super().__init__(coordinator)
self._device_label = device_id self.serial_number = serial_number
self._change_timestamp = 0 self._change_timestamp = 0
self._state = False self._state = False
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name or location of the smartplug.""" """Return the name or location of the smartplug."""
return self.coordinator.get_first( return self.coordinator.data["smart_plugs"][self.serial_number]["area"]
"$.smartPlugs[?(@.deviceLabel == '%s')].area", self._device_label
)
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
@ -60,10 +57,7 @@ class VerisureSmartplug(CoordinatorEntity, SwitchEntity):
if monotonic() - self._change_timestamp < 10: if monotonic() - self._change_timestamp < 10:
return self._state return self._state
self._state = ( self._state = (
self.coordinator.get_first( self.coordinator.data["smart_plugs"][self.serial_number]["currentState"]
"$.smartPlugs[?(@.deviceLabel == '%s')].currentState",
self._device_label,
)
== "ON" == "ON"
) )
return self._state return self._state
@ -72,20 +66,18 @@ class VerisureSmartplug(CoordinatorEntity, SwitchEntity):
def available(self) -> bool: def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return ( return (
self.coordinator.get_first( super().available
"$.smartPlugs[?(@.deviceLabel == '%s')]", self._device_label and self.serial_number in self.coordinator.data["smart_plugs"]
)
is not None
) )
def turn_on(self, **kwargs) -> None: def turn_on(self, **kwargs) -> None:
"""Set smartplug status on.""" """Set smartplug status on."""
self.coordinator.session.set_smartplug_state(self._device_label, True) self.coordinator.verisure.set_smartplug_state(self.serial_number, True)
self._state = True self._state = True
self._change_timestamp = monotonic() self._change_timestamp = monotonic()
def turn_off(self, **kwargs) -> None: def turn_off(self, **kwargs) -> None:
"""Set smartplug status off.""" """Set smartplug status off."""
self.coordinator.session.set_smartplug_state(self._device_label, False) self.coordinator.verisure.set_smartplug_state(self.serial_number, False)
self._state = False self._state = False
self._change_timestamp = monotonic() self._change_timestamp = monotonic()

View File

@ -829,7 +829,6 @@ influxdb==5.2.3
iperf3==0.1.11 iperf3==0.1.11
# homeassistant.components.rest # homeassistant.components.rest
# homeassistant.components.verisure
jsonpath==0.82 jsonpath==0.82
# homeassistant.components.kaiterra # homeassistant.components.kaiterra

View File

@ -446,7 +446,6 @@ influxdb-client==1.14.0
influxdb==5.2.3 influxdb==5.2.3
# homeassistant.components.rest # homeassistant.components.rest
# homeassistant.components.verisure
jsonpath==0.82 jsonpath==0.82
# homeassistant.components.konnected # homeassistant.components.konnected