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, Any from collections import OrderedDict import xmltodict class TriggerTuple(NamedTuple): name: str interval_in_seconds: int governor: str custom_config: OrderedDict[str, Any] def __repr__(self) -> str: return f"{self.name}, interval: {self.interval_in_seconds}, governor: {self.governor}, custom_config: {self.custom_config}" 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: _tree = self.root.xpath('/cpuGovernorAutoAdjust') assert isinstance(_tree, list) and len(_tree) == 1, "main config section cpuGovernorAutoAdjust not found" ret_val = _tree.pop() assert isinstance(ret_val, etree._Element), "Unexpected error has occurred, cpuGovernorAutoAdjust is not of type etree._Element" return ret_val @cached_property def loglevel(self) -> str: _loglevel = self._get_single_text_value_from_xpath('logLevel') value = _loglevel.upper() _available_loglevels = ", ".join(logging.getLevelNamesMapping().keys()) if value not in logging.getLevelNamesMapping().keys(): raise ValueError( "logLevel can only contain one of these values: " f"{_available_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: _testmode = self._get_single_text_value_from_xpath('testMode') value = _testmode.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 = [] _trigger_elements = self._get_multiple_elements_by_xpath('triggers/trigger') for trigger_elem in _trigger_elements: name = self._get_single_text_value_from_xpath('name', trigger_elem) interval_in_seconds = self._get_single_text_value_from_xpath('intervalInSeconds', trigger_elem) governor = self._get_single_text_value_from_xpath('governor', trigger_elem) try: custom_config_elem = self._get_single_element_by_xpath('customConfig', trigger_elem) custom_config = xmltodict.parse(etree.tostring(custom_config_elem))['customConfig'] except AssertionError: self.log.debug("Trigger %s has no custom config", name) custom_config = OrderedDict({}) _new_trigger = TriggerTuple( name=name, interval_in_seconds=int(interval_in_seconds), governor=governor, custom_config=custom_config ) 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 def _get_single_text_value_from_xpath(self, xpath_str: str, element: Optional[etree._Element] = None) -> str: _elem = self._get_single_element_by_xpath(xpath_str, element) value = _elem.text assert isinstance(value, str) return value def _get_single_element_by_xpath(self, xpath_str: str, element: Optional[etree._Element] = None) -> etree._Element: if element is None: element = self.app_root _list_of_elements = element.xpath(xpath_str) assert isinstance(_list_of_elements, list) and len(_list_of_elements) == 1, f"config section {xpath_str} not found" _elem = _list_of_elements.pop() assert isinstance(_elem, etree._Element) return _elem def _get_multiple_elements_by_xpath(self, xpath_str: str, element: Optional[etree._Element] = None) -> list[etree._Element]: if element is None: element = self.app_root _list_of_elements = element.xpath(xpath_str) assert isinstance(_list_of_elements, list) and len(_list_of_elements) >= 1, f"config section {xpath_str} not found" assert all(isinstance(_elem, etree._Element) for _elem in _list_of_elements) return _list_of_elements # type: ignore