From f3125c3803ad307d5c036114d888842ee2bdbaa7 Mon Sep 17 00:00:00 2001 From: Martin Reurekas Date: Sun, 13 Sep 2026 23:13:48 +0200 Subject: [PATCH] trimmed spaces --- cpu_governor_auto_adjust/config.py | 32 +++++++------- .../cpu_governor_auto_adjust.py | 5 ++- cpu_governor_auto_adjust/governor.py | 42 +++++++++---------- cpu_governor_auto_adjust/helper.py | 2 +- cpu_governor_auto_adjust/logger.py | 6 +-- cpu_governor_auto_adjust/mapping.py | 10 ++--- cpu_governor_auto_adjust/schedule.py | 14 +++---- cpu_governor_auto_adjust/trigger.py | 27 ++++++------ .../cpu-governor-auto-adjust.service | 2 +- install_scripts/start.sh | 2 +- pyproject.toml | 4 +- 11 files changed, 73 insertions(+), 73 deletions(-) diff --git a/cpu_governor_auto_adjust/config.py b/cpu_governor_auto_adjust/config.py index bf35872..94ab2e7 100644 --- a/cpu_governor_auto_adjust/config.py +++ b/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 diff --git a/cpu_governor_auto_adjust/cpu_governor_auto_adjust.py b/cpu_governor_auto_adjust/cpu_governor_auto_adjust.py index 0352a9f..6c7c7bf 100755 --- a/cpu_governor_auto_adjust/cpu_governor_auto_adjust.py +++ b/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() + diff --git a/cpu_governor_auto_adjust/governor.py b/cpu_governor_auto_adjust/governor.py index 44ff798..ba5a3a9 100644 --- a/cpu_governor_auto_adjust/governor.py +++ b/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) diff --git a/cpu_governor_auto_adjust/helper.py b/cpu_governor_auto_adjust/helper.py index 79e25b2..4fa1220 100644 --- a/cpu_governor_auto_adjust/helper.py +++ b/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}" \ No newline at end of file + return f"{hours:02}:{minutes:02}:{seconds:02}" diff --git a/cpu_governor_auto_adjust/logger.py b/cpu_governor_auto_adjust/logger.py index 6a15bcf..dd57af3 100644 --- a/cpu_governor_auto_adjust/logger.py +++ b/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})" diff --git a/cpu_governor_auto_adjust/mapping.py b/cpu_governor_auto_adjust/mapping.py index 66ab638..dfbf7e9 100644 --- a/cpu_governor_auto_adjust/mapping.py +++ b/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, ) diff --git a/cpu_governor_auto_adjust/schedule.py b/cpu_governor_auto_adjust/schedule.py index a352519..8b394c4 100644 --- a/cpu_governor_auto_adjust/schedule.py +++ b/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)) diff --git a/cpu_governor_auto_adjust/trigger.py b/cpu_governor_auto_adjust/trigger.py index 90cca7c..b9c8490 100644 --- a/cpu_governor_auto_adjust/trigger.py +++ b/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) - \ No newline at end of file diff --git a/install_scripts/cpu-governor-auto-adjust.service b/install_scripts/cpu-governor-auto-adjust.service index 35ae74b..cd55f94 100644 --- a/install_scripts/cpu-governor-auto-adjust.service +++ b/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 \ No newline at end of file +WantedBy=multi-user.target diff --git a/install_scripts/start.sh b/install_scripts/start.sh index aeb931e..333738d 100644 --- a/install_scripts/start.sh +++ b/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 \ No newline at end of file +exec cpu_governor_auto_adjust -c /opt/cpu_governor_auto_adjust/cpu_governor_auto_adjust.xml diff --git a/pyproject.toml b/pyproject.toml index 2ccae31..364f48b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,8 +18,8 @@ dependencies = [ [project.optional-dependencies] dev = [ - "pytest", - "black", + "pytest", + "black", "mypy" ]