Feature Request: Add "alarm_triggered" binary sensor for Amcrest

I would like to refer to this feature request which has been closed. Looks like nothing has been done with the original idea of this request.

Request is about to add “alarm_tiggered” binary sensor to the Amcrest component.
On the AD110 Amcrest Doorbell, the AlarmLocal event appears analogous to the actual PIR motion detection on the doorbell. The already existing motion_detected sensor acts when there’s any change in the video signal detection area. In the Amcrest Smart Home app, PIR motion and video motion are combined as motion sensor, which is a much better detection that results in less false alarms.

I’m running HASSIO 2021.2.3 on a Raspberry PI3b+ with an Amcrest AD110 doorbell connected.
I created “custom_components/amcrest/” folder and copied the amcrest code into it.
Next I modified the binary_sensor.py file and added “alarm_triggered” binary sensor (PIR motion) to it (see below).
PIR motion is working now and I can combine the two binary sensors (alarm_triggered and motion_detected) in a new binary sensor which acts the same as the motion detection in the Amcrest Smart Home app:

amcrest:
  - host: xxx.xxx.xxx.xxx
    name: Amcrest Doorbell
    username: !secret my_amcrest_doorbell_username
    password: !secret my_amcrest_doorbell_password
    scan_interval: 60
    stream_source: rtsp
    resolution: high
    binary_sensors:
      - motion_detected
      - alarm_triggered
    sensors:
      - sdcard

binary_sensor:
  - platform: template
    sensors:
      amcrest_doorbell_motion_detected_combined:
        friendly_name: "Amcrest doorbell motion detected combined"
        device_class: motion
        value_template: >-
          {{ is_state('binary_sensor.amcrest_doorbell_alarm_detected', 'on') 
             and  is_state('binary_sensor.amcrest_doorbell_motion_detected', 'on') }}

The modified binary_sensor.py file of the Amcrest custom component looks like this:



"""Support for Amcrest IP camera binary sensors."""
from datetime import timedelta
import logging

from amcrest import AmcrestError
import voluptuous as vol

from homeassistant.components.binary_sensor import (
    DEVICE_CLASS_CONNECTIVITY,
    DEVICE_CLASS_MOTION,
    DEVICE_CLASS_SOUND,
    BinarySensorEntity,
)
from homeassistant.const import CONF_BINARY_SENSORS, CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.util import Throttle

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

_LOGGER = logging.getLogger(__name__)

SCAN_INTERVAL = timedelta(seconds=BINARY_SENSOR_SCAN_INTERVAL_SECS)
_ONLINE_SCAN_INTERVAL = timedelta(seconds=60 - BINARY_SENSOR_SCAN_INTERVAL_SECS)


# MijnGedacht: Added BINARY_SENSOR_ALARM_TRIGGERED
BINARY_SENSOR_ALARM_TRIGGERED = "alarm_triggered"
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_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")
# MijnGedacht: Added _ALARM_DETECTED_PARAMS
_ALARM_DETECTED_PARAMS = ("Alarm Detected", DEVICE_CLASS_MOTION, "AlarmLocal")
BINARY_SENSORS = {
    BINARY_SENSOR_AUDIO_DETECTED: _AUDIO_DETECTED_PARAMS,
    BINARY_SENSOR_AUDIO_DETECTED_POLLED: _AUDIO_DETECTED_PARAMS,
    # MijnGedacht: Added BINARY_SENSOR_ALARM_TRIGGERED
    BINARY_SENSOR_ALARM_TRIGGERED: _ALARM_DETECTED_PARAMS,
    BINARY_SENSOR_MOTION_DETECTED: _MOTION_DETECTED_PARAMS,
    BINARY_SENSOR_MOTION_DETECTED_POLLED: _MOTION_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 BINARY_SENSORS.items()
}
_EXCLUSIVE_OPTIONS = [
    {BINARY_SENSOR_MOTION_DETECTED, BINARY_SENSOR_MOTION_DETECTED_POLLED},
]

_UPDATE_MSG = "Updating %s binary sensor"


def check_binary_sensors(value):
    """Validate binary sensor configurations."""
    for exclusive_options in _EXCLUSIVE_OPTIONS:
        if len(set(value) & exclusive_options) > 1:
            raise vol.Invalid(
                f"must contain at most one of {', '.join(exclusive_options)}."
            )
    return value


async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    """Set up a binary sensor for an Amcrest IP Camera."""
    if discovery_info is None:
        return

    name = discovery_info[CONF_NAME]
    device = hass.data[DATA_AMCREST][DEVICES][name]
    async_add_entities(
        [
            AmcrestBinarySensor(name, device, sensor_type)
            for sensor_type in discovery_info[CONF_BINARY_SENSORS]
        ],
        True,
    )


class AmcrestBinarySensor(BinarySensorEntity):
    """Binary sensor for Amcrest camera."""

    def __init__(self, name, device, sensor_type):
        """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._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

    def update(self):
        """Update entity."""
        if self._sensor_type == BINARY_SENSOR_ONLINE:
            self._update_online()
        else:
            self._update_others()

    @Throttle(_ONLINE_SCAN_INTERVAL)
    def _update_online(self):
        if not (self._api.available or self.is_on):
            return
        _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.
            try:
                self._api.current_time
            except AmcrestError:
                pass
        self._state = self._api.available

    def _update_others(self):
        if not self.available:
            return
        _LOGGER.debug(_UPDATE_MSG, self._name)

        try:
            self._state = "channels" in self._api.event_channels_happened(
                self._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
            self.async_write_ha_state()
            return
        self.async_schedule_update_ha_state(True)

    @callback
    def async_event_received(self, start):
        """Update state from received event."""
        _LOGGER.debug(_UPDATE_MSG, self._name)
        self._state = start
        self.async_write_ha_state()

    async def async_added_to_hass(self):
        """Subscribe to signals."""
        self._unsub_dispatcher.append(
            async_dispatcher_connect(
                self.hass,
                service_signal(SERVICE_UPDATE, self._signal_name),
                self.async_on_demand_update,
            )
        )
        if self._event_code and self._sensor_type not in BINARY_POLLED_SENSORS:
            self._unsub_dispatcher.append(
                async_dispatcher_connect(
                    self.hass,
                    service_signal(SERVICE_EVENT, self._signal_name, self._event_code),
                    self.async_event_received,
                )
            )

    async def async_will_remove_from_hass(self):
        """Disconnect from update signal."""
        for unsub_dispatcher in self._unsub_dispatcher:
            unsub_dispatcher()

Having looked at the original closed PR, it was closed because work was done to the Amcrest component that the developer who made this feature did not respond to. That being said, it looks like you’ve done the work to implement it already, I’d say go ahead and take what you’ve done and open a PR with it!

Hi Nathan, it’s not clear to me how I can open a PR for this ? I thought this FR was enough.

No worries! If you check out the instructions on the Home Assistant Developers site, starting at Development Environment, there are instructions for how to set up an environment and then the next page has instructions on how to open a pull request. If you do it that way you can have the internal pride of having contributed to Home Assistant yourself! If you don’t want to, I’m sure there is someone else who can do the contribution, I may be able to do it myself, it’ll just take me some time to get to.

Thank for the info, I will give it a try :grinning:

The PR has been created.

1 Like

Don’t forget to vote for your own request