Source code for caspia.gateway.config.schema.component

import typing

from marshmallow import Schema, ValidationError, fields, post_load, validate

import caspia.node
from caspia.node import components


class NodeRefField(fields.String):
    def _validate(self, value):
        super()._validate(value)
        if value not in self.context.get('nodes', []):
            raise ValidationError(f'undefined node {value}')


class PinField(fields.Integer):
    def _deserialize(self, value, attr, data):
        int_val = super()._deserialize(value, attr, data)
        return caspia.node.Pin(int_val)


class PinModeField(fields.String):
    def _deserialize(self, value, attr, data):
        str_val = super()._deserialize(value, attr, data)
        if str_val == 'up':
            return components.DigitalInput.Config.MODE_PULL_UP
        elif str_val == 'down':
            return components.DigitalInput.Config.MODE_PULL_DOWN
        else:
            raise ValidationError('invalid pin mode', data=value)


class TSL258XSensorGainControlField(fields.Integer):
    def _deserialize(self, value, attr, data):
        TSL258XSensor = components.sensors.tsl258x.TSL258XSensor
        int_val = super()._deserialize(value, attr, data)
        if int_val == 1:
            return TSL258XSensor.GainControl.Gain1x
        elif int_val == 8:
            return TSL258XSensor.GainControl.Gain8x
        elif int_val == 16:
            return TSL258XSensor.GainControl.Gain16x
        elif int_val == 111:
            return TSL258XSensor.GainControl.Gain111x
        else:
            msg = 'invalid gain, valid values are 1, 8, 16 and 111'
            raise ValidationError(msg, data=data)


class ComponentSchema(Schema):
    config_cls: typing.Callable
    node = NodeRefField(required=True)

    def make_config(self, data):
        self.context['node'] = data.pop('node', None)
        return self.config_cls(**data)

    @post_load
    def _make_config(self, data):
        return self.make_config(data)


class ButtonSchema(ComponentSchema):
    config_cls = components.Button.Config
    pin = PinField(required=True)


class RelaySchema(ComponentSchema):
    config_cls = components.Relay.Config
    pin = PinField(required=True)
    inverted = fields.Boolean(missing=False)
    initial_state = fields.Boolean(missing=False)
    timeout = fields.Number(missing=0.0, validate=lambda x: x >= 0.0)


class SensorSchema(ComponentSchema):
    interval = fields.Integer(missing=0, validate=lambda x: 0 <= x < 0xFFFF)


class MCP980XSensorSchema(SensorSchema):
    config_cls = components.sensors.MCP980XSensor.Config
    address = fields.Integer()
    resolution = fields.Integer(
        validate=lambda x: 0 <= x < components.sensors.MCP980XSensor.Config.RESOLUTION_12b)


class AnalogSensorSchema(SensorSchema):
    config_cls = components.sensors.AnalogSensor.Config
    pin = PinField(required=True)
    power_pin = PinField()
    samples = fields.Integer(missing=1, validate=lambda x: 1 <= x < 0xFF)


class BlindsSchema(ComponentSchema):
    class TimingSchema(Schema):
        move_up_delay = fields.Number(required=True)
        move_down_delay = fields.Number(required=True)
        direction_change_delay = fields.Number(required=True)
        tilt_time = fields.Number(required=True)
        move_up_time = fields.Number(required=True)
        move_down_time = fields.Number(required=True)
        arrest_time = fields.Number(required=True)
        backlash = fields.Number()

    config_cls = components.Blinds.Config
    power_pin = PinField(required=True, attribute='pwr_pin')
    direction_pin = PinField(required=True, attribute='dir_pin')
    direction_inverted = fields.Boolean(missing=False, attribute='dir_pin_inverted')
    calibrate_at_bottom = fields.Boolean(missing=False)
    timing = fields.Nested(TimingSchema, required=True)

    def make_config(self, data):
        timing = data.pop('timing')
        data.update(timing)
        return super().make_config(data)


class S300Schema(SensorSchema):
    config_cls = components.sensors.S300Sensor.Config


class DigitalInputSchema(ComponentSchema):
    config_cls = components.DigitalInput.Config
    mode = PinModeField(missing='up', load_from='pull')
    pin = PinField(required=True)
    sampling_freq = fields.Integer(missing=100)
    samples_till_change = fields.Integer(missing=3)
    broadcast_on_low_to_high = fields.Boolean(missing=True)
    broadcast_on_high_to_low = fields.Boolean(missing=True)
    inverted = fields.Boolean(missing=False)


class SerialSchema(ComponentSchema):
    config_cls = components.Serial.Config
    baudrate = fields.Integer(required=True)


class SHT2XSensorSchema(SensorSchema):
    config_cls = components.sensors.SHT2XSensor.Config


class APDS9300SensorSchema(SensorSchema):
    config_cls = components.sensors.APDS9300Sensor.Config
    address = fields.Integer()


class SCD30SensorSchema(SensorSchema):
    config_cls = components.sensors.SCD30Sensor.Config
    temperature_offset = fields.Integer(validate=lambda x: 0 <= x <= 0xFFFF)
    ambient_pressure = fields.Integer(validate=lambda x: x == 0 or 700 <= x <= 1200)


class TSL258XSensorSchema(SensorSchema):
    config_cls = components.sensors.TSL258XSensor.Config
    gain_control = TSL258XSensorGainControlField(missing=1, load_from='gain')
    integration_cycles = fields.Integer(missing=1, validate=validate.Range(min=1, max=255))