Source code for caspia.gateway.services.sensors.light.apds9300

from caspia.gateway.services import GatewayService
from caspia.meadow.client import characteristic_read_handler
from caspia.meadow.errors import SensorFailureError
from caspia.meadow.services import LightSensorBase
from caspia.node.components.sensors.apds9300 import APDS9300Sensor


class APDS9300LightSensor(GatewayService, LightSensorBase):
    auto_discovery = False

    def configure(self):
        assert self.config['apds9300'].interval > 0, 'Sensor APDS-9300 must have interval > 0'
        self.apds9300_c: APDS9300Sensor = self.network.get_configured_component(
            self.get_node(), APDS9300Sensor, self.config['apds9300'])

    @property
    def is_updated_periodically(self):
        return self.config['apds9300'].interval != 0

    @property
    def dependant_components(self):
        return {self.apds9300_c}

    async def on_component_event(self, component, event):
        if self.apds9300_c == component:
            if isinstance(event, APDS9300Sensor.MeasurementEvent):
                if not event.error:
                    value = event.lux
                else:
                    value = SensorFailureError('sensor reported an error')
                await self.notify(self.light, value)

    @characteristic_read_handler('light')
    async def light_read(self, **kwargs):
        if not self.is_updated_periodically or self.apds9300_c.state.lux is None:
            lux = await self.apds9300_c.measure()
            error_code = self.apds9300_c.state.error
        else:
            lux = self.apds9300_c.state.lux
            error_code = self.apds9300_c.state.error
        if error_code:
            raise SensorFailureError(message=f'sensor failure (code {error_code})')
        return round(float(lux), 2)