diff --git a/homeassistant/components/amcrest/__init__.py b/homeassistant/components/amcrest/__init__.py index ca99524f611..d248a3d8f7c 100644 --- a/homeassistant/components/amcrest/__init__.py +++ b/homeassistant/components/amcrest/__init__.py @@ -34,7 +34,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_s from homeassistant.helpers.event import track_time_interval from homeassistant.helpers.service import async_extract_entity_ids -from .binary_sensor import BINARY_POLLED_SENSORS, BINARY_SENSORS, check_binary_sensors +from .binary_sensor import BINARY_SENSOR_KEYS, BINARY_SENSORS, check_binary_sensors from .camera import CAMERA_SERVICES, STREAM_SOURCE_LIST from .const import ( CAMERAS, @@ -43,7 +43,6 @@ from .const import ( DATA_AMCREST, DEVICES, DOMAIN, - SENSOR_EVENT_CODE, SERVICE_EVENT, SERVICE_UPDATE, ) @@ -99,7 +98,10 @@ AMCREST_SCHEMA = vol.Schema( vol.Optional(CONF_FFMPEG_ARGUMENTS, default=DEFAULT_ARGUMENTS): cv.string, vol.Optional(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL): cv.time_period, vol.Optional(CONF_BINARY_SENSORS): vol.All( - cv.ensure_list, [vol.In(BINARY_SENSORS)], vol.Unique(), check_binary_sensors + cv.ensure_list, + [vol.In(BINARY_SENSOR_KEYS)], + vol.Unique(), + check_binary_sensors, ), vol.Optional(CONF_SENSORS): vol.All( cv.ensure_list, [vol.In(SENSOR_KEYS)], vol.Unique() @@ -276,9 +278,9 @@ def setup(hass, config): config, ) event_codes = [ - BINARY_SENSORS[sensor_type][SENSOR_EVENT_CODE] - for sensor_type in binary_sensors - if sensor_type not in BINARY_POLLED_SENSORS + sensor.event_code + for sensor in BINARY_SENSORS + if sensor.key in binary_sensors and not sensor.should_poll ] _start_event_monitor(hass, name, api, event_codes) diff --git a/homeassistant/components/amcrest/binary_sensor.py b/homeassistant/components/amcrest/binary_sensor.py index 98e0be73ef4..fcbadc73147 100644 --- a/homeassistant/components/amcrest/binary_sensor.py +++ b/homeassistant/components/amcrest/binary_sensor.py @@ -1,5 +1,8 @@ """Support for Amcrest IP camera binary sensors.""" +from __future__ import annotations + from contextlib import suppress +from dataclasses import dataclass from datetime import timedelta import logging @@ -11,6 +14,7 @@ from homeassistant.components.binary_sensor import ( DEVICE_CLASS_MOTION, DEVICE_CLASS_SOUND, BinarySensorEntity, + BinarySensorEntityDescription, ) from homeassistant.const import CONF_BINARY_SENSORS, CONF_NAME from homeassistant.core import callback @@ -21,54 +25,93 @@ from .const import ( BINARY_SENSOR_SCAN_INTERVAL_SECS, DATA_AMCREST, DEVICES, - SENSOR_DEVICE_CLASS, - SENSOR_EVENT_CODE, - SENSOR_NAME, SERVICE_EVENT, SERVICE_UPDATE, ) from .helpers import log_update_error, service_signal + +@dataclass +class AmcrestSensorEntityDescription(BinarySensorEntityDescription): + """Describe Amcrest sensor entity.""" + + event_code: str | None = None + should_poll: bool = False + + _LOGGER = logging.getLogger(__name__) SCAN_INTERVAL = timedelta(seconds=BINARY_SENSOR_SCAN_INTERVAL_SECS) _ONLINE_SCAN_INTERVAL = timedelta(seconds=60 - BINARY_SENSOR_SCAN_INTERVAL_SECS) -BINARY_SENSOR_AUDIO_DETECTED = "audio_detected" -BINARY_SENSOR_AUDIO_DETECTED_POLLED = "audio_detected_polled" -BINARY_SENSOR_MOTION_DETECTED = "motion_detected" -BINARY_SENSOR_MOTION_DETECTED_POLLED = "motion_detected_polled" -BINARY_SENSOR_ONLINE = "online" -BINARY_SENSOR_CROSSLINE_DETECTED = "crossline_detected" -BINARY_SENSOR_CROSSLINE_DETECTED_POLLED = "crossline_detected_polled" -BINARY_POLLED_SENSORS = [ - BINARY_SENSOR_AUDIO_DETECTED_POLLED, - BINARY_SENSOR_MOTION_DETECTED_POLLED, - BINARY_SENSOR_ONLINE, -] -_AUDIO_DETECTED_PARAMS = ("Audio Detected", DEVICE_CLASS_SOUND, "AudioMutation") -_MOTION_DETECTED_PARAMS = ("Motion Detected", DEVICE_CLASS_MOTION, "VideoMotion") -_CROSSLINE_DETECTED_PARAMS = ( - "CrossLine Detected", - DEVICE_CLASS_MOTION, - "CrossLineDetection", +_AUDIO_DETECTED_KEY = "audio_detected" +_AUDIO_DETECTED_POLLED_KEY = "audio_detected_polled" +_AUDIO_DETECTED_NAME = "Audio Detected" +_AUDIO_DETECTED_EVENT_CODE = "AudioMutation" + +_CROSSLINE_DETECTED_KEY = "crossline_detected" +_CROSSLINE_DETECTED_POLLED_KEY = "crossline_detected_polled" +_CROSSLINE_DETECTED_NAME = "CrossLine Detected" +_CROSSLINE_DETECTED_EVENT_CODE = "CrossLineDetection" + +_MOTION_DETECTED_KEY = "motion_detected" +_MOTION_DETECTED_POLLED_KEY = "motion_detected_polled" +_MOTION_DETECTED_NAME = "Motion Detected" +_MOTION_DETECTED_EVENT_CODE = "VideoMotion" + +_ONLINE_KEY = "online" + +BINARY_SENSORS: tuple[AmcrestSensorEntityDescription, ...] = ( + AmcrestSensorEntityDescription( + key=_AUDIO_DETECTED_KEY, + name=_AUDIO_DETECTED_NAME, + device_class=DEVICE_CLASS_SOUND, + event_code=_AUDIO_DETECTED_EVENT_CODE, + ), + AmcrestSensorEntityDescription( + key=_AUDIO_DETECTED_POLLED_KEY, + name=_AUDIO_DETECTED_NAME, + device_class=DEVICE_CLASS_SOUND, + event_code=_AUDIO_DETECTED_EVENT_CODE, + should_poll=True, + ), + AmcrestSensorEntityDescription( + key=_CROSSLINE_DETECTED_KEY, + name=_CROSSLINE_DETECTED_NAME, + device_class=DEVICE_CLASS_MOTION, + event_code=_CROSSLINE_DETECTED_EVENT_CODE, + ), + AmcrestSensorEntityDescription( + key=_CROSSLINE_DETECTED_POLLED_KEY, + name=_CROSSLINE_DETECTED_NAME, + device_class=DEVICE_CLASS_MOTION, + event_code=_CROSSLINE_DETECTED_EVENT_CODE, + should_poll=True, + ), + AmcrestSensorEntityDescription( + key=_MOTION_DETECTED_KEY, + name=_MOTION_DETECTED_NAME, + device_class=DEVICE_CLASS_MOTION, + event_code=_MOTION_DETECTED_EVENT_CODE, + ), + AmcrestSensorEntityDescription( + key=_MOTION_DETECTED_POLLED_KEY, + name=_MOTION_DETECTED_NAME, + device_class=DEVICE_CLASS_MOTION, + event_code=_MOTION_DETECTED_EVENT_CODE, + should_poll=True, + ), + AmcrestSensorEntityDescription( + key=_ONLINE_KEY, + name="Online", + device_class=DEVICE_CLASS_CONNECTIVITY, + ), ) -RAW_BINARY_SENSORS = { - BINARY_SENSOR_AUDIO_DETECTED: _AUDIO_DETECTED_PARAMS, - BINARY_SENSOR_AUDIO_DETECTED_POLLED: _AUDIO_DETECTED_PARAMS, - BINARY_SENSOR_MOTION_DETECTED: _MOTION_DETECTED_PARAMS, - BINARY_SENSOR_MOTION_DETECTED_POLLED: _MOTION_DETECTED_PARAMS, - BINARY_SENSOR_CROSSLINE_DETECTED: _CROSSLINE_DETECTED_PARAMS, - BINARY_SENSOR_CROSSLINE_DETECTED_POLLED: _CROSSLINE_DETECTED_PARAMS, - BINARY_SENSOR_ONLINE: ("Online", DEVICE_CLASS_CONNECTIVITY, None), -} -BINARY_SENSORS = { - k: dict(zip((SENSOR_NAME, SENSOR_DEVICE_CLASS, SENSOR_EVENT_CODE), v)) - for k, v in RAW_BINARY_SENSORS.items() -} +BINARY_SENSOR_KEYS = [description.key for description in BINARY_SENSORS] _EXCLUSIVE_OPTIONS = [ - {BINARY_SENSOR_MOTION_DETECTED, BINARY_SENSOR_MOTION_DETECTED_POLLED}, - {BINARY_SENSOR_CROSSLINE_DETECTED, BINARY_SENSOR_CROSSLINE_DETECTED_POLLED}, + {_AUDIO_DETECTED_KEY, _AUDIO_DETECTED_POLLED_KEY}, + {_MOTION_DETECTED_KEY, _MOTION_DETECTED_POLLED_KEY}, + {_CROSSLINE_DETECTED_KEY, _CROSSLINE_DETECTED_POLLED_KEY}, ] _UPDATE_MSG = "Updating %s binary sensor" @@ -91,10 +134,12 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info= name = discovery_info[CONF_NAME] device = hass.data[DATA_AMCREST][DEVICES][name] + binary_sensors = discovery_info[CONF_BINARY_SENSORS] async_add_entities( [ - AmcrestBinarySensor(name, device, sensor_type) - for sensor_type in discovery_info[CONF_BINARY_SENSORS] + AmcrestBinarySensor(name, device, entity_description) + for entity_description in BINARY_SENSORS + if entity_description.key in binary_sensors ], True, ) @@ -103,45 +148,23 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info= class AmcrestBinarySensor(BinarySensorEntity): """Binary sensor for Amcrest camera.""" - def __init__(self, name, device, sensor_type): + def __init__(self, name, device, entity_description): """Initialize entity.""" - self._name = f"{name} {BINARY_SENSORS[sensor_type][SENSOR_NAME]}" self._signal_name = name self._api = device.api - self._sensor_type = sensor_type - self._state = None - self._device_class = BINARY_SENSORS[sensor_type][SENSOR_DEVICE_CLASS] - self._event_code = BINARY_SENSORS[sensor_type][SENSOR_EVENT_CODE] + self.entity_description = entity_description + self._attr_name = f"{name} {entity_description.name}" + self._attr_should_poll = entity_description.should_poll self._unsub_dispatcher = [] - @property - def should_poll(self): - """Return True if entity has to be polled for state.""" - return self._sensor_type in BINARY_POLLED_SENSORS - - @property - def name(self): - """Return entity name.""" - return self._name - - @property - def is_on(self): - """Return if entity is on.""" - return self._state - - @property - def device_class(self): - """Return device class.""" - return self._device_class - @property def available(self): """Return True if entity is available.""" - return self._sensor_type == BINARY_SENSOR_ONLINE or self._api.available + return self.entity_description.key == _ONLINE_KEY or self._api.available def update(self): """Update entity.""" - if self._sensor_type == BINARY_SENSOR_ONLINE: + if self.entity_description.key == _ONLINE_KEY: self._update_online() else: self._update_others() @@ -150,32 +173,33 @@ class AmcrestBinarySensor(BinarySensorEntity): def _update_online(self): if not (self._api.available or self.is_on): return - _LOGGER.debug(_UPDATE_MSG, self._name) + _LOGGER.debug(_UPDATE_MSG, self.name) if self._api.available: # Send a command to the camera to test if we can still communicate with it. # Override of Http.command() in __init__.py will set self._api.available # accordingly. with suppress(AmcrestError): self._api.current_time # pylint: disable=pointless-statement - self._state = self._api.available + self._attr_is_on = self._api.available def _update_others(self): if not self.available: return - _LOGGER.debug(_UPDATE_MSG, self._name) + _LOGGER.debug(_UPDATE_MSG, self.name) + event_code = self.entity_description.event_code try: - self._state = "channels" in self._api.event_channels_happened( - self._event_code + self._attr_is_on = "channels" in self._api.event_channels_happened( + event_code ) except AmcrestError as error: log_update_error(_LOGGER, "update", self.name, "binary sensor", error) async def async_on_demand_update(self): """Update state.""" - if self._sensor_type == BINARY_SENSOR_ONLINE: - _LOGGER.debug(_UPDATE_MSG, self._name) - self._state = self._api.available + if self.entity_description.key == _ONLINE_KEY: + _LOGGER.debug(_UPDATE_MSG, self.name) + self._attr_is_on = self._api.available self.async_write_ha_state() return self.async_schedule_update_ha_state(True) @@ -183,8 +207,8 @@ class AmcrestBinarySensor(BinarySensorEntity): @callback def async_event_received(self, start): """Update state from received event.""" - _LOGGER.debug(_UPDATE_MSG, self._name) - self._state = start + _LOGGER.debug(_UPDATE_MSG, self.name) + self._attr_is_on = start self.async_write_ha_state() async def async_added_to_hass(self): @@ -196,11 +220,18 @@ class AmcrestBinarySensor(BinarySensorEntity): self.async_on_demand_update, ) ) - if self._event_code and self._sensor_type not in BINARY_POLLED_SENSORS: + if ( + self.entity_description.event_code + and not self.entity_description.should_poll + ): self._unsub_dispatcher.append( async_dispatcher_connect( self.hass, - service_signal(SERVICE_EVENT, self._signal_name, self._event_code), + service_signal( + SERVICE_EVENT, + self._signal_name, + self.entity_description.event_code, + ), self.async_event_received, ) ) diff --git a/homeassistant/components/amcrest/const.py b/homeassistant/components/amcrest/const.py index ba7597d61af..89cde63a08a 100644 --- a/homeassistant/components/amcrest/const.py +++ b/homeassistant/components/amcrest/const.py @@ -13,7 +13,3 @@ SNAPSHOT_TIMEOUT = 20 SERVICE_EVENT = "event" SERVICE_UPDATE = "update" - -SENSOR_DEVICE_CLASS = "class" -SENSOR_EVENT_CODE = "code" -SENSOR_NAME = "name"