log.py code improvements

cpu_governor_auto_adjust.xml

@@ -5,19 +5,19 @@
     <defaultGovernor>powersave</defaultGovernor>
     <statisticsIntervalInSeconds>5</statisticsIntervalInSeconds>
     <triggers>
-        <trigger>
+        <!--trigger>
             <name>test_trigger1</name>
             <logLevel>Info</logLevel>
             <intervalInSeconds>1</intervalInSeconds>
             <governor>ondemand</governor>
-        </trigger>
+        </trigger-->
-        <trigger>
+        <!--trigger>
             <name>test_trigger2</name>
             <logLevel>Info</logLevel>
             <intervalInSeconds>3</intervalInSeconds>
             <governor>conservative</governor>
-        </trigger>
+        </trigger-->
-        <trigger>
+        <!--trigger>
             <name>roon</name>
             <logLevel>Debug</logLevel>
             <intervalInSeconds>1</intervalInSeconds>
@@ -30,8 +30,8 @@
                 <coreIdFilePath>/home/martin/dev/cpu_governor_auto_adjust/cpu_governor_auto_adjust/triggers/.data/core_id_file</coreIdFilePath>
                 <tokenFilePath>/home/martin/dev/cpu_governor_auto_adjust/cpu_governor_auto_adjust/triggers/.data/token_file</tokenFilePath>
             </customConfig>
-        </trigger>
+        </trigger-->
-        <trigger>
+        <!--trigger>
             <name>wakeup</name>
             <logLevel>Info</logLevel>
             <intervalInSeconds>5</intervalInSeconds>
@@ -40,8 +40,8 @@
                 <startTime>7:00</startTime>
                 <endTime>23:59</endTime>
             </customConfig>
-        </trigger>
+        </trigger-->
-        <trigger>
+        <!--trigger>
             <name>cpu_load</name>
             <logLevel>Info</logLevel>
             <intervalInSeconds>1</intervalInSeconds>
@@ -54,7 +54,7 @@
                 <fiveMinuteLowThreshold>0.2</fiveMinuteLowThreshold>
                 <tenMinuteLowThreshold>0.1</tenMinuteLowThreshold>
             </customConfig>
-        </trigger>
+        </trigger-->
         <trigger>
             <name>roon_arc</name>
             <logLevel>Info</logLevel>
@@ -72,5 +72,18 @@
                 </triggerString>
             </customConfig>
         </trigger>
+        <trigger>
+            <name>roon_client</name>
+            <logLevel>Info</logLevel>
+            <intervalInSeconds>1</intervalInSeconds>
+            <governor>performance</governor>
+            <customConfig>
+                <logFile>/home/martin/dev/roon_arc/RoonServer_log.txt</logFile>
+                <timeoutInMinutes>1</timeoutInMinutes>
+                <triggerString>
+                    <string>GetImageData</string>
+                </triggerString>
+            </customConfig>
+        </trigger>
     </triggers>
 </cpuGovernorAutoAdjust>

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:

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:

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:
+        ret_val = self.json_data.get('string', None)
-            return self.json_data['string']
+        if ret_val is not None and isinstance(ret_val, str):
-        else:
+            return ret_val
-            raise ValueError("TriggerString must contain 'string'")
+        raise ValueError("TriggerString must contain 'string'")
         
     @cached_property
     def substrings(self) -> list[str]:
-        _substrings = []
+        ret_val: list[str] = []
-        if 'substrings' in self.json_data:
+        
-            if isinstance(self.json_data['substrings']['substring'], list):
+        _substrings = self.json_data.get('substrings', None)
-                _substrings.extend(self.json_data['substrings'])
+        if not _substrings:
-            elif isinstance(self.json_data['substrings']['substring'], str):
+            return ret_val
-                _substrings.append(self.json_data['substrings']['substring'])
+        
-            else:
+        if not isinstance(_substrings, dict):
-                raise ValueError("substrings must be a list or a string")
+            raise ValueError("substrings must be a dict with 'substring' and 'condition'")
-        return _substrings
+        
+        _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._file: TextIO | None = None
-        self._current_inode = 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 = []
+        ret_val: list[TriggerString] = []
-        if 'triggerString' not in self.config.custom_config:
+        trigger_string = self.config.custom_config.get('triggerString', None)
-            return _trigger_strings
+        if trigger_string is None:
-        elif isinstance(self.config.custom_config['triggerString'], dict):
+            return ret_val
-            _trigger_strings.append(TriggerString(self.config.custom_config['triggerString']))
+        
-        elif isinstance(self.config.custom_config['triggerString'], list):
+        if isinstance(trigger_string, dict):
-            for trigger_string in self.config.custom_config['triggerString']:
+            ret_val.append(TriggerString(trigger_string))
-                _trigger_strings.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()

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "cpu_governor_auto_adjust"
-version = "1.1.0"
+version = "1.2.0"
-description = "This application has been developed to automatically change cpu governor based on certain triggers."
+description = "CPU Governor Auto Adjust based on Roon Activity and System Load"
 authors = [
     { name = "Martin Reurekas", email = "martin@semrks.nl" }
 ]