Source code for cowidev.utils.params.config

from dataclasses import dataclass
from pyaml_env import parse_config

from cowidev.utils.paths import CONFIG_FILE
from cowidev.utils.exceptions import ConfigFileError


config_raw = parse_config(CONFIG_FILE, raise_if_na=False)
if config_raw is None:
    raise ConfigFileError(f"Empty configuration file! Check content from {CONFIG_FILE}.")


[docs]@dataclass() class VaccinationsProcessConfig: skip_complete: list skip_monotonic_check: list skip_anomaly_check: list
[docs]@dataclass() class BaseGetConfig: countries: list skip_countries: list
[docs]@dataclass() class TestingGetConfig(BaseGetConfig): pass
[docs]@dataclass() class VaccinationsGetConfig(BaseGetConfig): pass
[docs]@dataclass() class HospitalizationsGenerateConfig(BaseGetConfig): pass
[docs]@dataclass() class Base4Config: get: BaseGetConfig process: dict generate: dict export: dict
[docs]@dataclass() class TestingConfig(Base4Config): get: TestingGetConfig def __post_init__(self): if self.get["countries"] is None: self.get["countries"] = ["all"] if self.get["skip_countries"] is None: self.get["skip_countries"] = [] self.get = TestingGetConfig(**self.get)
[docs]@dataclass() class VaccinationsConfig(Base4Config): get: VaccinationsGetConfig process: VaccinationsProcessConfig def __post_init__(self): # Get if self.get["countries"] is None: self.get["countries"] = ["all"] if self.get["skip_countries"] is None: self.get["skip_countries"] = [] self.get = VaccinationsGetConfig(**self.get) # Process if self.process["skip_complete"] is None: self.process["skip_complete"] = [] if self.process["skip_monotonic_check"] is None: self.process["skip_monotonic_check"] = {} if self.process["skip_anomaly_check"] is None: self.process["skip_anomaly_check"] = {} self.process = VaccinationsProcessConfig(**self.process)
[docs]@dataclass() class HospitalizationsConfig: generate: HospitalizationsGenerateConfig def __post_init__(self): # Generate if self.generate["countries"] is None: self.generate["countries"] = ["all"] if self.generate["skip_countries"] is None: self.generate["skip_countries"] = [] self.generate = HospitalizationsGenerateConfig(**self.generate)
[docs]@dataclass() class PipelineConfig: testing: TestingConfig vaccinations: VaccinationsConfig hospitalizations: HospitalizationsConfig def __post_init__(self): self.testing = TestingConfig(**self.testing) self.vaccinations = VaccinationsConfig(**self.vaccinations) self.hospitalizations = HospitalizationsConfig(**self.hospitalizations)
[docs]@dataclass() class ExecutionConfig: parallel: bool njobs: int
[docs]@dataclass() class Config: pipeline: PipelineConfig execution: ExecutionConfig def __post_init__(self): self.pipeline = PipelineConfig(**self.pipeline) self.execution = ExecutionConfig(**self.execution)
# config_raw["global_"] = config_raw.pop("global") try: CONFIG = Config(**config_raw) except TypeError as e: err_msg = ( f"The format of the configuration file is not correct! Please check file {CONFIG_FILE} that it contains all" " required fields. Original raised error was: " + str(e) ) raise ConfigFileError(err_msg)