frobar-ng: Alerting sections

hm/desktop/frobar/.dev/new.py

@@ -5,6 +5,7 @@ import datetime
 import enum
 import ipaddress
 import logging
+import os
 import signal
 import socket
 import typing
@@ -29,6 +30,8 @@ P = typing.TypeVar("P", bound="ComposableText")
 C = typing.TypeVar("C", bound="ComposableText")
 Sortable = str | int
 
+# Display utilities
+
 
 def humanSize(numi: int) -> str:
     """
@@ -38,11 +41,20 @@ def humanSize(numi: int) -> str:
     for unit in ("B  ", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB"):
         if abs(num) < 1000:
             if num >= 10:
-                return "{:3d}{}".format(int(num), unit)
+                return f"{int(num):3d}{unit}"
             else:
-                return "{:.1f}{}".format(num, unit)
+                return f"{num:.1f}{unit}"
         num /= 1024
-    return "{:d}YiB".format(numi)
+    return f"{numi:d}YiB"
+
+
+def ramp(p: float, ramp: str = " ▁▂▃▄▅▆▇█") -> str:
+    if p > 1:
+        return ramp[-1]
+    elif p < 0:
+        return ramp[0]
+    else:
+        return ramp[round(p * (len(ramp) - 1))]
 
 
 class ComposableText(typing.Generic[P, C]):
@@ -544,6 +556,7 @@ class I3WindowTitleProvider(SingleSectionProvider):
 class I3WorkspacesProvider(Provider):
     COLOR_URGENT = rich.color.Color.parse("red")
     COLOR_FOCUSED = rich.color.Color.parse("yellow")
+    # TODO Should be orange (not a terminal color)
     COLOR_VISIBLE = rich.color.Color.parse("blue")
     COLOR_DEFAULT = rich.color.Color.parse("bright_black")
 
@@ -624,6 +637,149 @@ class I3WorkspacesProvider(Provider):
         await i3.main()
 
 
+class AlertingProvider(Provider):
+    COLOR_NORMAL = rich.color.Color.parse("green")
+    COLOR_WARNING = rich.color.Color.parse("yellow")
+    COLOR_DANGER = rich.color.Color.parse("red")
+
+    warningThreshold: float
+    dangerThreshold: float
+
+    def updateLevel(self, level: float) -> None:
+        if level > self.dangerThreshold:
+            color = self.COLOR_DANGER
+        elif level > self.warningThreshold:
+            color = self.COLOR_WARNING
+        else:
+            color = self.COLOR_NORMAL
+        for module in self.modules:
+            for section in module.getSections():
+                section.color = color
+
+
+class CpuProvider(AlertingProvider, PeriodicStatefulProvider):
+    async def init(self) -> None:
+        self.section.numberStates = 3
+        self.warningThreshold = 75
+        self.dangerThreshold = 95
+
+    async def loop(self) -> None:
+        percent = psutil.cpu_percent(percpu=False)
+        self.updateLevel(percent)
+
+        text = ""
+        if self.section.state >= 2:
+            percents = psutil.cpu_percent(percpu=True)
+            text += " " + "".join([ramp(p / 100) for p in percents])
+        elif self.section.state >= 1:
+            text += " " + ramp(percent / 100)
+        self.section.setText(text)
+
+
+class LoadProvider(AlertingProvider, PeriodicStatefulProvider):
+    async def init(self) -> None:
+        self.section.numberStates = 3
+        self.warningThreshold = 5
+        self.dangerThreshold = 10
+
+    async def loop(self) -> None:
+        load = os.getloadavg()
+        self.updateLevel(load[0])
+
+        text = ""
+        loads = 3 if self.section.state >= 2 else self.section.state
+        for load_index in range(loads):
+            text += f" {load[load_index]:.2f}"
+        self.section.setText(text)
+
+
+class RamProvider(AlertingProvider, PeriodicStatefulProvider):
+
+    async def init(self) -> None:
+        self.section.numberStates = 4
+        self.warningThreshold = 75
+        self.dangerThreshold = 95
+
+    async def loop(self) -> None:
+        mem = psutil.virtual_memory()
+        self.updateLevel(mem.percent)
+
+        text = ""
+        if self.section.state >= 1:
+            text += " " + ramp(mem.percent / 100)
+            if self.section.state >= 2:
+                text += humanSize(mem.total - mem.available)
+                if self.section.state >= 3:
+                    text += "/" + humanSize(mem.total)
+        self.section.setText(text)
+
+
+class TemperatureProvider(AlertingProvider, PeriodicStatefulProvider):
+    RAMP = ""
+    MAIN_TEMPS = ["coretemp", "amdgpu", "cpu_thermal"]
+    # For Intel, AMD and ARM respectively.
+
+    main: str
+
+    async def init(self) -> None:
+        self.section.numberStates = 2
+        self.warningThreshold = 75
+        self.dangerThreshold = 95
+
+        allTemp = psutil.sensors_temperatures()
+        for main in self.MAIN_TEMPS:
+            if main in allTemp:
+                self.main = main
+                break
+        else:
+            raise IndexError("Could not find suitable temperature sensor")
+
+        temp = allTemp[self.main][0]
+        self.warningThresold = temp.high or 90.0
+        self.dangerThresold = temp.critical or 100.0
+
+    async def loop(self) -> None:
+        allTemp = psutil.sensors_temperatures()
+        temp = allTemp[self.main][0]
+        self.updateLevel(temp.current)
+
+        text = ramp(temp.current / self.warningThreshold, self.RAMP)
+        if self.section.state >= 1:
+            text += f" {temp.current:.0f}°C"
+        self.section.setText(text)
+
+
+class BatteryProvider(AlertingProvider, PeriodicStatefulProvider):
+    # TODO Support ACPID for events
+    RAMP = ""
+
+    async def init(self) -> None:
+        self.section.numberStates = 3
+        # TODO 1 refresh rate is too quick
+
+        self.warningThreshold = 75
+        self.dangerThreshold = 95
+
+    async def loop(self) -> None:
+        bat = psutil.sensors_battery()
+        if not bat:
+            self.section.setText(None)
+
+        self.updateLevel(100 - bat.percent)
+
+        text = "" if bat.power_plugged else ""
+        text += ramp(bat.percent / 100, self.RAMP)
+
+        if self.section.state >= 1:
+            text += f" {bat.percent:.0f}%"
+            if self.section.state >= 2:
+                h = int(bat.secsleft / 3600)
+                m = int((bat.secsleft - h * 3600) / 60)
+                text += f" ({h:d}:{m:02d})"
+
+        self.section.setText(text)
+
+
 class NetworkProviderSection(StatefulSection):
     def __init__(
         self,
@@ -782,6 +938,8 @@ async def main() -> None:
         "#d43982",
     ]
     # TODO Configurable
+    # TODO Not super happy with the color management,
+    # while using an existing library is great, it's limited to ANSI colors
 
     def base16_color(color: int) -> tuple[int, int, int]:
         hexa = FROGARIZED[color]
@@ -829,9 +987,11 @@ async def main() -> None:
         alignment=Alignment.CENTER,
     )
 
-    bar.addProvider(
-        StaticProvider("C L M T B", color=color("green")), alignment=Alignment.RIGHT
-    )
+    bar.addProvider(CpuProvider(), alignment=Alignment.RIGHT)
+    bar.addProvider(LoadProvider(), alignment=Alignment.RIGHT)
+    bar.addProvider(RamProvider(), alignment=Alignment.RIGHT)
+    bar.addProvider(TemperatureProvider(), alignment=Alignment.RIGHT)
+    bar.addProvider(BatteryProvider(), alignment=Alignment.RIGHT)
     bar.addProvider(
         StaticProvider("pulse", color=color("magenta")),
         screenNum=1 if dualScreen else None,