Source code for caspia.gateway.config.config

import warnings
from pathlib import Path

import yaml
from cached_property import cached_property

import caspia.node.utils
from caspia.gateway import errors

from .loading import ConfigLoader, deserialize_service_config, iterate_definitions

__all__ = ('GatewayConfig', )


class GatewayConfig:
    """Enables easy access to gateway's configuration."""
    def __init__(self, path: Path):
        self.path = path

    def _get_config_value(self, key):
        path = self.path / 'config.yaml'
        if not path.exists():
            return None
        with open(path, 'r') as f:
            data = yaml.load(f, ConfigLoader)
        return data.get(key)

    @cached_property
    def lightgroup_identifiers_range(self):
        range_desc = self._get_config_value('lightgroup-identifiers-range') or '0x0-0xFFFF'
        start, end = [int(i, 0) for i in range_desc.split('-')]
        return range(start, end + 1)

    @cached_property
    def allowed_canids(self):
        canids = self._get_config_value('allowed-canids')
        if canids:
            return canids
        canid_yaml = self.path / 'canid.yaml'
        if canid_yaml.is_file():
            with open(canid_yaml, 'r') as f:
                return yaml.load(f)
        raise RuntimeError('allowed-canids not configured')

    def iterate_services(self):
        """Return iterator over configs of all defined services."""
        context = dict(nodes=dict(self.iterate_nodes()))
        try:
            services = self._get_config_value('services')
            if services:
                for name, _, service_data in iterate_definitions(services):
                    yield deserialize_service_config(name, service_data, context)
            else:
                for file_path in self.path.iterdir():
                    if not file_path.is_file() or not file_path.name.endswith('services.yaml'):
                        continue
                    warnings.warn(f'{file_path} file found: please use config.yaml instead')
                    with open(file_path, 'r') as f:
                        data = yaml.load(f, ConfigLoader)
                    for name, _, service_data in iterate_definitions(data):
                        yield deserialize_service_config(name, service_data, context)
        except yaml.YAMLError as error:
            raise errors.ConfigurationSyntaxError(str(error))

    def iterate_nodes(self):
        """Return iterator over all tuples (node_name, hwid)."""
        def process_data(data):
            for name, _, hwidstr in iterate_definitions(data, hwid=True):
                yield name, caspia.node.utils.hwid_read(hwidstr)

        nodes = self._get_config_value('nodes')
        if nodes:
            yield from process_data(nodes)
        else:
            for file_path in self.path.iterdir():
                if not file_path.is_file() or not file_path.name.endswith('hwid.yaml'):
                    continue
                warnings.warn(f'{file_path} file found: please use config.yaml instead')
                with open(file_path, 'r') as f:
                    data = yaml.load(f, ConfigLoader)
                    yield from process_data(data)