Source code for nemo_nowcast.config

# Copyright 2016-2021 Doug Latornell, 43ravens

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at


# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

"""NEMO_Nowcast framework system configuration object.

Provides :py:class:`dict`-like access to the configuration loaded from the
YAML system configuration file.
import os
import re

import attr
import yaml

[docs]@attr.s class Config: """Construct a :py:class:`nemo_nowcast.config.Config` instance. """ #: Path/name of YAML configuration file for the NEMO nowcast system. #: Assigned when :py:meth:`~nemo_nowcast.config.Config.load` method #: is called. file = attr.ib(init=False, default="") #: :py:class:`dict` containing the nowcast system configuration #: that is read from the configuration file by the #: :py:meth:`~nemo_nowcast.config.Config.load` method. _dict = attr.ib(init=False, repr=False, default=attr.Factory(dict)) def __contains__(self, item): return item in self._dict def __getitem__(self, item): return self._dict[item] def __setitem__(self, key, value): self._dict[key] = value def get(self, key, default=None): try: return self._dict[key] except KeyError: return default
[docs] def load(self, config_file): """Load the YAML config_file. The value of config_file is stored on the :py:attr:`nemo_nowcast.config.Config.file` attribute. :arg config_file: Path/name of YAML configuration file for the NEMO nowcast system. :type config_file: :py:class:`pathlib.Path` or str """ self.file = config_file with open(config_file, "rt") as f: self._dict = yaml.safe_load(f) envvar_pattern = re.compile(r"\$\(NOWCAST\.ENV\.(\w*)\)\w*") envvar_sub_keys = ("checklist file", "python") for key in envvar_sub_keys: self._dict[key] = envvar_pattern.sub(self._replace_env, self._dict[key]) try: # Local logging self._replace_handler_envvars( envvar_pattern, self._dict["logging"]["handlers"] ) except KeyError: # Distributed logging self._replace_handler_envvars( envvar_pattern, self._dict["logging"]["aggregator"]["handlers"] ) self._replace_handler_envvars( envvar_pattern, self._dict["logging"]["publisher"]["handlers"] )
def _replace_handler_envvars(self, envvar_pattern, handlers): for handler in handlers: try: handlers[handler]["filename"] = envvar_pattern.sub( self._replace_env, handlers[handler]["filename"] ) except KeyError: # Not a file handler pass @staticmethod def _replace_env(var): try: return os.environ[] except KeyError: raise KeyError(f"environment variable not set: {}")