Source code for aiohap.model.accessory

import asyncio
from inspect import isclass
from typing import Any, Callable, List

from .defines import Category
from .service import Service


class Accessory:
    category = Category.OTHER

    def __init__(self, identifier, include=()):
        self.aid = None
        self.identifier = identifier
        self.primary_service = None
        self._notify_handlers: List[Callable[[int, int, Any], None]] = []
        self._services = dict()
        self._init_services(includes=self._parse_include(include))

    def _parse_include(self, include):
        if isinstance(include, str):
            include = (include, )
        includes = []
        for i in include:
            if isinstance(i, tuple):
                includes.append(i)
            else:
                includes.append(i.split(':'))
        return includes

    def _init_services(self, includes):
        for name in dir(type(self)):
            service_cls = getattr(type(self), name)
            if isinstance(service_cls, tuple):
                service_cls, *attrs = service_cls
            else:
                attrs = ()
            if not isclass(service_cls) or not issubclass(service_cls, Service):
                continue
            if not all(attr in ['hidden', 'primary'] for attr in attrs):
                raise ValueError('Invalid atributes %s' % attrs)
            include = tuple(char for nm, char in includes if nm == name)
            service = service_cls(hidden='hidden' in attrs, include=include)
            service.accessory = self
            self._services[name] = service
            if 'primary' in attrs:
                self.primary_service = service
            setattr(self, name, service)

    def handle_notify(self, func):
        self._notify_handlers.append(func)

    def notify(self, aid, iid, value):
        for handler in self._notify_handlers:
            handler(aid, iid, value)

    @property
    def services(self):
        return set(self._services.values())

    async def serialize(self, context, loop):
        if self.aid is None:
            raise RuntimeError('Aid not set')
        services_f = [
            s.serialize(context=dict(primary=s is self.primary_service, **context), loop=loop)
            for s in self.services
        ]
        services = await asyncio.gather(*services_f, loop=loop)
        return dict(aid=self.aid, services=services)

    def find_characteristic(self, iid):
        for service in self.services:
            for characteristic in service.characteristics:
                if characteristic.iid == iid:
                    return characteristic