[New Binding] Eufy Doorbell

Hey there,

I wanted to activate my lights when some motion is detected, so I routed the Events directly to the items over Openhab-Rest-API.

Have a look at my python script: (I know, it’s not perfect. It’s my first pythonscript, maybe someone can optimize it :wink: )

from ast import Continue
import asyncio
from math import e
import websockets
import json
import logging as LOG
import coloredlogs
from openhab import OpenHAB
import os

# Configure logging
LOG.basicConfig(level=LOG.INFO, format='%(asctime)s %(levelname)s %(message)s',
                        datefmt='%d-%b-%y %H:%M:%S')
coloredlogs.install(isatty=True) 


rest_url = os.environ.get('OPENHAB')
eufy_security_ws_url = os.environ.get('EUFY_SECURITY_WS')

if rest_url is None:
    LOG.error("OPENHAB environment variable not set. Falling back to default")
    rest_url = "http://redqueen:8081/rest" 

if eufy_security_ws_url is None:    
    LOG.error("EUFY_SECURITY_WS environment variable not set. Falling back to default")
    eufy_security_ws_url = "ws://redqueen:3000"
   

# Connect OpenHAB Rest API
openhab = OpenHAB(rest_url)


# Listen to events
async def listen():
    while True:
        try:
            async with websockets.connect(eufy_security_ws_url) as ws:
                
                #initialize connection
                LOG.info("Command: Initialize connection")
                await ws.send(json.dumps({
                    "messageId": "api-schema-id",
                    "command": "set_api_schema",
                    "schemaVersion": 9,}))
                
                answer = await ws.recv() 
                proceed_init(answer)

                #Start listening on events
                LOG.info("Command: Start listening")
                await ws.send(json.dumps({
                    "messageId": "start_listening",
                    "command": "start_listening",}))
                answer = await ws.recv() 
                proceed_answer(answer)

                # endless loop for receiving messages
                while True:
                    msg = await ws.recv()
                    process_event(msg)
        except ConnectionResetError:
            LOG.error("ConnectionResetError, reconnecting in 10 seconds...")
            await asyncio.sleep(10)
            Continue
        except websockets.ConnectionClosedError:
            LOG.error("ConnectionResetError, reconnecting in 10 seconds...")
            await asyncio.sleep(10)
            Continue
        except websockets.ConnectionClosedOK:
            LOG.error("ConnectionResetError, reconnecting in 10 seconds...")
            await asyncio.sleep(10)
            Continue
        except Exception as error:
            LOG.error("ConnectionResetError, reconnecting in 10 seconds...")
            await asyncio.sleep(10)
            Continue

def proceed_init(answer):
    obj = json.loads(answer)
    try:
        LOG.info(f"Success! driverVersion{obj['driverVersion']}, serverVersion: {obj['serverVersion']}")
    except KeyError:
        null = 0

# process the answer
def proceed_answer(answer):
    obj = json.loads(answer)
    try:
        success = obj["success"]
        if success == True:
            LOG.info("Success!")
        else:
            LOG.error("Error!")
    except KeyError:
        Continue

# process the message
def process_event(message):
    obj = json.loads(message)
    try:
        if obj["type"] == "event":
            eventname = obj["event"]["event"]
            if str(eventname) == "motion detected" or str(eventname) == "person detected":
                LOG.info(f"Motion detected: {obj['event']['event']}, Device: {obj['event']['serialNumber']}, State: {obj['event']['state']}")
                item = openhab.get_item(obj['event']['serialNumber'])
                state = obj['event']['state']
                if state == True:
                    item.on()
                else:
                    item.off()
        elif obj["type"] == "result":
            for station in obj["result"]["state"]["stations"]:
                LOG.info(f"Station online: {station['serialNumber']}, Name: {station['name']}")
            for device in obj["result"]["state"]["devices"]:
                LOG.info(f"Device online: {device['serialNumber']}, Name: {device['name']}")
        else:
            Continue
    except KeyError:
        Continue


# Start the loop
loop = asyncio.get_event_loop()
try: 
   # loop.create_task(listen())
    asyncio.ensure_future(listen())
    loop.run_forever()
except KeyboardInterrupt:
    pass

finally:
    print("Closing Loop")
    loop.close()

UPDATE: Script Updated, running stable!

Regards,
Daniel

1 Like

That’s what I have in mind too! This add the value of a “better motion detector for automations” to the Eufy products. By doing so we could even distinguish between pets, humans or vehicles triggering the detection.

Many thanks for sharing!

1 Like

What’s the status of a binding? I’, looking for support of the sirene

It is possible to connect a eufy doorbell?
I bought an S230 doorbell. It would be great to connect it to openhab and create rules.