trimmed spaces

cpu_governor_auto_adjust/config.py

@@ -16,7 +16,7 @@ class TriggerTuple(NamedTuple):
     interval_in_seconds: int
     governor: str
     custom_config: OrderedDict[str, Any]
-    
+
     def __repr__(self) -> str:
         return f"name: {self.name}, loglevel: {self.loglevel}, interval: {self.interval_in_seconds}, governor: {self.governor}, custom_config: {self.custom_config}"
 
@@ -25,26 +25,26 @@ class Config:
     def __init__(self, _basepath: Optional[Path] = None) -> None:
         self._basepath = _basepath
         self.config = ArgumentsParser().parser.config
-        
+
     @cached_property
     def basepath(self) -> Path:
         if self._basepath is None:
             return Path.cwd()
         return self._basepath
-    
+
     @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')
@@ -56,11 +56,11 @@ class Config:
                 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')
@@ -70,7 +70,7 @@ class 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 statistics_interval_in_seconds(self) -> int:
         _statistics_interval = self._get_single_text_value_from_xpath('statisticsIntervalInSeconds')
@@ -78,12 +78,12 @@ class Config:
         if value < 1:
             raise ValueError("statisticsIntervalInSeconds must be greater than 0")
         return value
-    
+
     @cached_property
     def default_governor(self) -> str:
         _default_governor = self._get_single_text_value_from_xpath('defaultGovernor')
         return _default_governor
-    
+
     @cached_property
     def triggertuples(self) -> list[TriggerTuple]:
         ret_val = []
@@ -108,9 +108,9 @@ class 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)
@@ -119,13 +119,13 @@ 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
@@ -134,7 +134,7 @@ class Config:
         _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

cpu_governor_auto_adjust/cpu_governor_auto_adjust.py

@@ -17,13 +17,14 @@ async def _main() -> None:
     log = getLogger('main', loglevel=config.loglevel.upper())
     if config.testmode:
         log.warning("starting in testmode, cpu adjustments have been disabled")
-    
+
     scheduler = TriggerScheduler(config)
     for triggertuple in config.triggertuples:
         _trigger = trigger_mapping[triggertuple.name](config)
         await scheduler.schedule_task(scheduler.start_trigger, _trigger)
     await scheduler.run()
-    
+
 
 if __name__ == "__main__":
     main()
+

cpu_governor_auto_adjust/governor.py

@@ -8,18 +8,18 @@ from time import monotonic_ns as monotonic
 from cpu_governor_auto_adjust.helper import format_time_ns
 import asyncio
 
-        
+
 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("performance", 0),
+    Governor("schedutil", 1),
+    Governor("ondemand", 2),
+    Governor("conservative", 3),
+    Governor("userspace", 4),
     Governor("powersave", 5)
 )
 
@@ -28,10 +28,10 @@ class DummyCpuFreq:
     def __init__(self) -> None:
         self.available_governors = [g.name for g in _governor_list]
         self._governor = "performance"
-    
+
     def get_governors(self) -> dict[str, str]:
         return {g.name: g.name for g in _governor_list}
-    
+
     def set_governors(self, governor_name: str) -> None:
         self._governor = governor_name
 
@@ -44,14 +44,14 @@ class GovernorControl(AppClass):
         self._timestamp_last_statistics = monotonic()
         self._governor_statistics = self._statistics_dict()
         self._last_statistic_update = monotonic()
-        
+
     def _statistics_dict(self) -> dict[str, int]:
         ret_dict = {}
         for gov in _governor_list:
             if gov.name in self._cpufreq.available_governors:
                 ret_dict[gov.name] = 0
         return ret_dict
-    
+
     def _update_statistics(self) -> None:
         _now = monotonic()
         self._governor_statistics[self.current_governor.name] += _now - self._last_statistic_update
@@ -72,7 +72,7 @@ class GovernorControl(AppClass):
         except cpufreq.CPUFreqErrorInit:
             self.log.warning("cpu architecture has no governor support")
             return DummyCpuFreq()
-    
+
     @property
     def current_governor(self) -> Governor:
         if self._config.testmode:
@@ -89,26 +89,26 @@ class GovernorControl(AppClass):
                 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.info("setting cpu governor to %s", governor_name)
-        
+
         if self._config.testmode:
             self.log.warning("application is running in testmode, cpu governor not set")
             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)
-        
+
     async def run(self) -> None:
         self.log.debug("running governor control")
         while True:
@@ -117,9 +117,9 @@ class GovernorControl(AppClass):
             if self._timestamp_last_statistics + (int(self._config.statistics_interval_in_seconds) * 1_000_000_000) < start:
                 self.log_statistics()
                 self._timestamp_last_statistics = start
-                
+
             if self.current_governor != self.governor:
                 self.log.info("governor has changed from %s to %s", self.current_governor, self.governor)
                 self._set_governor(self.governor.name)
-                
+
             await asyncio.sleep(1)

cpu_governor_auto_adjust/helper.py

@@ -15,4 +15,4 @@ def format_time_ns(elapsed_ns):
         hours = total_seconds // 3600
         minutes = (total_seconds % 3600) // 60
         seconds = total_seconds % 60
-        return f"{hours:02}:{minutes:02}:{seconds:02}"
+        return f"{hours:02}:{minutes:02}:{seconds:02}"

cpu_governor_auto_adjust/logger.py

@@ -11,15 +11,15 @@ class LoggerFormat(NamedTuple):
     message: str = '%(message)s'
     filename: str = '%(filename)s'
     lineno: str = '%(lineno)d'
-    
+
     def _base(self, class_name: Optional[str] = None) -> str:
         if class_name is not None:
             return f"{self.time} {self.level} [{class_name}] {self.message} "
         return f"{self.time} {self.level} {self.message}"
-    
+
     def info(self, class_name: Optional[str] = None) -> str:
         return f"{self._base(class_name)}"
-    
+
     def debug(self, class_name: Optional[str] = None) -> str:
         return f"{self._base(class_name)} ({self.filename}:{self.lineno})"
 

cpu_governor_auto_adjust/mapping.py

@@ -1,10 +1,10 @@
 from types import MappingProxyType
 from cpu_governor_auto_adjust.triggers import (
-    TestTrigger1, 
-    TestTrigger2, 
-    RoonTrigger, 
-    WakeupTrigger, 
-    CpuLoadTrigger, 
+    TestTrigger1,
+    TestTrigger2,
+    RoonTrigger,
+    WakeupTrigger,
+    CpuLoadTrigger,
     RoonArcTrigger,
     RoonClientTrigger,
 )

cpu_governor_auto_adjust/schedule.py

@@ -16,11 +16,11 @@ class TriggerScheduler(AppClass):
         super().__init__(_config)
         self.running_triggers: list[Trigger] = []
         self.loop = asyncio.get_event_loop()
-        
+
     @cached_property
     def governor_control(self) -> GovernorControl:
         return GovernorControl(self._config)
-    
+
     def establish_preferred_governor(self) -> Governor:
         self.log.debug('establishing preferred governor, default governor: %s', self._config.default_governor)
         preferred_governor: Governor = [gov for gov in _governor_list if gov.name == self._config.default_governor].pop()
@@ -31,7 +31,7 @@ class TriggerScheduler(AppClass):
                     preferred_governor = trigger.governor
         self.log.debug('preferred governor: %s', preferred_governor)
         return preferred_governor
-            
+
     async def start_trigger(self, _trigger: Trigger) -> None:
         """Start a new trigger."""
         self.loop.create_task(_trigger.run())
@@ -45,21 +45,21 @@ class TriggerScheduler(AppClass):
             task.cancel()
         await asyncio.gather(*tasks, return_exceptions=True)
         self.loop.stop()
-        
+
     async def schedule_task(self, task: Callable, *args: Any) -> None:
         self.log.debug("scheduling task: %s", task.__name__)
         start = monotonic()
         await self.loop.create_task(task(*args))
-        self.log.debug("task %s finished in %s", task.__name__, format_time_ns(monotonic() - start))  
+        self.log.debug("task %s finished in %s", task.__name__, format_time_ns(monotonic() - start))
 
     async def run(self) -> None:
         """Run the scheduler and keep it alive until stopped."""
-                
+
         def signal_handler(sig):
             self.log.info("received signal: %s", sig)
             self.log.info("Exiting, stopping all triggers")
             self.loop.create_task(self.stop_triggers())
-        
+
         for sig in [signal.SIGINT, signal.SIGTERM]:
             self.loop.add_signal_handler(sig, partial(signal_handler, sig=signal.SIGINT))
 

cpu_governor_auto_adjust/trigger.py

@@ -19,10 +19,10 @@ class Trigger(AppClass):
         self._minimum_consumed_time = 10000000000
         self._maximum_consumed_time = 0
         self.runs = 0
-        
+
     def __hash__(self):
         return super().__hash__(hash(self.name))
-    
+
     @property
     def consumed_time(self) -> dict[str, str]:
         _dict = {
@@ -35,15 +35,15 @@ class Trigger(AppClass):
             return _dict
         _dict["average"] = format_time_ns(self._total_consumed_time / self.runs)
         return _dict
-    
+
     def log_statistics(self):
         class_vars = {}
-        
+
         start = monotonic()
-        for base in reversed(self.__class__.__mro__[:-1]):  
+        for base in reversed(self.__class__.__mro__[:-1]):
             if hasattr(base, '__dict__'):
                 for key, value in base.__dict__.items():
-                    # Ignore methods and private attributes 
+                    # Ignore methods and private attributes
                     if not key.startswith("__") and not key.startswith("_") and not callable(value):
                         if isinstance(value, property):
                             # If it's a property, call the getter method
@@ -59,13 +59,13 @@ class Trigger(AppClass):
                 if not isinstance(value, (int, float, Governor)):
                     value = str(value)
                 class_vars[key] = value
-                
+
         self.log.info("info and statistics [time taken: %s]: %s", format_time_ns(monotonic() - start),  class_vars)
-    
+
     @property
     def trigger_state(self) -> str:
         return "active" if self._active else "not active"
-    
+
     @property
     def name(self) -> str:
         if self.__module__ == 'trigger':
@@ -74,24 +74,24 @@ class Trigger(AppClass):
             )
         _name =  self.__module__.split('.').pop()
         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 trigger_code(self) -> None:
         # this is the main function of the trigger class and can be used to execute a callback function
-        
+
         raise TriggerImportError(
             "the Trigger class can't used directly, but must be inherited in a trigger specific class"
         )
@@ -111,4 +111,3 @@ class Trigger(AppClass):
             self.runs += 1
             self.log.debug("trigger state: %s, finished in %s", self.trigger_state, format_time_ns(end - start))
             await asyncio.sleep(self.config.interval_in_seconds)
-    

install_scripts/cpu-governor-auto-adjust.service

@@ -15,4 +15,4 @@ StandardOutput=append:/var/log/cpu_governor_auto_adjust.log
 StandardError=append:/var/log/cpu_governor_auto_adjust.log
 
 [Install]
-WantedBy=multi-user.target
+WantedBy=multi-user.target

install_scripts/start.sh

@@ -3,4 +3,4 @@ export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
 export TZ='Europe/Amsterdam'
 source /opt/cpu_governor_auto_adjust/.venv/bin/activate
 pip install -U cpu-governor-auto-adjust
-exec cpu_governor_auto_adjust -c /opt/cpu_governor_auto_adjust/cpu_governor_auto_adjust.xml
+exec cpu_governor_auto_adjust -c /opt/cpu_governor_auto_adjust/cpu_governor_auto_adjust.xml

pyproject.toml

@@ -18,8 +18,8 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
-    "pytest", 
-    "black", 
+    "pytest",
+    "black",
     "mypy"
 ]