import logging from logger import getLogger from pathlib import Path from lxml import etree from functools import cached_property, lru_cache from arguments import ArgumentsParser from typing import NamedTuple, Optional class TriggerTuple(NamedTuple): name: str interval_in_seconds: int governor: str def __repr__(self): return f"{self.name}, interval: {self.interval_in_seconds}, governor: {self.governor}" class Config: def __init__(self, basepath: Path) -> None: self.basepath = basepath self.config = ArgumentsParser().parser.config @cached_property def root(self) -> etree.ElementTree: return etree.parse(self.basepath / self.config) @cached_property def app_root(self) -> etree.Element: return self.root.xpath('/cpuGovernorAutoAdjust').pop() @cached_property def loglevel(self) -> str: value = self.app_root.xpath('logLevel').pop().text.upper() _loglevels = ", ".join(logging.getLevelNamesMapping().keys()) if value not in logging.getLevelNamesMapping().keys(): raise ValueError( "logLevel can only contain one of these values: " f"{_loglevels}" ) return value @cached_property def log(self) -> logging.Logger: return getLogger(self.__class__.__name__, loglevel=self.loglevel.upper()) @cached_property def testmode(self) -> bool: value = self.app_root.xpath('testMode').pop().text.lower() if value is None: self.log.warning("testMode not set in config") if value not in ['true', 'false']: raise ValueError("testMode can only be 'true' or 'false") return True if value == 'true' else False @cached_property def triggertuples(self) -> list[TriggerTuple]: ret_val = [] _triggers = self.app_root.xpath('triggers/trigger') for trigger in _triggers: name = trigger.xpath('name').pop().text interval_in_seconds = trigger.xpath('intervalInSeconds').pop().text governor = trigger.xpath('governor').pop().text _new_trigger = TriggerTuple( name=name, interval_in_seconds=int(interval_in_seconds), governor=governor ) ret_val.append(_new_trigger) self.log.info("Loaded trigger: %s", _new_trigger) return ret_val @lru_cache(maxsize=99) def get_trigger_by_name(self, name: str) -> Optional[TriggerTuple]: self.log.debug("getting trigger: %s", name) _all_trigger_tuples = self.triggertuples.copy() ret_val = next((t for t in _all_trigger_tuples if t.name == name), None) if ret_val is None: self.log.error("trigger not found: %s", ret_val) return ret_val