added governor statistics

cpu_governor_auto_adjust/governor.py

@@ -3,7 +3,9 @@ from cpu_governor_auto_adjust.config import Config
 from cpufreq import cpuFreq, cpufreq  # type: ignore
 from functools import cached_property
 from cpu_governor_auto_adjust.exceptions import GovernorNotFound
-from typing import Optional, NamedTuple
+from typing import Optional, NamedTuple, Union
+from time import monotonic_ns as monotonic
+from cpu_governor_auto_adjust.helper import format_time_ns
 import asyncio
 
         
@@ -22,19 +24,52 @@ _governor_list: tuple[Governor, Governor, Governor, Governor, Governor, Governor
 )
 
 
+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
+
+
 class GovernorControl(AppClass):
     def __init__(self, _config: Config) -> None:
         super().__init__(_config)
         self.governor = self.current_governor
         self.log.info("current governor is: %s", self.governor)
+        self._timestamp_last_statistics = monotonic()
+        self._governor_statistics = self._statistics_dict()
+        self._last_statistic_update = monotonic()
+        
+    def _statistics_dict(self) -> dict[str, str]:
+        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
+        self._last_statistic_update = _now
+
+    def log_statistics(self) -> None:
+        _log_dict = {}
+        for key, value in self._governor_statistics.items():
+            _log_dict[key] = format_time_ns(value)
+        self.log.info("governor statistics: %s", str(_log_dict))
 
     @cached_property
-    def _cpufreq(self) -> Optional[cpuFreq]:
+    def _cpufreq(self) -> Union[cpuFreq, DummyCpuFreq]:
         try:
             return cpuFreq()
         except cpufreq.CPUFreqErrorInit:
             self.log.warning("cpu architecture has no governor support")
-        return None
+            return DummyCpuFreq()
     
     @property
     def current_governor(self) -> Governor:
@@ -66,9 +101,6 @@ class GovernorControl(AppClass):
             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
@@ -78,7 +110,14 @@ class GovernorControl(AppClass):
     async def run(self) -> None:
         self.log.debug("running governor control")
         while True:
+            start = monotonic()
+            self._update_statistics()
+            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

@@ -1,10 +1,18 @@
-def format_time_ns(elapsed_ns: float) -> str:
-    """Format elapsed time dynamically in an appropriate unit."""
-    if elapsed_ns < 1_000:  # Less than 1 µs
-        return f"{elapsed_ns} ns"  # Nanoseconds
-    elif elapsed_ns < 1_000_000:  # Less than 1 ms
-        return f"{elapsed_ns / 1_000:.2f} µs"  # Microseconds
-    elif elapsed_ns < 1_000_000_000:  # Less than 1 second
-        return f"{elapsed_ns / 1_000_000:.2f} ms"  # Milliseconds
+def format_time_ns(elapsed_ns):
+    """Format elapsed time dynamically, using hh:mm:ss for durations ≥ 1s."""
+    one_sec_ns = 1_000_000_000
+    one_min_ns = 60 * one_sec_ns
+    one_hour_ns = 60 * one_min_ns
+
+    if elapsed_ns < 1_000:
+        return f"{elapsed_ns} ns"
+    elif elapsed_ns < 1_000_000:
+        return f"{elapsed_ns / 1_000:.2f} µs"
+    elif elapsed_ns < one_sec_ns:
+        return f"{elapsed_ns / 1_000_000:.2f} ms"
     else:
-        return f"{elapsed_ns / 1_000_000_000:.2f} s"  # Seconds
+        total_seconds = elapsed_ns // one_sec_ns
+        hours = total_seconds // 3600
+        minutes = (total_seconds % 3600) // 60
+        seconds = total_seconds % 60
+        return f"{hours:02}:{minutes:02}:{seconds:02}"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "cpu_governor_auto_adjust"
-version = "0.5.21"
+version = "0.5.23"
 description = "This application has been developed to automatically change cpu governor based on certain triggers."
 authors = [
     { name = "Martin Reurekas", email = "martin@semrks.nl" }