improved aggregation by refactoring metrics._report

ReportData mutable, Report immutable
class Report has builtin `aggregate`

kiwi_simple_metrics/metrics/_report.py

@@ -4,25 +4,56 @@ from typing import Self
 from ..settings import MetricSettings
 
 
+@dataclass(slots=True, kw_only=True)
+class ReportData:
+    name: str
+    value: float
+
+    @classmethod
+    def from_free_total(cls, *, name: str, free: float, total: float) -> Self:
+        return cls(
+            name=name,
+            value=(total - free) / total * 100,
+        )
+
+    def report(self, settings: MetricSettings) -> "Report":
+        result = settings.report.format(
+            name=self.name,
+            value=self.value,
+        )
+
+        return Report(result, failed=(
+            self.value > settings.threshold and not settings.inverted
+            or self.value < settings.threshold and settings.inverted
+        ))
+
+
 @dataclass(slots=True, frozen=True)
 class Report:
     result: str
     failed: bool = field(default=False, kw_only=True)
 
     @classmethod
-    def new(
+    def aggregate(
         cls, *,
         settings: MetricSettings,
-        name: str,
-        value: float,
+        data: list[ReportData],
     ) -> Self:
-        result = settings.report.format(name=name, value=value)
+        reports = [
+            data.report(settings)
+            for data in data
+        ]
 
-        if (
-            value > settings.threshold and not settings.inverted
-            or value < settings.threshold and settings.inverted
-        ):
-            return cls(result, failed=True)
-
-        else:
-            return cls(result)
+        return cls(
+            settings.report_outer.format(
+                name=settings.name,
+                inner=", ".join(
+                    report.result
+                    for report in reports[:settings.count]
+                ),
+            ),
+            failed=any(
+                report.failed
+                for report in reports
+            ),
+        )

kiwi_simple_metrics/metrics/cpu.py

@@ -1,16 +1,20 @@
 import psutil
 
 from ..settings import SETTINGS
-from ._report import Report
+from ._report import Report, ReportData
+
+
+def _hwdata() -> ReportData:
+    return ReportData(
+        name=SETTINGS.cpu.name,
+        value=psutil.cpu_percent(interval=1),
+    )
 
 
 def cpu() -> Report | None:
     if not SETTINGS.cpu.enabled:
         return None
 
-    value = psutil.cpu_percent(interval=1)
-    return Report.new(
-        settings=SETTINGS.cpu,
-        name=SETTINGS.cpu.name,
-        value=value,
-    )
+    data = _hwdata()
+
+    return data.report(SETTINGS.cpu)

kiwi_simple_metrics/metrics/disk.py

@@ -1,43 +1,32 @@
 import os
 
 from ..settings import SETTINGS
-from ._report import Report
+from ._report import Report, ReportData
+
+
+def _hwdata() -> list[ReportData]:
+    def get_path_statvfs(path: os.PathLike) -> dict[str, float]:
+        sv = os.statvfs(path)
+        return {
+            "free": sv.f_bavail,
+            "total": sv.f_blocks,
+        }
+
+    return sorted([
+        ReportData.from_free_total(
+            name=str(path),
+            **get_path_statvfs(path),
+        ) for path in SETTINGS.disk.paths
+    ], key=lambda d: d.value, reverse=True)
 
 
 def disk() -> Report | None:
     if not SETTINGS.disk.enabled:
         return None
 
-    def path_to_used_percent(path: os.PathLike) -> float:
-        try:
-            sv = os.statvfs(path)
-            return (1 - sv.f_bavail / sv.f_blocks) * 100
-        except ZeroDivisionError:
-            return 0
+    data = _hwdata()
 
-    data = sorted([
-        (str(path), path_to_used_percent(path))
-        for path in SETTINGS.disk.paths
-    ], key=lambda d: d[1], reverse=True)
-
-    reports = [Report.new(
+    return Report.aggregate(
         settings=SETTINGS.disk,
-        name=path,
-        value=percent,
-    ) for path, percent in data]
-
-    report_inner = ", ".join(
-        report.result
-        for report in reports[:SETTINGS.disk.count]
-    )
-
-    return Report(
-        SETTINGS.disk.report_outer.format(
-            name=SETTINGS.disk.name,
-            inner=report_inner,
-        ),
-        failed=any(
-            report.failed
-            for report in reports
-        ),
+        data=data,
     )

kiwi_simple_metrics/metrics/memory.py

@@ -1,56 +1,43 @@
 import psutil
 
 from ..settings import SETTINGS
-from ._report import Report
+from ._report import Report, ReportData
+
+
+def _hwdata() -> list[ReportData]:
+    vmem = psutil.virtual_memory()
+    swap = psutil.swap_memory()
+
+    if SETTINGS.memory.swap == "exclude":
+        return [ReportData(
+            name=SETTINGS.memory.name_ram,
+            value=vmem.percent,
+        )]
+
+    elif SETTINGS.memory.swap == "combine":
+        return [ReportData.from_free_total(
+            name=SETTINGS.memory.name,
+            free=vmem.available + swap.free,
+            total=vmem.total + swap.total,
+        )]
+
+    else:  # SETTINGS.memory.swap == "include"
+        return [ReportData(
+            name=SETTINGS.memory.name_ram,
+            value=vmem.percent,
+        ), ReportData(
+            name=SETTINGS.memory.name_swap,
+            value=swap.percent,
+        )]
 
 
 def memory() -> Report | None:
     if not SETTINGS.memory.enabled:
         return None
 
-    def get_used_percent(free: float, total: float) -> float:
-        return (total - free) / total * 100
+    data = _hwdata()
 
-    vmem = psutil.virtual_memory()
-    swap = psutil.swap_memory()
-
-    if SETTINGS.memory.swap == "exclude":
-        data = {
-            SETTINGS.memory.name_ram: vmem.percent,
-        }
-
-    elif SETTINGS.memory.swap == "combine":
-        data = {
-            SETTINGS.memory.name: get_used_percent(
-                vmem.available + swap.free,
-                vmem.total + swap.total,
-            )
-        }
-
-    else:  # SETTINGS.memory.swap == "include"
-        data = {
-            SETTINGS.memory.name_ram: vmem.percent,
-            SETTINGS.memory.name_swap: swap.percent,
-        }
-
-    reports = [Report.new(
+    return Report.aggregate(
         settings=SETTINGS.memory,
-        name=name,
-        value=value,
-    ) for name, value in data.items()]
-
-    report_inner = ", ".join(
-        report.result
-        for report in reports
-    )
-
-    return Report(
-        SETTINGS.memory.report_outer.format(
-            name=SETTINGS.memory.name,
-            inner=report_inner,
-        ),
-        failed=any(
-            report.failed
-            for report in reports
-        ),
+        data=data,
     )

kiwi_simple_metrics/settings.py

@@ -12,27 +12,28 @@ class MetricSettings(BaseModel):
     # metric will be reported
     enabled: bool = True
 
-    # format string to report the metric
-    report: str = "{name}: {value:.2f}%"
-
     # if the metric value exceeds this percentage, the report fails
     threshold: float
 
     # if True, this metric fails when the value falls below the `threshold`
     inverted: bool = False
 
+    # per-value format string for reporting
+    report: str = "{name}: {value:.2f}%"
+
+    # per-metric format string for reporting
+    report_outer: str = "{name}: [{inner}]"
+
+    # include only `count` many items (None: include all)
+    count: int | None = None
+
 
 class CpuMS(MetricSettings):
     name: str = "CPU"
     threshold: float = math.inf
 
 
-class MultiMS(MetricSettings):
-    # outer format string for reporting
-    report_outer: str = "{name}: [{inner}]"
-
-
-class MemoryMS(MultiMS):
+class MemoryMS(MetricSettings):
     name: str = "Memory"
     threshold: float = 90
     report_outer: str = "{inner}"
@@ -48,16 +49,14 @@ class MemoryMS(MultiMS):
     name_swap: str = "Swap"
 
 
-class DiskMS(MultiMS):
+class DiskMS(MetricSettings):
     name: str = "Disk Used"
     threshold: float = 85
+    count: int = 1
 
     # paths to check for disk space
     paths: list[DirectoryPath] = Field(default_factory=list)
 
-    # include only `count` many of the paths with the least free space
-    count: int = 1
-
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(