Source code for caspia.hapbridge.config

import os.path
from typing import Iterable

import yaml
from aiohap import FileStorage

from caspia.toolbox.name import create_pattern_re


class BridgeConfig:
    def __init__(self, name, config, parent):
        self.name = name  # type: str
        self._config = config
        self._parent = parent

    def password(self) -> str:
        return self._config['password']

    def port(self) -> int:
        if 'port' not in self._config:
            raise RuntimeError('Port not specified for bridge %s' % self.name)
        return self._config['port']

    def storage(self) -> FileStorage:
        return FileStorage(self._parent.storage_path / self.name)

    def allows_service(self, service_name: str) -> bool:
        blacklist = self._parent._config.get('blacklist', [])  # pylint: disable=protected-access
        blacklist += self._config.get('blacklist', [])
        for pattern in blacklist:
            re = create_pattern_re(pattern)
            if re.fullmatch(service_name):
                return False
        for pattern in self._config['services']:
            re = create_pattern_re(pattern)
            if re.fullmatch(service_name):
                return True
        return False


class Config:
    def __init__(self, path, storage_path):
        self._path = path
        self.storage_path = storage_path
        with open(path, 'rb') as f:
            self._config = yaml.load(f.read())

    @property
    def host(self) -> str:
        return self._config['host']

    @property
    def external_address(self) -> str:
        if os.environ.get('HAP_EXTERNAL_IP_ADDRESS'):
            return os.environ.get('HAP_EXTERNAL_IP_ADDRESS')
        return self._config.get('external-ip-address')

    @property
    def notification_coalesce_delay(self) -> float:
        return self._config.get('notification-coalesce-delay', 0.05)

    @property
    def notification_coalesce_interval(self) -> float:
        return self._config.get('notification-coalesce-interval', 1.0)

    @property
    def bridges(self) -> Iterable[BridgeConfig]:
        for name, bridge_cfg in self._config['bridges'].items():
            yield BridgeConfig(name, bridge_cfg, self)