"""Support for Sesame Japanese Version, by CANDY HOUSE.""" from __future__ import annotations import aiohttp import asyncio import base64 import datetime import pysesame2 import requests import voluptuous as vol from Crypto.Cipher import AES from Crypto.Hash import CMAC import homeassistant.helpers.config_validation as cv from homeassistant.components.lock import PLATFORM_SCHEMA, LockEntity from homeassistant.const import ATTR_BATTERY_LEVEL, ATTR_DEVICE_ID, CONF_NAME, CONF_API_KEY, CONF_DEVICE_ID, \ CONF_CLIENT_SECRET from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType ATTR_SERIAL_NO = "serial" PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ vol.Required(CONF_NAME): cv.string, vol.Required(CONF_DEVICE_ID): cv.string, vol.Required(CONF_API_KEY): cv.string, vol.Required(CONF_CLIENT_SECRET): cv.string, }) def setup_platform( hass: HomeAssistant, config: ConfigType, add_entities: AddEntitiesCallback, discovery_info: DiscoveryInfoType | None = None, ) -> None: name = config.get(CONF_NAME) device_id = config.get(CONF_DEVICE_ID) api_key = config.get(CONF_API_KEY) client_secret = config.get(CONF_CLIENT_SECRET) add_entities( [SesameJPDevice( name=name, uuid=device_id, api_key=api_key, secret_key=client_secret, )], update_before_add=True, ) class SesameJPDevice(LockEntity): def __init__( self, name, uuid, api_key, secret_key, ) -> None: self._name: str = name self._uuid: str = uuid self._api_key: str = api_key self._secret_key: str = secret_key self._is_locked = False self._responsive = False self._battery = -1 self.update() @property def name(self) -> str | None: return self._name @property def available(self) -> bool: return self._responsive @property def is_locked(self) -> bool: return self._is_locked def lock(self, **kwargs) -> None: asyncio.run(self._sesame_command(action="LOCK")) def unlock(self, **kwargs) -> None: asyncio.run(self._sesame_command(action="UNLOCK")) def open(self, **kwargs) -> None: asyncio.run(self._sesame_command(action="UNLOCK")) def update(self) -> None: response = requests.get( "https://app.candyhouse.co/api/sesame2/{sesame2_uuid}".format(sesame2_uuid=self._uuid), headers={ "x-api-key": self._api_key, }, ) state = response.json() if state: self._responsive = True if state['CHSesame2Status'] == "locked": self._is_locked = True else: self._is_locked = False self._battery = state['batteryPercentage'] else: self._responsive = False @property def extra_state_attributes(self) -> dict: return {} # def _sesame_history(self) -> None: # response = requests.get( # "https://app.candyhouse.co/api/sesame2/{sesame2_uuid}/history?page=0&lg=1".format(sesame2_uuid=self._uuid), # headers={ # "x-api-key": self._api_key, # }, # ) # return response.json() async def _sesame_command(self, action) -> None: if action == "LOCK": cmd = 82 elif action == "UNLOCK": cmd = 83 else: return ts = int(datetime.datetime.now().timestamp()) message = ts.to_bytes(4, byteorder='little') message = message.hex()[2:8] cmac = CMAC.new(bytes.fromhex(self._secret_key), ciphermod=AES) cmac.update(bytes.fromhex(message)) sign = cmac.hexdigest() async with aiohttp.request( "POST", "https://app.candyhouse.co/api/sesame2/{sesame2_uuid}/cmd".format(sesame2_uuid=self._uuid), headers={ "x-api-key": self._api_key, }, json={ "cmd": cmd, "history": base64.b64encode(bytes("Home-Assistant {action}".format(action=action), 'utf-8')).decode(), "sign": sign, }, ) as resp: await resp.text()