From c29bb3a82a5759a55ad527caa76579400fd19f24 Mon Sep 17 00:00:00 2001 From: Martin Reurekas Date: Mon, 29 Dec 2025 23:24:10 +0100 Subject: [PATCH] log.py code improvements --- cpu_governor_auto_adjust.xml | 33 ++++++--- cpu_governor_auto_adjust/governor.py | 2 +- cpu_governor_auto_adjust/logger.py | 2 +- cpu_governor_auto_adjust/triggers/log.py | 87 +++++++++++++++--------- pyproject.toml | 4 +- 5 files changed, 83 insertions(+), 45 deletions(-) diff --git a/cpu_governor_auto_adjust.xml b/cpu_governor_auto_adjust.xml index 5cd4aa1..f9d7ace 100644 --- a/cpu_governor_auto_adjust.xml +++ b/cpu_governor_auto_adjust.xml @@ -5,19 +5,19 @@ powersave 5 - + + + + + roon_arc Info @@ -72,5 +72,18 @@ + + roon_client + Info + 1 + performance + + /home/martin/dev/roon_arc/RoonServer_log.txt + 1 + + GetImageData + + + diff --git a/cpu_governor_auto_adjust/governor.py b/cpu_governor_auto_adjust/governor.py index f4a66d1..44ff798 100644 --- a/cpu_governor_auto_adjust/governor.py +++ b/cpu_governor_auto_adjust/governor.py @@ -45,7 +45,7 @@ class GovernorControl(AppClass): self._governor_statistics = self._statistics_dict() self._last_statistic_update = monotonic() - def _statistics_dict(self) -> dict[str, str]: + def _statistics_dict(self) -> dict[str, int]: ret_dict = {} for gov in _governor_list: if gov.name in self._cpufreq.available_governors: diff --git a/cpu_governor_auto_adjust/logger.py b/cpu_governor_auto_adjust/logger.py index ecb7182..6a15bcf 100644 --- a/cpu_governor_auto_adjust/logger.py +++ b/cpu_governor_auto_adjust/logger.py @@ -10,7 +10,7 @@ class LoggerFormat(NamedTuple): level: str = '%(levelname)-8s' message: str = '%(message)s' filename: str = '%(filename)s' - lineno: int = '%(lineno)d' + lineno: str = '%(lineno)d' def _base(self, class_name: Optional[str] = None) -> str: if class_name is not None: diff --git a/cpu_governor_auto_adjust/triggers/log.py b/cpu_governor_auto_adjust/triggers/log.py index acebeb8..14af241 100644 --- a/cpu_governor_auto_adjust/triggers/log.py +++ b/cpu_governor_auto_adjust/triggers/log.py @@ -3,41 +3,58 @@ from cpu_governor_auto_adjust.config import Config from datetime import datetime, timedelta from functools import cached_property from dataclasses import dataclass +from typing import TextIO import os +JSONStrValue = str | list[str] | dict[str, "JSONStrValue"] @dataclass class TriggerString: - json_data: dict[str, str] + json_data: dict[str, JSONStrValue] @cached_property def string(self) -> str: - if 'string' in self.json_data: - return self.json_data['string'] - else: - raise ValueError("TriggerString must contain 'string'") + ret_val = self.json_data.get('string', None) + if ret_val is not None and isinstance(ret_val, str): + return ret_val + raise ValueError("TriggerString must contain 'string'") @cached_property def substrings(self) -> list[str]: - _substrings = [] - if 'substrings' in self.json_data: - if isinstance(self.json_data['substrings']['substring'], list): - _substrings.extend(self.json_data['substrings']) - elif isinstance(self.json_data['substrings']['substring'], str): - _substrings.append(self.json_data['substrings']['substring']) - else: - raise ValueError("substrings must be a list or a string") - return _substrings + ret_val: list[str] = [] + + _substrings = self.json_data.get('substrings', None) + if not _substrings: + return ret_val + + if not isinstance(_substrings, dict): + raise ValueError("substrings must be a dict with 'substring' and 'condition'") + + _substring = _substrings.get('substring', None) + if isinstance(_substring, list): + ret_val.extend(_substring) + elif isinstance(_substring, str): + ret_val.append(_substring) + else: + raise ValueError("substring must be a list or a string") + + return ret_val @cached_property def condition(self) -> str: - if not self.substrings: + substrings = self.json_data.get('substrings', None) + if not substrings: return 'OR' - if not 'condition' in self.json_data['substrings']: + + if not isinstance(substrings, dict): + raise ValueError("substrings must be a dict with 'substring' and 'condition'") + + _condition = substrings.get('condition', None) + if not _condition: raise ValueError("substrings must contain 'condition'") - if not self.json_data['substrings']['condition'] in ['OR', 'AND']: + if not _condition in ['OR', 'AND']: raise ValueError("substrings condition must be 'OR' or 'AND'") - return self.json_data['substrings']['condition'] + return str(_condition) def __repr__(self) -> str: return f"TriggerString(string={self.string}, substrings={self.substrings}, condition={self.condition})" @@ -47,30 +64,38 @@ class LogTrigger(Trigger): super().__init__(_config) self.timestamp_last_active_change = datetime.now() - timedelta(minutes=self.timeout_in_minutes + 1) self.line_count = 0 - self._file = None - self._current_inode = None + self._file: TextIO | None = None + self._current_inode: int | None = None @cached_property def file(self) -> str: - return self.config.custom_config['logFile'] + _file = self.config.custom_config.get('logFile', None) + if _file is None or not isinstance(_file, str): + raise ValueError("logFile must be set in custom_config and be a string") + return _file @cached_property def timeout_in_minutes(self) -> float: - return float(self.config.custom_config['timeoutInMinutes']) + _timeout = self.config.custom_config.get('timeoutInMinutes', None) + if _timeout is None: + raise ValueError("timeoutInMinutes must be set in custom_config") + return float(_timeout) @cached_property def trigger_strings(self) -> list[TriggerString]: - _trigger_strings = [] - if 'triggerString' not in self.config.custom_config: - return _trigger_strings - elif isinstance(self.config.custom_config['triggerString'], dict): - _trigger_strings.append(TriggerString(self.config.custom_config['triggerString'])) - elif isinstance(self.config.custom_config['triggerString'], list): - for trigger_string in self.config.custom_config['triggerString']: - _trigger_strings.append(TriggerString(trigger_string)) + ret_val: list[TriggerString] = [] + trigger_string = self.config.custom_config.get('triggerString', None) + if trigger_string is None: + return ret_val + + if isinstance(trigger_string, dict): + ret_val.append(TriggerString(trigger_string)) + elif isinstance(trigger_string, list): + for ts in trigger_string: + ret_val.append(TriggerString(ts)) else: raise ValueError("triggerString must be a dict or a list of dicts") - return _trigger_strings + return ret_val def set_active(self) -> None: self.timestamp_last_active_change = datetime.now() diff --git a/pyproject.toml b/pyproject.toml index 637c727..67988d6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "cpu_governor_auto_adjust" -version = "1.1.0" -description = "This application has been developed to automatically change cpu governor based on certain triggers." +version = "1.2.0" +description = "CPU Governor Auto Adjust based on Roon Activity and System Load" authors = [ { name = "Martin Reurekas", email = "martin@semrks.nl" } ]