moved code to subdir cpu_governor_auto_adjust

cpu_governor_auto_adjust/app_class.py

@@ -0,0 +1,10 @@
+from logger import getLogger
+from cpu_governor_auto_adjust.config import Config
+
+
+class AppClass:
+    def __init__(self, _config: Config) -> None:
+        self._config = _config
+        self.log = getLogger(self.__class__.__name__, loglevel=_config.loglevel.upper())
+        self.log.info(f'initializing {self.__class__.__name__}')
+    

cpu_governor_auto_adjust/arguments.py

@@ -0,0 +1,19 @@
+import argparse
+from argparse import Namespace
+from functools import cached_property
+
+class ArgumentsParser:
+    @cached_property
+    def parser(self) -> Namespace:
+        parser = argparse.ArgumentParser(
+                prog=__file__,
+                description='Trigger based CPU governor auto adjust program.',
+                epilog='Happy CPU controlling'
+        )
+        parser.add_argument(
+            '-c', '--config',
+            type=str,
+            required=False,
+            default='cpu_governor_auto_adjust.xml',
+        )
+        return parser.parse_args()

cpu_governor_auto_adjust/config.py

@@ -0,0 +1,123 @@
+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

cpu_governor_auto_adjust/cpu_governor_auto_adjust.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+import asyncio
+from pathlib import Path
+from logger import getLogger
+from mapping import trigger_mapping
+from config import Config
+from governor import GovernorControl
+from schedule import TriggerScheduler
+
+
+async def main() -> None:
+    basepath = Path(__file__).parent.resolve()
+    config = Config(basepath)
+    log = getLogger('main', loglevel=config.loglevel.upper())
+    if config.testmode:
+        log.warning("starting in testmode, cpu adjustments have been disabled")
+    
+    governor = GovernorControl(config)
+    scheduler = TriggerScheduler(config)
+    for triggertuple in config.triggertuples:
+        _trigger = trigger_mapping[triggertuple.name](config)
+        scheduler.start_trigger(_trigger)
+        
+    await scheduler.run()
+    
+
+if __name__ == "__main__":
+    asyncio.run(main())

cpu_governor_auto_adjust/exceptions.py

@@ -0,0 +1,10 @@
+class TriggerImportError(Exception):
+    pass
+
+
+class MissingConfig(Exception):
+    pass
+
+
+class GovernorNotFound(Exception):
+    pass

cpu_governor_auto_adjust/governor.py

@@ -0,0 +1,77 @@
+from app_class import AppClass
+from config import Config
+from cpufreq import cpuFreq, cpufreq  # type: ignore
+from functools import cached_property
+from exceptions import GovernorNotFound
+from typing import Optional, NamedTuple
+
+        
+class Governor(NamedTuple):
+    name: str
+    priority: int
+    
+    
+_governor_list: tuple[Governor, Governor, Governor, Governor, Governor, Governor] = (
+    Governor("performance", 0), 
+    Governor("schedutil", 1), 
+    Governor("ondemand", 2), 
+    Governor("conservative", 3), 
+    Governor("userspace", 4), 
+    Governor("powersave", 5)
+)
+
+
+class GovernorControl(AppClass):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)
+        self.log.info("current governor is: %s", self.current_governor)
+
+    @cached_property
+    def _cpufreq(self) -> Optional[cpuFreq]:
+        try:
+            return cpuFreq()
+        except cpufreq.CPUFreqErrorInit:
+            self.log.warning("cpu architecture has no governor support")
+        return None
+    
+    @property
+    def current_governor(self) -> Optional[Governor]:
+        if self._cpufreq is None:
+            return None
+        _governor = set(self._cpufreq.get_governors().values())
+        _governor_to_return = self._establish_governor(list(_governor)[0])
+        if len(_governor) == 0:
+            raise GovernorNotFound("Unable to retrieve current governor")
+        if len(_governor) > 1:
+            self.log.error("multiple governors have been set, which is not expected.")
+            # returning governor with highest priority based on performance
+            for gov in _governor:
+                t_gov = self._establish_governor(gov)
+                if t_gov.priority < _governor_to_return.priority:
+                    _governor_to_return == t_gov
+        return _governor_to_return
+    
+    def _establish_governor(self, governor_name: str) -> Governor:
+        governor = next((g for g in _governor_list if g.name == governor_name), None)
+        if governor is None:
+            raise RuntimeError("could not establish governor")
+        return governor
+                  
+    def set_governor(self, governor_name: str) -> None:
+        self.log.debug(
+            "setting cpu governor to %s by using command self._cpufreq.set_governors(%s)", 
+            governor_name, governor_name
+        )
+        
+        if self._config.testmode:
+            self.log.warning("application is running in testmode, cpu governor not set")
+            return None
+        
+        if self._cpufreq is None:
+            return None
+        
+        if governor_name not in self._cpufreq.available_governors:
+            self.log.error("governor %s not supported by cpu", governor_name)
+            return None
+        
+        self._cpufreq.set_governors(governor_name)

cpu_governor_auto_adjust/logger.py

@@ -0,0 +1,34 @@
+import logging
+from logging import Logger
+import os
+import sys
+
+
+def getLogger(name: str = 'root', loglevel: str = 'INFO') -> Logger:
+    logger = logging.getLogger(name)
+
+    # if logger 'name' already exists, return it to avoid logging duplicate
+    # messages by attaching multiple handlers of the same type
+    if logger.handlers:
+        return 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())
+        logger.setLevel(loglevel)
+        if loglevel == logging.DEBUG:
+            fmt = '%(asctime)s %(levelname)-8s %(message)s (%(filename)s:%(lineno)d)'
+        else:
+            fmt = '%(asctime)s %(levelname)-8s %(message)s'
+        fmt_date = '%Y-%m-%dT%T'
+        formatter = logging.Formatter(fmt, fmt_date)
+        handler = logging.StreamHandler()
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+
+        if logger.name == 'root':
+            logger.warning(
+                'Running: %s %s',
+                os.path.basename(sys.argv[0]),
+                ' '.join(sys.argv[1:]))
+    return logger

cpu_governor_auto_adjust/mapping.py

@@ -0,0 +1,9 @@
+from types import MappingProxyType
+from triggers import TestTrigger1, TestTrigger2, RoonTrigger
+
+_trigger_mapping = {
+    'test_trigger1': TestTrigger1,
+    'test_trigger2': TestTrigger2,
+    'roon': RoonTrigger
+}
+trigger_mapping = MappingProxyType(_trigger_mapping)

cpu_governor_auto_adjust/schedule.py

@@ -0,0 +1,67 @@
+from app_class import AppClass
+from config import Config
+from trigger import Trigger
+from governor import GovernorControl, Governor, _governor_list
+import asyncio
+from asyncio import Task
+from functools import cached_property
+
+
+class TriggerScheduler(AppClass):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)
+        self.tasks: list[Task] = []
+        self.running_triggers: list[Trigger] = []
+        
+    @cached_property
+    def governor_control(self) -> GovernorControl:
+        return GovernorControl(self._config)
+    
+    def establish_preferred_governor(self) -> str:
+        preferred_governor: Governor = [gov for gov in _governor_list if gov.name == 'powersave'].pop()
+        for trigger in self.running_triggers:
+            self.log.debug('trigger %s, state: %s, preferred governor: %s', trigger.name, trigger.trigger_state, trigger.governor.name)
+            if trigger.active:
+                if trigger.governor.priority < preferred_governor.priority:
+                    preferred_governor = trigger.governor
+        self.log.debug('preferred governor: %s', preferred_governor.name)
+        return preferred_governor.name
+        
+    async def callback_trigger(self, _trigger: Trigger) -> None:
+        """Run a callback_trigger with a specific name and check its status at a given interval."""
+        await _trigger.async_run()
+        while True:
+            self.log.debug("Checking state of trigger %s, state: %s", _trigger.name, _trigger.trigger_state)
+            await asyncio.sleep(_trigger.config.interval_in_seconds)
+            
+    async def run_once_trigger(self, _trigger: Trigger) -> None:
+        """Run a trigger with a specific name at a given interval."""
+        while True:
+            _trigger.run()
+            self.log.debug("Checking state of trigger %s, state: %s", _trigger.name, _trigger.trigger_state)
+            await asyncio.sleep(_trigger.config.interval_in_seconds)
+
+    def start_trigger(self, _trigger: Trigger) -> None:
+        """Start a new trigger."""
+        if _trigger.config.type == "callback":
+            task = asyncio.create_task(self.callback_trigger(_trigger))
+        else:
+            task = asyncio.create_task(self.run_once_trigger(_trigger))
+        self.running_triggers.append(_trigger)
+        self.tasks.append(task)
+
+    async def stop_triggers(self) -> None:
+        """Stop all triggers."""
+        for task in self.tasks:
+            task.cancel()
+        await asyncio.gather(*self.tasks, return_exceptions=True)
+
+    async def run(self) -> None:
+        """Run the scheduler and keep it alive until stopped."""
+        try:
+            while True:
+                self.establish_preferred_governor()
+                await asyncio.sleep(1)  # Keep the main function alive
+        except asyncio.exceptions.CancelledError:
+            self.log.info("Exiting, stopping all triggers")
+            await self.stop_triggers()

cpu_governor_auto_adjust/trigger.py

@@ -0,0 +1,57 @@
+from app_class import AppClass
+from config import Config, TriggerTuple
+from exceptions import MissingConfig, TriggerImportError
+from governor import Governor, _governor_list
+from functools import cached_property
+
+
+class Trigger(AppClass):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)
+        self.active: bool = False
+        
+    def __hash__(self):
+        return super().__hash__(hash(self.name))
+    
+    @property
+    def trigger_state(self) -> str:
+        return "active" if self.active else "not active"
+    
+    @property
+    def name(self) -> str:
+        if self.__module__ == 'trigger':
+            raise TriggerImportError(
+                "the Trigger class can't used directly, but must be inherited in a trigger specific class"
+            )
+        _ , _name =  self.__module__.split('.')
+        return _name
+    
+    @property
+    def config(self) -> TriggerTuple:
+        _config = self._config.get_trigger_by_name(self.name)
+        if _config is None:
+            raise MissingConfig(f"Trigger {self.name} hasn't been properly configured")
+        return _config
+    
+    @cached_property
+    def governor(self) -> Governor:
+        _governor = next((gov for gov in _governor_list if self.config.governor == gov.name), None)
+        if _governor is None:
+            raise ValueError("unknown governor: %s", _governor)
+        return _governor
+    
+    def run(self) -> None:
+        # this is the main function of the trigger class and has to stay active by using a while True loop
+        
+        while True:
+            raise TriggerImportError(
+                "the Trigger class can't used directly, but must be inherited in a trigger specific class"
+            )
+    
+    async def async_run(self) -> None:
+        # this is the main function of the trigger class and can be used to execute a callback function
+        
+        while True:
+            raise TriggerImportError(
+                "the Trigger class can't used directly, but must be inherited in a trigger specific class"
+            )

cpu_governor_auto_adjust/triggers/__init__.py

@@ -0,0 +1,5 @@
+__all__ = ["test_trigger1", "test_trigger2", "roon"]
+
+from .test_trigger1 import TestTrigger1
+from .test_trigger2 import TestTrigger2
+from .roon import RoonTrigger

cpu_governor_auto_adjust/triggers/roon.py

@@ -0,0 +1,91 @@
+from trigger import Trigger
+from config import Config
+from roonapi import RoonApi, RoonDiscovery  # type: ignore
+from functools import cached_property
+from typing import Any, NamedTuple, Union
+from pathlib import Path
+
+
+class RoonServer(NamedTuple):
+    ip: str
+    port: str
+
+    def __repr__(self) -> str:
+        return f"RoonServer: ip: {self.ip}, port: {self.port}"
+
+class RoonTrigger(Trigger):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)
+        self.set_status()
+        
+    def set_status(self) -> None:
+        zones_state = list()
+
+        for zone_id in self.roonapi.zones:
+            self.log.debug(self.roonapi.zones[zone_id])
+            zone = self.roonapi.zones[zone_id]
+            self.log.debug("zone_id: %s zone_info: %s" , zone_id, zone)
+            zone_name = zone['display_name']
+            zone_state = zone['state']
+            self.log.info("zone_id: %s name: %s, state: %s" , zone_id, zone_name, zone_state)
+            zones_state.append(zone_state)
+            
+        self.active = True if "playing" in zones_state else False
+
+    @cached_property
+    def appinfo(self) -> dict[str, Any]:
+        _appinfo = {
+            "extension_id": self.config.custom_config['extensionId'],
+            "display_name": self.config.custom_config['displayName'],
+            "display_version": self.config.custom_config['displayVersion'],
+            "publisher": self.config.custom_config['publisher'],
+            "email": self.config.custom_config['email'],
+        }
+        self.log.debug("appinfo: %s", _appinfo)
+        return _appinfo
+    
+    @cached_property
+    def core_id(self) -> str:
+        _core_id_path = Path(self.config.custom_config['coreIdFilePath'])
+        if not _core_id_path.exists():
+            raise FileExistsError(f'unable to open {_core_id_path}')
+        _core_id = _core_id_path.read_text()
+        self.log.debug('roon core id: %s', _core_id)
+        return _core_id
+    
+    @cached_property
+    def token(self) -> str:
+        _token_path = Path(self.config.custom_config['tokenFilePath'])
+        if not _token_path.exists():
+            raise FileExistsError(f'unable to open {_token_path}')
+        _token = _token_path.read_text()
+        self.log.debug('roon token: %s', _token)
+        return _token
+    
+    @cached_property
+    def discover(self) -> RoonDiscovery:
+        return RoonDiscovery(self.core_id)
+    
+    @cached_property
+    def server(self) -> RoonServer:
+        _server = self.discover.first()
+        assert isinstance(_server, tuple) and len(_server) == 2, "failed to discover roon server"
+        self.discover.stop()
+        roon_server = RoonServer(_server[0], _server[1])
+        self.log.info("found %s", roon_server)
+        return roon_server
+    
+    @cached_property
+    def roonapi(self) -> RoonApi:
+        _roonapi = RoonApi(self.appinfo, self.token, self.server.ip, self.server.port, True)
+        return _roonapi
+    
+    def roon_state_callback(self, event: str, changed_ids: Union[str, list[str]]) -> None:
+        """Call when something changes in roon."""
+        self.log.debug(
+            "%s event: %s changed_ids: %s" , RoonTrigger.roon_state_callback.__qualname__, event, changed_ids
+        )
+        self.set_status()
+            
+    async def async_run(self) -> None:
+        self.roonapi.register_state_callback(self.roon_state_callback, "zones_changed")

cpu_governor_auto_adjust/triggers/test_trigger1.py

@@ -0,0 +1,12 @@
+from trigger import Trigger
+from config import Config
+import random
+
+class TestTrigger1(Trigger):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)
+        
+    def run(self) -> None:
+        choices = [False, True]
+        self.log.info("run check code of %s", self.__class__.__name__)
+        self.active = random.choice(choices)

cpu_governor_auto_adjust/triggers/test_trigger2.py

@@ -0,0 +1,6 @@
+from .test_trigger1 import TestTrigger1
+from config import Config
+
+class TestTrigger2(TestTrigger1):
+    def __init__(self, _config: Config) -> None:
+        super().__init__(_config)