from copy import copy
from typing import Dict, Iterable
from caspia.meadow import name as mname
from caspia.meadow.serialization import Serializable
from caspia.toolbox.subclasses import all_subclasses
from .characteristic import Characteristic
[docs]class ServiceBase(Serializable, metaclass=ServiceMeta):
""" Base class for Meadow services. """
type: str = None
""" Type of the service
(subclasses should set this)
"""
auto_discovery: bool = True
""" True if this class should be discoverable using ``get_subclass``
"""
optional = set()
""" Names of optional characteristic.
(subclasses might set this)
"""
characteristics: Dict[str, Characteristic] = {}
""" List of characteristics defined for the given class/service type.
Subclasses should not set this directly. This attribute is initialized
automatically by `ServiceMeta` metaclass.
"""
metadata = Characteristic('json',
'RWN',
name='$metadata',
description="""
Special characteristic providing access to characteristic's metadata.
This characteristic is not implemented by any gateway, but one centralized
authority (homeserver).
""")
def __init__(self, name, include=()):
""" Initialize new service.
:param name: Name of the service (or just its path)
:param include: What optional characteristics to include. Tuple of their names.
"""
self.name: str = mname.update(name, type=self.type)
char_names = {
chname
for chname in type(self).characteristics
if chname not in self.optional or chname in include
}
self.characteristics: Dict[str, Characteristic] = {
prototype.name: self._create_characteristic(prototype.name)
for prototype in type(self).characteristics.values() if prototype.name in char_names
}
super().__init__()
[docs] def load_definition(self, spec):
""" Update characteristics of this instance based on serialized specs.
:param spec: Dict most likely going to be coming from meadow - from some gateway
informing us about its services and their characteristics.
"""
if spec['name'] != self.name:
raise RuntimeError(f'Trying to load definition of {spec["name"]} into {self.name}')
for char_name, char_spec in spec['characteristics'].items():
# get/create the instance
if char_name not in self.characteristics:
characteristic = self._create_characteristic(char_name, spec=char_spec)
else:
characteristic = self.characteristics[char_name]
# get minimum permissions for the characteristic
if char_name in type(self).characteristics and char_name not in type(self).optional:
required_perms = set(type(self).characteristics[char_name].permissions)
else:
required_perms = set()
spec_perms = set(char_spec['perm'].upper())
characteristic.permissions = ''.join(required_perms.union(spec_perms))
self.characteristics[char_name] = characteristic
def _create_characteristic(self, name, spec=None):
if name in type(self).characteristics:
char = copy(type(self).characteristics[name])
char.service = self
return char
else:
return Characteristic(spec['type'],
spec['perm'],
name=name,
service=self,
accepts_null=True)
[docs] async def characteristic_write(self, characteristic, value, **kwargs):
""" Write a value to the characteristic.
:param characteristic: The characteristic to which the `value` should be written.
:param value: The value to be written.
:param kwargs: Dictionary with `extra` key (dictionary with metadata) and optionally
a `timeout` (float specifing explicit timeout for the operation).
"""
raise NotImplementedError
[docs] async def characteristic_read(self, characteristic, **kwargs):
""" Read value of the characteristic.
:param characteristic: The characteristic to which the `value` should be written.
:param value: The value to be written.
:param kwargs: Dictionary with `extra` key (dictionary with metadata) and optionally
a `timeout` (float specifing explicit timeout for the operation).
:returns: Tuple (value, extra).
"""
raise NotImplementedError
[docs] def characteristic_subscribe(self, characteristic, on_value, on_error=None):
raise NotImplementedError
[docs] def characteristic_cached_read(self, characteristic):
""" This method implements return value of characteristic's `value` property. """
raise NotImplementedError('value property not supported')
[docs] def characteristic_cached_write(self, characteristic, value):
""" This method implements writing to characteristic's `value` property. """
raise NotImplementedError('value property not supported')
def __getitem__(self, key) -> Characteristic:
""" Get a characteristic with given name.
Allows easy access to the service characteristics.
Example:
>>> light = MyLightService(...)
>>> light['is_on'] is light.is_on # True
"""
if key in self.characteristics:
return self.characteristics[key]
else:
raise KeyError('No characteristic with name %s found.' % key)
def __iter__(self) -> Iterable[Characteristic]:
""" Iterate over characteristics """
yield from self.characteristics.values()
def __str__(self):
return self.name
def __repr__(self):
return '<%s(%s) at %x>' % (type(self).__name__, self.name, id(self))
#
# Subclass lookup
#
_subclass_cache = dict()
[docs] @staticmethod
def get_subclass(stype, mixin=None):
def validate(cls):
if not getattr(cls, 'auto_discovery', True):
return False
if mixin:
return issubclass(cls, mixin) and cls.type == stype
else:
return cls.__name__.endswith('Base') and cls.type == stype
key = (stype, mixin)
# try to find cached subclass
if key in ServiceBase._subclass_cache:
return ServiceBase._subclass_cache[key]
# try to find existing requested subclass
for subclass in all_subclasses(ServiceBase):
if validate(subclass):
ServiceBase._subclass_cache[key] = subclass
return subclass
# get type base class
base_cls = None
for subclass in all_subclasses(ServiceBase):
if subclass.type == stype and subclass.__name__.endswith('Base'):
base_cls = subclass
break
else:
base_cls = ServiceBase
# create subclass and return
typecls = type(''.join(x.title() for x in stype.split('-')), (mixin, base_cls),
{'type': stype})
ServiceBase._subclass_cache[key] = typecls
return typecls
#
# Observable Collection
#
@property
def all_observables(self):
return list(self.characteristics.values())
[docs] def on(self, name, *args, **kwargs):
for characteristic in self.characteristics.values():
if characteristic.name == name:
return characteristic
raise ValueError('Invalid trigger name %r' % name)
#
# Serializable
#
[docs] def serialize(self):
return {
'name': self.name,
'type': self.type,
'characteristics': {
char.name: {
'perm': char.permissions,
'type': char.value_type
}
for name, char in self.characteristics.items() if not char.name.startswith('$')
}
}