from types import ModuleType
from marshmallow import (Schema, ValidationError, fields, post_load, pre_load, validates,
validates_schema)
from . import component
[docs]class ServiceSchema(Schema):
type_ = fields.String(load_from='type', attribute='type', required=True)
name = fields.String()
[docs] @pre_load
def check_one_of(self, data):
if not hasattr(self, 'one_of'):
return data
one_of = getattr(self, 'one_of')
count = sum(1 if k in one_of else 0 for k in data)
if count == 0:
msg = f'one of the following keys is required: {one_of}'
raise ValidationError(msg)
if count > 1:
msg = f'only one of the following keys is allowed: {one_of}'
raise ValidationError(msg)
return data
[docs] @post_load
def add_namespace(self, data):
data['node'] = self.context.get('node', None)
if 'name' in data:
data['name'] = '.'.join([self.context['namespace'], data['name']])
else:
data['name'] = self.context['namespace']
[docs] @classmethod
def get(cls, service_type):
if not hasattr(ServiceSchema, 'subclasses'):
cls.subclasses = {schema.type: schema for schema in cls.__subclasses__()}
return cls.subclasses[service_type]
[docs]class LightSchema(ServiceSchema):
type = 'light'
relay = fields.Nested(component.RelaySchema, required=True)
[docs]class OutletSchema(ServiceSchema):
type = 'outlet'
relay = fields.Nested(component.RelaySchema, required=True)
[docs]class FanSchema(ServiceSchema):
type = 'fan'
relay = fields.Nested(component.RelaySchema, required=True)
[docs]class PumpSchema(ServiceSchema):
type = 'pump'
relay = fields.Nested(component.RelaySchema, required=True)
[docs]class IrrigationSchema(ServiceSchema):
type = 'irrigation'
relay = fields.Nested(component.RelaySchema, required=True)
[docs]class TemperatureSensorSchema(ServiceSchema):
type = 'temperature-sensor'
analog = fields.Nested(component.AnalogSensorSchema)
mcp980x = fields.Nested(component.MCP980XSensorSchema)
sht2x = fields.Nested(component.SHT2XSensorSchema)
scd30 = fields.Nested(component.SCD30SensorSchema)
one_of = ('analog', 'mcp980x', 'sht2x', 'scd30')
[docs] @staticmethod
def default_equation(params, resistance):
import math
t_0 = 25.0 + 273.15
r_i = params['r0'] * math.exp(-params['beta'] / t_0)
return (params['beta'] / math.log(resistance / r_i)) - 273.15 + params.get('offset', 0.0)
# when `analog`
equation = fields.Raw()
params = fields.Dict(keys=fields.String())
[docs] @pre_load
def set_default_equation(self, data):
equation = data.get('equation', self.default_equation)
data['equation'] = getattr(equation, 'resistance_to_temp', equation)
[docs] @validates_schema
def validate_analog(self, data):
if 'analog' in data:
if 'params' not in data:
msg = f'you have to provide params (for default equation `beta` and `r0`)'
raise ValidationError(msg, 'params')
[docs]class HumiditySensorSchema(ServiceSchema):
type = 'humidity-sensor'
sht2x = fields.Nested(component.SHT2XSensorSchema)
scd30 = fields.Nested(component.SCD30SensorSchema)
one_of = 'sht2x', 'scd30'
[docs]class LightSensorSchema(ServiceSchema):
type = 'light-sensor'
analog = fields.Nested(component.AnalogSensorSchema)
apds9300 = fields.Nested(component.APDS9300SensorSchema)
tsl258x = fields.Nested(component.TSL258XSensorSchema)
one_of = ('analog', 'apds9300', 'tsl258x')
# when `analog`
equation = fields.Raw()
params = fields.Dict(keys=fields.String(), missing=dict())
[docs] @validates_schema
def validate_analog(self, data):
if 'analog' in data:
equation = data.get('equation')
if not equation:
msg = f'you have to provide equation'
raise ValidationError(msg, 'equation')
if isinstance(equation, ModuleType):
if not hasattr(equation, 'resistance_to_lux'):
msg = 'specified module has no "resistance_to_lux" func'
raise ValidationError(msg, 'equation')
equation = equation.resistance_to_lux
data['equation'] = equation
[docs]class CarbonDioxideSensorSchema(ServiceSchema):
type = 'carbon-dioxide-sensor'
s300 = fields.Nested(component.S300Schema)
scd30 = fields.Nested(component.SCD30SensorSchema)
one_of = 's300', 'scd30'
[docs]class DoorSchema(ServiceSchema):
type = 'door'
handle_input = fields.Nested(component.DigitalInputSchema)
opened_input = fields.Nested(component.DigitalInputSchema)
[docs]class LockMechanismSchema(ServiceSchema):
type = 'lock-mechanism'
relay = fields.Nested(component.RelaySchema, required=True)
power = fields.Nested(component.RelaySchema)
[docs]class ElectricityMeterSchema(ServiceSchema):
type = 'electricity-meter'
input = fields.Nested(component.DigitalInputSchema, required=True)
impulses_per_kwh = fields.Integer(required=True)
[docs]class BlindsSchema(ServiceSchema):
type = 'blinds'
component = fields.Nested(component.BlindsSchema, required=True)
angle_mapping = fields.Dict(keys=fields.Float(), values=fields.Integer())
[docs] @validates('angle_mapping')
def check_angle_mapping(self, data):
if len(data) != 2:
raise ValidationError('angle mapping must contain exactly two pairs')
for key in data:
if not 0 <= key <= 1.0:
raise ValidationError(f'{key} is outside of range (0, 1)')