Domoticz

Code: Select all

import asyncio
import logging
import os

from aiohttp import ClientSession

from eufy_security import async_login
from eufy_security.types import GuardMode

logging.basicConfig(level=logging.ERROR)
# CRITICAL
# ERROR
# WARNING
# INFO
# DEBUG
# NOTSET

EUFY_EMAIL = "your_email_eufy"
EUFY_PASSWORD = "your_password"

async def main() -> None:
"""Create the aiohttp session and run the example."""
async with ClientSession() as websession:
# Create an API client:
api = await async_login(EUFY_EMAIL, EUFY_PASSWORD, websession)

for station in api.stations.values():
print("--ARMEMENT DE LA MAISON : PASSAGE EN MODE ABSENT--")
print(f"Station Name: {station.name}")
print(f"Serial Number: {station.serial}")
# print(f"Station params: {station.params}")
print(f"Station type: {station.device_type}")

async with station.connect() as session:
await station.set_guard_mode(GuardMode.AWAY, session)
await asyncio.sleep(10)


asyncio.get_event_loop().run_until_complete(main())

Code: Select all

import asyncio
import logging
import os

from aiohttp import ClientSession

from eufy_security import async_login
from eufy_security.types import GuardMode

logging.basicConfig(level=logging.ERROR)
# CRITICAL
# ERROR
# WARNING
# INFO
# DEBUG
# NOTSET

EUFY_EMAIL = "your_email_eufy"
EUFY_PASSWORD = "your_password"

async def main() -> None:
"""Create the aiohttp session and run the example."""
async with ClientSession() as websession:
# Create an API client:
api = await async_login(EUFY_EMAIL, EUFY_PASSWORD, websession)

for station in api.stations.values():
print("--ARMEMENT DE LA MAISON : PASSAGE EN MODE ABSENT--")
print(f"Station Name: {station.name}")
print(f"Serial Number: {station.serial}")
# print(f"Station params: {station.params}")
print(f"Station type: {station.device_type}")

async with station.connect() as session:
await station.set_guard_mode(GuardMode.HOME, session)
await asyncio.sleep(10)


asyncio.get_event_loop().run_until_complete(main())

Code: Select all

import asyncio
import logging
import os

from aiohttp import ClientSession

from eufy_security import async_login
from eufy_security.types import GuardMode

logging.basicConfig(level=logging.ERROR)
# CRITICAL
# ERROR
# WARNING
# INFO
# DEBUG
# NOTSET

EUFY_EMAIL = "your_email_eufy"
EUFY_PASSWORD = "your_password"

async def main() -> None:
"""Create the aiohttp session and run the example."""
async with ClientSession() as websession:
# Create an API client:
api = await async_login(EUFY_EMAIL, EUFY_PASSWORD, websession)

for station in api.stations.values():
print("--ARMEMENT DE LA MAISON : PASSAGE EN MODE ABSENT--")
print(f"Station Name: {station.name}")
print(f"Serial Number: {station.serial}")
# print(f"Station params: {station.params}")
print(f"Station type: {station.device_type}")

async with station.connect() as session:
await station.set_guard_mode(GuardMode.DISARMED, session)
await asyncio.sleep(10)


asyncio.get_event_loop().run_until_complete(main())

Code: Select all

import asyncio
import logging
import os

from aiohttp import ClientSession

from eufy_security import async_login
from eufy_security.types import GuardMode

logging.basicConfig(level=logging.DEBUG)


EUFY_EMAIL = ""
EUFY_PASSWORD = ""

async def main() -> None:
    """Create the aiohttp session and run the example."""
    async with ClientSession() as websession:
        # Create an API client:
        api = await async_login(EUFY_EMAIL, EUFY_PASSWORD, websession)

        for station in api.stations.values():
            print("------------------")
            print(f"Station Name: {station.name}")
            print(f"Serial Number: {station.serial}")
            print(f"Station params: {station.params}")
            print(f"Station type: {station.device_type}")

            async with station.connect() as session:
                await station.set_guard_mode(GuardMode.AWAY, session)
                await asyncio.sleep(10)
             


asyncio.get_event_loop().run_until_complete(main())

Code: Select all

/usr/bin/python3.7 /home/pi/domoticz/scripts/personnels/eufy_armedaway.py &

Code: Select all

-- Debug
local LOGGING = true

-- TELEGRAM settings
telegram_token = "****"
telegram_chatid = 00000
local t_actif = true

--Others settings
local version = 'EUFY on ArmedAway v0.0.1 du 2020/07'

--EMOJI
local emoji_CocheVerte = '\xE2\x9C\x85'
local emoji_PointExclamationRouge = '\xE2\x9D\x97'
local emoji_CroixRouge = '\xE2\x9D\x8C'
local emoji_Warning = '\xE2\x9A\xA0'

function telegram_EUFY_on_armedaway(message)
if (t_actif) then
os.execute('curl --data chat_id='..telegram_chatid..' --data-urlencode "text='..message..'" "https://api.telegram.org/bot'..telegram ... endMessage" ')
end
end

function isempty(s)
return s == nil or s == ''
end

function round(num, dec)
if num == 0 then
return 0
else
local mult = 10^(dec or 0)
return math.floor(num * mult + 0.5) / mult
end
end

return {
active = true,
--on = {['timer'] = {'every minute'}},
on = {security = { domoticz.SECURITY_ARMEDAWAY }},
execute = function(domoticz, item)
-- message pour telegram
local msg = domoticz.time.rawTime..'\n\r'..version..'\n\r\n\r'

if (LOGGING) then
print("Armement de la maison : Activation du mode ABSENT de EufySecurity")
end

--cmd='/usr/bin/python3.7 /home/pi/domoticz/scripts/personnels/eufy_armedaway.py'
--utilisation d'un shell avec l'execution en backgroud pour eviter de bloquer domoticz le temps de l'execution
cmd='/home/pi/domoticz/scripts/personnels/eufy_armedaway.sh'
--local f = assert(io.popen(cmd, 'r'))
--s = assert(f:read('*a'))
--f:close()
--print(s)
os.execute(cmd)
msg=msg..'Armement de la maison : Activation du mode ABSENT de EufySecurity'
telegram_EUFY_on_armedaway(msg)
end
}

Code: Select all

   async def connect(self, addr: str = None):
        dsk_key_resp = await self._api.request(
            "post", "app/equipment/get_dsk_keys", json={"station_sns": [self.serial]}
        )
        for item in dsk_key_resp.get("data")["dsk_keys"]:
            if item["station_sn"] == self.serial:
                p2p_session = P2PSession(
                    self.serial,
                    self.station_info["p2p_did"],
                    item["dsk_key"],
                    self.station_info["member"]["action_user_id"],
                )
                is_connected = await p2p_session.connect(addr)
                if is_connected:
                    try:
                        yield p2p_session
                    finally:
                        await p2p_session.close()
                    return
                else:
                    raise EufySecurityP2PError(f"Could not connect to {self.name}")
        else:
            raise EufySecurityP2PError(f"Could not find discovery key for {self.name}")