config.py

"""
Channel configuration loaded from the channels section of config.yml.
"""
import dataclasses
import datetime
import typing

from .logger import LOGGER
from .util import parse_duration

@dataclasses.dataclass(frozen=True)
class HistoryProviderConfig:
    """Identity used to fetch TimescaleDB history for bootstrap."""
    dp_name: str
    dp_location_code: typing.Optional[str] = None
    dp_unit: typing.Optional[str] = None
    dp_data_provider: typing.Optional[str] = None
    dp_device_id: typing.Optional[str] = None
    init_when_source_available: bool = False
    bootstrap_delay: datetime.timedelta = datetime.timedelta(seconds=10)


@dataclasses.dataclass(frozen=True)
class ChannelConfig:
    """
    Static configuration of a single measurement channel.

    Args:
        name: name of the channel (string)
        input_stream: name of the input stream (string)
        output_stream: name of the output stream (string)
        polling_interval: interval at which to poll the input stream (timedelta)
        update_interval: interval at which to update the regularizer (timedelta)
        jitter_tolerance: tolerance for jitter in the input stream (timedelta)
        window: window of the regularizer (timedelta)
        history_provider: optional TimescaleDB history provider (HistoryProviderConfig, default=None)
        offset: offset of the channel (timedelta, default=0)
        lag_time: lag time of the channel (timedelta, default=0)
        output_maxlen: maximum length of the output stream (int, default=200)
        data_provider_name: name of the data provider (string, default='rdp-regularizer')
        imputer: imputer to use (string, default='default')
        imputer_kwargs: constructor kwargs from an imputer mapping (dict, default empty)
        forecaster: forecaster to use (string, default='default')
        forecaster_kwargs: constructor kwargs from a forecaster mapping (dict, default empty)
    """
    name: str
    input_stream: str
    output_stream: str
    polling_interval: datetime.timedelta
    update_interval: datetime.timedelta
    jitter_tolerance: datetime.timedelta
    window: datetime.timedelta
    history_provider: typing.Optional[HistoryProviderConfig] = None
    offset: datetime.timedelta = dataclasses.field(default=datetime.timedelta(seconds=0))
    lag_time: datetime.timedelta = dataclasses.field(default=datetime.timedelta(seconds=0))
    output_maxlen: int = dataclasses.field(default=200)
    data_provider_name: str = dataclasses.field(default='rdp-regularizer')
    imputer: str = dataclasses.field(default='default')
    imputer_kwargs: typing.Dict[str, typing.Any] = dataclasses.field(default_factory=dict)
    forecaster: str = dataclasses.field(default='default')
    forecaster_kwargs: typing.Dict[str, typing.Any] = dataclasses.field(default_factory=dict)

    @staticmethod
    def load_channel_configs(channels: dict) -> typing.List['ChannelConfig']:
        """
        Builds one ChannelConfig per entry in the channels dict.
        """
        if not channels:
            raise RuntimeError('channels config is empty')

        configs = []
        for channel_key, entry in channels.items():
            # Convert the entry to a dictionary.
            entry = dict(entry)

            # Optional: omit history_provider to skip Timescale bootstrap.
            hp_entry = entry.pop('history_provider', None)
            if hp_entry is not None:
                try:
                    if 'bootstrap_delay' in hp_entry:
                        hp_entry['bootstrap_delay'] = parse_duration(hp_entry['bootstrap_delay'])
                    entry['history_provider'] = HistoryProviderConfig(**hp_entry)
                except TypeError as exc:
                    raise RuntimeError(
                        f'Invalid history provider config for channel {channel_key}: {exc}'
                    ) from exc

            # Parse channel config, handle special cases.
            try:
                entry['name'] = channel_key
                entry['window'] = parse_duration(entry['window'])
                entry['update_interval'] = parse_duration(entry['update_interval'])
                entry['polling_interval'] = parse_duration(entry['polling_interval'])
                entry['jitter_tolerance'] = (
                    0.5 * entry['update_interval'] if 'jitter_tolerance' not in entry
                    else parse_duration(entry['jitter_tolerance'])
                )
                if 'offset' in entry:
                    entry['offset'] = parse_duration(entry['offset'])
                if 'lag_time' in entry:
                    entry['lag_time'] = parse_duration(entry['lag_time'])
                # String name, or mapping {name, ...constructor kwargs}. Sibling
                # imputer_kwargs / forecaster_kwargs keys are not a YAML form.
                if 'imputer_kwargs' in entry or 'forecaster_kwargs' in entry:
                    raise TypeError(
                        'imputer/forecaster constructor args belong in a mapping '
                        'with name, not imputer_kwargs/forecaster_kwargs'
                    )
                entry['imputer'], entry['imputer_kwargs'] = ChannelConfig._parse_strategy(
                    entry.pop('imputer', 'default'), 'imputer'
                )
                entry['forecaster'], entry['forecaster_kwargs'] = ChannelConfig._parse_strategy(
                    entry.pop('forecaster', 'default'), 'forecaster'
                )

                config = ChannelConfig(**entry)
                ChannelConfig._warn_unless_frontier_on_grid(config)
            except KeyError as exc:
                raise RuntimeError(
                    f'Missing config field for channel {channel_key}: {exc}'
                ) from exc
            except TypeError as exc:
                raise RuntimeError(
                    f'Invalid channel config for channel {channel_key}: {exc}'
                ) from exc

            # Add the config to the list of configs.
            configs.append(config)

        return configs

    @staticmethod
    def _parse_strategy(
            value: typing.Any, key: str
        ) -> typing.Tuple[str, typing.Dict[str, typing.Any]]:
        """
        For forecasting and imputation strategies, this function accepts a name, or a mapping with ``name`` plus constructor kwargs.
        """
        if value is None:
            return 'default', {}
        if isinstance(value, str):
            if not value:
                raise TypeError(f'{key} must be a non-empty string or a mapping with name')
            return value, {}
        if isinstance(value, dict):
            spec = dict(value)
            name = spec.pop('name', None)
            if not isinstance(name, str) or not name:
                raise TypeError(f'{key} mapping requires a string name')
            return name, spec
        raise TypeError(f'{key} must be a string or a mapping with name')

    @staticmethod
    def _warn_unless_frontier_on_grid(config: 'ChannelConfig') -> None:
        """
        The frontier starts at ``floor(start, polling_interval) - lag_time`` and is
        then stepped by ``update_interval``. It stays on the measurement grid only
        when ``update_interval`` divides both ``polling_interval`` and ``lag_time``.
        """
        if config.update_interval <= datetime.timedelta(0):
            return
        for key in ('polling_interval', 'lag_time'):
            value = getattr(config, key)
            if value % config.update_interval != datetime.timedelta(0):
                LOGGER.warning(
                    f'Channel {config.name}: update_interval ({config.update_interval}) '
                    f'does not divide {key} ({value})'
                )