124 lines
5.2 KiB
Python
124 lines
5.2 KiB
Python
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
|
|
type: 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)
|
|
type = self._get_single_text_value_from_xpath('type', 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,
|
|
type=type,
|
|
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
|