import asyncio
import logging
from caspia.meadow.client import ServiceGatewayMixin
from caspia.meadow.services import LightBase
logger = logging.getLogger(__name__)
[docs]class HueLight(ServiceGatewayMixin, LightBase):
def __init__(self, name, hue_identifier, attrs, hue_api):
self.hue_identifier = hue_identifier
self.hue_api = hue_api
self.state = attrs['state']
super().__init__(name, include=('brightness', ))
[docs] async def load_current_state(self, state):
if self.state != state:
self.state = state
await self.send_notifications_if_needed()
[docs] async def update_state(self, update):
await self.hue_api.set_light_state(self.hue_identifier, update)
self.state.update(update)
await self.send_notifications_if_needed()
[docs] async def send_notifications_if_needed(self):
await self.notify(self.is_on, self.state['on'], if_changed=True)
await self.notify(self.brightness, self.state['bri'] / 254, if_changed=True)
[docs] async def characteristic_write(self, characteristic, value, **kwargs):
if characteristic is self.is_on:
await self.update_state({'on': value, 'bri': self.state['bri']})
elif characteristic is self.toggle:
await self.characteristic_write(self.is_on, not self.state['on'])
elif characteristic is self.brightness:
await self.update_state({'bri': int(value * 254)})
else:
raise NotImplementedError
[docs] async def characteristic_read(self, characteristic, **kwargs):
if characteristic is self.is_on:
return self.state['on']
elif characteristic is self.brightness:
return self.state['bri'] / 254
else:
raise NotImplementedError
[docs]class HueServicesProvider:
def __init__(self, hue_api, config_data):
self.services = []
self.config_data = config_data
self.api = hue_api
self._sync_task = None
[docs] def get_light_config(self, hue_identifier):
for name, config in self.config_data.get('lights', {}).items():
if config.get('hue-identifier') == hue_identifier:
return name, config
return None, None
[docs] async def setup(self):
assert self._sync_task is None
lights = await self.api.list_lights()
for hue_identifier, light_attrs in lights.items():
name, _ = self.get_light_config(hue_identifier)
if name:
service = HueLight(name, hue_identifier, light_attrs, self.api)
self.services.append(service)
else:
logger.warning('Hue light "%s" (with identifier %d) not configured',
light_attrs['name'], hue_identifier)
self._sync_task = asyncio.ensure_future(self.sync_periodically())
[docs] async def sync_periodically(self, interval=10.0):
while True:
try:
logger.debug('starting periodic update')
lights = await self.api.list_lights()
updates = []
for hue_identifier, light_attrs in lights.items():
state = light_attrs['state']
for service in self.services:
if service.hue_identifier == hue_identifier:
updates.append(service.load_current_state(state))
await asyncio.wait(updates)
except Exception: # pylint: disable=broad-except
logger.exception('Failure when doing periodic update')
await asyncio.sleep(interval)