import asyncio
import logging
import aiohttp
from caspia.toolbox.storage import storage_property
from caspia.toolbox.upnp import msearch
logger = logging.getLogger(__name__)
[docs]class HueError(Exception):
def __init__(self, data):
super().__init__()
self.type = data.get('type')
self.address = data.get('address')
self.description = data.get('description')
def __str__(self):
return self.description
[docs]class HueApi:
def __init__(self, storage, devicetype='python#test'):
self.devicetype = devicetype
self.session = aiohttp.ClientSession()
self.storage = storage
self.ip_address = None
self._bridge_detected = asyncio.Event()
self._scan_task = asyncio.ensure_future(self.scan_for_ip_address())
user_name = storage_property('user_name')
[docs] async def connect(self):
await self._bridge_detected.wait()
while not self.user_name:
try:
await self.create_user()
except HueError as e:
if e.type == 101:
logger.warning('Not paired: please push the big button')
else:
logger.exception('Failed to pair')
except Exception: # pylint: disable=broad-except
logger.exception('Failed to pair')
await asyncio.sleep(3)
[docs] async def create_user(self):
request = {'devicetype': self.devicetype}
async with self.session.post(f'http://{self.ip_address}/api', json=request) as response:
data = (await response.json())[0]
if 'success' in data:
self.user_name = data['success']['username']
else:
raise HueError(data.get('error', {}))
[docs] async def list_lights(self):
url = f'http://{self.ip_address}/api/{self.user_name}/lights'
async with self.session.get(url) as response:
data = await response.json()
return {int(k): v for k, v in data.items()}
[docs] async def set_light_state(self, identifier, state):
url = f'http://{self.ip_address}/api/{self.user_name}/lights/{identifier}/state'
async with self.session.put(url, json=state) as response:
response.raise_for_status()
[docs] async def scan_for_ip_address(self):
while True:
ip_address = self.ip_address
async for service in msearch():
if 'hue-bridgeid' in service.data:
ip_address = service.src_ip
self._bridge_detected.set()
break
if not ip_address:
logger.warning('No hue-bridge detected')
elif self.ip_address != ip_address:
self.ip_address = ip_address
logger.info('Discovered hue-bridge ip address: %s', self.ip_address)
await asyncio.sleep(3.0 if self.ip_address is None else 60 * 10)