CPU Temp Graph, displayable values builtin to graphs, positional option for displayable values

CPUTemperature.py

@@ -0,0 +1,108 @@
+from GtkHelper.ComboRow import SimpleComboRowItem
+
+import psutil
+
+# The units the temperature can be shown in, the sensors always report celsius
+UNITS = ["C", "F"]
+
+def get_unit_items() -> list[SimpleComboRowItem]:
+    return [SimpleComboRowItem("C", "°C"), SimpleComboRowItem("F", "°F")]
+
+def celcius_to_fahrenheit(celsius: float) -> float:
+    return celsius * 1.8 + 32
+
+# The sensor to read per chip, in order of preference. These are the package and control
+# sensors that monitoring tools like btop show. The per core and per die sensors next to them
+# (Core 0, Tccd1, ...) follow every boost of a single core, which swings by tens of degrees
+# between two reads and is far too sporadic to put on a key.
+PREFERRED_SENSORS = {
+    "coretemp": ["Package id 0"],  # intel
+    "k10temp": ["Tctl", "Tdie"],   # amd
+}
+
+# Picks the sensor above instead of a specific one
+AUTO = "auto"
+
+# Chips that only report cpu temperatures, so every sensor on them is one. Covers intel
+# (coretemp), amd (k10temp, k8temp, the third party zenpower), arm boards (cpu_thermal) and the
+# acpi thermal zone that laptops and virtual machines fall back to.
+CPU_CHIPS = ["coretemp", "k10temp", "k8temp", "zenpower", "cpu_thermal", "acpitz"]
+
+# Chips that mix cpu sensors with board ones only name them, like the "CPU" sensor of the
+# embedded controller on asus boards or the "CPUTIN" one of super i/o chips
+CPU_NAME = "cpu"
+
+def is_cpu_sensor(chip: str, sensor) -> bool:
+    if chip in CPU_CHIPS or CPU_NAME in chip.lower():
+        return True
+
+    return CPU_NAME in (sensor.label or "").lower()
+
+def get_sensor_item(chip: str, index: int, sensor) -> SimpleComboRowItem:
+    label = sensor.label or f"Sensor {index}"
+    return SimpleComboRowItem(f"{chip}:{index}", f"{chip}: {label}")
+
+def get_sensor_items() -> list[SimpleComboRowItem]:
+    """The cpu sensors of this machine, to let the user pick one themselves."""
+
+    temperatures = psutil.sensors_temperatures()
+
+    items = [SimpleComboRowItem(AUTO, "Auto")]
+    for chip, sensors in temperatures.items():
+        for index, sensor in enumerate(sensors):
+            if is_cpu_sensor(chip, sensor):
+                items.append(get_sensor_item(chip, index, sensor))
+
+    # Nothing on this machine is recognizable as a cpu sensor, offering all of them beats
+    # offering none of them
+    if len(items) == 1:
+        for chip, sensors in temperatures.items():
+            for index, sensor in enumerate(sensors):
+                items.append(get_sensor_item(chip, index, sensor))
+
+    return items
+
+def get_auto_temp(temperatures: dict) -> float | None:
+    for chip, labels in PREFERRED_SENSORS.items():
+        sensors = temperatures.get(chip)
+        if not sensors:
+            continue
+
+        for label in labels:
+            for sensor in sensors:
+                if sensor.label == label:
+                    return sensor.current
+
+        # Unknown sensor layout, the package sensor comes first on every chip we know of
+        return sensors[0].current
+
+    # Neither an intel nor an amd chip, take the first sensor that reads the cpu at all
+    for chip, sensors in temperatures.items():
+        for sensor in sensors:
+            if is_cpu_sensor(chip, sensor):
+                return sensor.current
+
+    return None
+
+def get_cpu_temp(sensor: str = AUTO) -> float | None:
+    """
+    Returns the temperature of the selected sensor in celsius, or None if it is not available.
+    A sensor is stored as "chip:index", anything else falls back to the preferred sensor.
+    """
+
+    temperatures = psutil.sensors_temperatures()
+
+    if sensor and sensor != AUTO:
+        chip, _, index = sensor.rpartition(":")
+        sensors = temperatures.get(chip, [])
+
+        if index.isdigit() and int(index) < len(sensors):
+            return sensors[int(index)].current
+
+        # The picked sensor is gone, a missing reading is worse than a different one
+    return get_auto_temp(temperatures)
+
+def convert_temp(celsius: float, unit: str) -> float:
+    if unit == "F":
+        return celcius_to_fahrenheit(celsius)
+    return celsius

CPUUsage.py

@@ -0,0 +1,69 @@
+from GtkHelper.ComboRow import SimpleComboRowItem
+
+import threading
+import time
+import psutil
+
+# The methods that can be selected in the "Usage Method" row of the cpu actions
+USAGE_METHODS = ["average", "max-core"]
+
+SAMPLE_INTERVAL = 1
+FIRST_SAMPLE_INTERVAL = 0.15
+
+def get_usage_method_items() -> list[SimpleComboRowItem]:
+    return [SimpleComboRowItem("average", "Average of all cores"),
+            SimpleComboRowItem("max-core", "Highest single core")]
+
+class CPUSampler(threading.Thread):
+    """
+    Samples the cpu usage on its own thread and hands the last reading to every cpu action.
+
+    psutil measures the usage since the last call made by the *calling thread*, so letting each
+    action call it directly makes the readings depend on what the other actions did in between.
+    Two actions on the same key are ticked one after the other in the same thread, which left the
+    second one measuring nothing but the work of the first (the graph rendering, for example).
+    """
+
+    def __init__(self):
+        super().__init__(daemon=True, name="OSPluginCPUSampler")
+
+        self.lock = threading.Lock()
+        self.average = 0.0
+        self.max_core = 0.0
+
+    def run(self):
+        # The first reading of each is meaningless, psutil needs a previous one to compare against
+        psutil.cpu_percent()
+        psutil.cpu_percent(percpu=True)
+
+        # The first window is kept short so the actions don't show 0% for a whole tick on startup
+        interval = FIRST_SAMPLE_INTERVAL
+
+        while True:
+            time.sleep(interval)
+            interval = SAMPLE_INTERVAL
+
+            average = psutil.cpu_percent()
+            per_core = psutil.cpu_percent(percpu=True)
+
+            with self.lock:
+                self.average = average
+                self.max_core = max(per_core) if per_core else 0.0
+
+    def get_percent(self, method: str) -> float:
+        with self.lock:
+            return self.max_core if method == "max-core" else self.average
+
+_sampler: CPUSampler | None = None
+_sampler_lock = threading.Lock()
+
+def get_sampler() -> CPUSampler:
+    global _sampler
+    with _sampler_lock:
+        if _sampler is None:
+            _sampler = CPUSampler()
+            _sampler.start()
+    return _sampler
+
+def get_cpu_percent(method: str) -> float:
+    return get_sampler().get_percent(method)

CPU_Graph.py

@@ -1,17 +1,122 @@
+from GtkHelper.ComboRow import SimpleComboRowItem
+from GtkHelper.GenerativeUI.ComboRow import ComboRow
+from plugins.com_core447_OSPlugin.CPUTemperature import AUTO, convert_temp, get_cpu_temp, get_sensor_items, get_unit_items
+from plugins.com_core447_OSPlugin.CPUUsage import get_cpu_percent, get_usage_method_items
 from plugins.com_core447_OSPlugin.GraphBase import GraphBase
 from src.backend.DeckManagement.DeckController import DeckController
 from src.backend.PageManagement.Page import Page
 from src.backend.PluginManager.PluginBase import PluginBase
 
-import psutil
-
 from PIL import Image
 
+# The y-axis of a temperature graph, cpus report celsius so fahrenheit needs a taller axis
+MAX_TEMP_C = 100
+MAX_TEMP_F = 212
+
 class CPU_Graph(GraphBase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.has_configuration = False
 
+        self.temp_available = True
+
+        self.usage_method_row = ComboRow(
+            action_core=self,
+            var_name="usage_method",
+            default_value="average",
+            items=get_usage_method_items(),
+            title="Usage Method",
+            can_reset=False
+        )
+
+        self.unit_row = ComboRow(
+            action_core=self,
+            var_name="unit",
+            default_value="C",
+            items=get_unit_items(),
+            title="Unit",
+            can_reset=False,
+            on_change=lambda *args: self.on_source_change()
+        )
+
+        self.sensor_row = ComboRow(
+            action_core=self,
+            var_name="sensor",
+            default_value=AUTO,
+            items=get_sensor_items(),
+            title="Sensor",
+            can_reset=False,
+            on_change=lambda *args: self.on_source_change()
+        )
+
+        # Created last so its on_change always finds the rows it enables and disables
+        self.graph_type_row = ComboRow(
+            action_core=self,
+            var_name="graph_type",
+            default_value="usage",
+            items=[SimpleComboRowItem("usage", "CPU Usage"),
+                   SimpleComboRowItem("temp", "CPU Temperature")],
+            title="Graph Type",
+            can_reset=False,
+            on_change=self.on_graph_type_change
+        )
+
+    def shows_temp(self) -> bool:
+        return self.graph_type_row.get_value() == "temp"
+
+    def on_graph_type_change(self, widget=None, new_item=None, old_item=None):
+        self.update_row_sensitivity()
+
+        old_value = old_item.get_value() if old_item is not None else None
+        new_value = new_item.get_value() if new_item is not None else self.graph_type_row.get_value()
+
+        # A history of percentages means nothing on a temperature axis and the other way around
+        if old_value is not None and old_value != new_value:
+            self.percentages.clear()
+
+        if self.on_ready_called:
+            self.show_graph()
+
+    def on_source_change(self):
+        # Readings of another sensor or in another unit cannot share a history either
+        self.percentages.clear()
+
+        if self.on_ready_called:
+            self.show_graph()
+
+    def update_row_sensitivity(self):
+        shows_temp = self.shows_temp()
+        self.unit_row.set_sensitive(shows_temp)
+        self.sensor_row.set_sensitive(shows_temp)
+        self.usage_method_row.set_sensitive(not shows_temp)
+
+    def get_y_max(self) -> float:
+        if not self.shows_temp():
+            return 100
+        return MAX_TEMP_F if self.unit_row.get_value() == "F" else MAX_TEMP_C
+
+    def get_current_value(self) -> float:
+        if not self.shows_temp():
+            self.temp_available = True
+            return get_cpu_percent(self.usage_method_row.get_value())
+
+        temperature = get_cpu_temp(self.sensor_row.get_value())
+        self.temp_available = temperature is not None
+        if temperature is None:
+            return 0.0
+
+        return convert_temp(temperature, self.unit_row.get_value())
+
+    def get_label_text(self) -> str:
+        if not self.shows_temp():
+            return super().get_label_text()
+
+        if not self.temp_available:
+            return "N/A"
+
+        value = round(self.percentages[-1]) if self.percentages else 0
+        return f"{value} °{self.unit_row.get_value()}"
+
     def on_tick(self):
-        self.percentages.append(psutil.cpu_percent())
-        self.show_graph()
+        self.percentages.append(self.get_current_value())
+        self.show_graph()

GraphBase.py

@@ -1,4 +1,5 @@
 from threading import Thread
+from plugins.com_core447_OSPlugin.LabelPosition import OFF, create_position_row, set_positioned_label
 from src.backend.PluginManager.ActionBase import ActionBase
 from src.backend.DeckManagement.DeckController import DeckController
 from src.backend.PageManagement.Page import Page
@@ -31,6 +32,14 @@ def __init__(self, *args, **kwargs):
 
         self.percentages: list[float] = []
 
+        self.label_position_row = create_position_row(
+            self,
+            on_change=lambda *args: self.show_label(),
+            default_value=OFF,
+            include_off=True,
+            title="Value Label Position"
+        )
+
         self.task_queue = Queue()
         self.result_queue = Queue()
         # self.process = Process(target=self.worker, args=(self.task_queue, self.result_queue), name="GraphBaseCreator")
@@ -52,12 +61,18 @@ def set_percentages_lenght(self, length: int):
 
         return self.percentages
 
+    def get_y_max(self) -> float:
+        """The top of the y-axis, overwrite this for values that are not a percentage."""
+        return 100
+
     def get_graph(self) -> Image:
         ## Get vars
-        settings = self.get_settings()
+        settings = dict(self.get_settings())
         time_period = settings.get("time-period", 15)
         self.set_percentages_lenght(time_period)
 
+        settings["y-max"] = self.get_y_max()
+
         self.task_queue.put((settings, self.percentages))
 
         img = self.result_queue.get()
@@ -68,6 +83,14 @@ def show_graph(self):
         if image is None:
             return
         self.set_media(image=image)
+        self.show_label()
+
+    def get_label_text(self) -> str:
+        percent = round(self.percentages[-1]) if self.percentages else 0
+        return f"{percent}%"
+
+    def show_label(self):
+        set_positioned_label(self, self.label_position_row, self.get_label_text(), fallback=OFF)
 
     def get_custom_config_area(self):
         return Gtk.Label(label=self.plugin_base.lm.get("actions.graph-base.memory-warning"), css_classes=["destructive-action"])
@@ -243,9 +266,9 @@ def generate_graph(self, settings: dict, percentages: list[float]):
         ax.margins(0)
         ax.axis('off')
 
-        # Set the y-axis to range from 0 to 100
+        # Set the y-axis to range from 0 to the maximum the action can report
         if not dynamic_scaling:
-            ax.set_ylim(0, 100)
+            ax.set_ylim(0, settings.get("y-max", 100))
 
         # Draw the canvas and retrieve the buffer
         canvas.draw()