Source code for tardis.configuration.configuration

from ..interfaces.borg import Borg
from ..utilities.attributedict import AttributeDict
from ..utilities.attributedict import convert_to_attribute_dict

# Need to import all pyyaml loadable classes (bootstrapping problem) FIX ME
from ..utilities.executors import *  # noqa: F403, F401
from ..utilities.simulators import *  # noqa: F403, F401

from cobald.daemon.config.mapping import Translator
from cobald.daemon.plugins import constraints as plugin_constraints

from base64 import b64encode
import os
import yaml

[docs]def translate_config(obj): if isinstance(obj, AttributeDict): translated_obj = AttributeDict(obj) for key, value in obj.items(): if key == "user_data": # base64 encode user data with open(os.path.join(os.getcwd(), obj[key]), "rb") as f: translated_obj[key] = b64encode( elif key == "__type__": # do legacy object initialisation return Translator().translate_hierarchy(obj) else: translated_obj[key] = translate_config(value) return translated_obj elif isinstance(obj, list): return [translate_config(item) for item in obj] else: return obj
[docs]@plugin_constraints(before={"pipeline"}) class Configuration(Borg): _shared_state = AttributeDict() def __init__(self, configuration: [str, dict] = None): super(Configuration, self).__init__() if configuration: if isinstance(configuration, str): # interpret string as file name self.load_config(configuration) else: self.update_config(configuration)
[docs] def load_config(self, config_file: str) -> None: """ Loads YAML configuration file into shared state of the configuration borg :param config_file: The name of the configuration file to be loaded :type config_file: str """ with open(config_file, "r") as config_file: self.update_config(yaml.safe_load(config_file))
[docs] def update_config(self, configuration: dict): """ Updates the shared state of the configuration borg :param configuration: Dictionary containing the configuration :type configuration: dict """ self._shared_state.update( translate_config(convert_to_attribute_dict(configuration)) )