"""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()
Comments