Source code for caspia.node.components.sensors.apds9300

import struct
from typing import Tuple

import arrow

from ..base import Component
from .base import SensorBase


class APDS9300Sensor(SensorBase):
    type = 0x0B

    class Config(SensorBase.Config):
        def __init__(self, *, interval, address=None):
            super().__init__(interval)
            self.address = address if address is not None else 0b0111001

        def to_config_dict(self):
            cfg = super().to_config_dict()
            cfg[0x11] = struct.pack('B', self.address)
            return cfg

        @classmethod
        def params_from_config_dict(cls, cfg):
            return dict(address=struct.unpack('B', cfg.get(0x11, None))[0],
                        **SensorBase.Config.params_from_config_dict(cfg))

        def __repr_fields__(self):
            return dict(address=self.address, **super().__repr_fields__())

    class MeasurementEvent(SensorBase.MeasurementEvent):
        def parse(self):
            super().parse()
            self.raw: Tuple[int, int] = self.measurement_data if not self.error else (None, None)

        @property
        def lux(self):
            if self.raw != (None, None):
                return APDS9300Sensor.raw_to_lux(self.raw)
            else:
                return None

        def __repr_fields__(self):
            return dict(time=self.timestamp, raw=self.raw, error=self.error, lux=self.lux)

    class State(Component.State):
        def __init__(self, raw=(None, None), error=None, timestamp=None):
            self.raw = raw
            self.error = error
            self.timestamp = timestamp

        def update_from_bytes(self, data):
            self.raw = APDS9300Sensor.parse_measurement_data(data)
            self.error = None
            self.timestamp = arrow.now()

        def update_from_error(self, error_code, data):
            self.raw = (None, None)
            self.error = error_code

        def update_from_event(self, event):
            if isinstance(event, APDS9300Sensor.MeasurementEvent):
                self.timestamp = event.timestamp
                if event.error:
                    self.error = event.error
                    self.raw = (None, None)
                else:
                    self.error = None
                    self.raw = event.measurement_data

        @property
        def lux(self):
            if self.raw != (None, None):
                return APDS9300Sensor.raw_to_lux(self.raw)
            else:
                return None

        def __repr_fields__(self):
            return dict(raw=self.raw, error=self.error, timestamp=self.timestamp, lux=self.lux)

    @classmethod
    def parse_measurement_data(cls, data):
        return struct.unpack('<HH', data)

    @classmethod
    def raw_to_lux(cls, raw):
        ch0, ch1 = raw
        if ch0 == 0:
            return 0.0
        ch1_ch0 = ch1 / ch0
        if ch1_ch0 <= 0.52:
            return (0.0315 * ch0) - (0.0593 * ch0 * (ch1_ch0**1.4))
        elif ch1_ch0 <= 0.65:
            return (0.0229 * ch0) - (0.0291 * ch1)
        elif ch1_ch0 <= 0.80:
            return (0.0157 * ch0) - (0.0180 * ch1)
        elif ch1_ch0 <= 1.3:
            return (0.00338 * ch0) - (0.00260 * ch1)
        else:
            return 0.0