fixed many mypy issues
This commit is contained in:
@@ -22,21 +22,27 @@ class Config:
|
||||
self.config = ArgumentsParser().parser.config
|
||||
|
||||
@cached_property
|
||||
def root(self) -> etree.ElementTree:
|
||||
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()
|
||||
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:
|
||||
value = self.app_root.xpath('logLevel').pop().text.upper()
|
||||
_loglevels = ", ".join(logging.getLevelNamesMapping().keys())
|
||||
_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"{_loglevels}"
|
||||
f"{_available_loglevels}"
|
||||
)
|
||||
return value
|
||||
|
||||
@@ -46,7 +52,8 @@ class Config:
|
||||
|
||||
@cached_property
|
||||
def testmode(self) -> bool:
|
||||
value = self.app_root.xpath('testMode').pop().text.lower()
|
||||
_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']:
|
||||
@@ -56,11 +63,11 @@ class Config:
|
||||
@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
|
||||
_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)
|
||||
_new_trigger = TriggerTuple(
|
||||
name=name,
|
||||
interval_in_seconds=int(interval_in_seconds),
|
||||
@@ -79,3 +86,26 @@ class Config:
|
||||
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
|
||||
|
||||
@@ -14,7 +14,7 @@ def getLogger(name: str = 'root', loglevel: str = 'INFO') -> Logger:
|
||||
# if logger 'name' does not already exist, create it and attach handlers
|
||||
else:
|
||||
# set logLevel to loglevel or to INFO if requested level is incorrect
|
||||
loglevel = getattr(logging, loglevel.upper(), logging.INFO)
|
||||
loglevel = getattr(logging, loglevel.upper(), "INFO")
|
||||
logger.setLevel(loglevel)
|
||||
if loglevel == logging.DEBUG:
|
||||
fmt = '%(asctime)s %(levelname)-8s %(message)s (%(filename)s)'
|
||||
|
||||
+3
-1
@@ -25,4 +25,6 @@ class Trigger(AppClass):
|
||||
|
||||
def run(self) -> bool:
|
||||
# return True if Trigger is valid
|
||||
return False
|
||||
raise TriggerImportError(
|
||||
"the Trigger class can't used directly, but must be inherited in a trigger specific class"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user