diff --git a/kiwi_simple_metrics/main.py b/kiwi_simple_metrics/main.py index 9f2840b..1139cf3 100644 --- a/kiwi_simple_metrics/main.py +++ b/kiwi_simple_metrics/main.py @@ -1,4 +1,4 @@ -from .metrics import cpu_metric, disk_metric +from . import metrics from .settings import SETTINGS @@ -7,12 +7,12 @@ def main() -> None: print(SETTINGS.model_dump()) # CPU metric - print(cpu_metric()) + print(metrics.cpu()) # MEM metric # DISK metric - print(disk_metric()) + print(metrics.disk()) if __name__ == "__main__": diff --git a/kiwi_simple_metrics/metrics.py b/kiwi_simple_metrics/metrics.py deleted file mode 100644 index 58242d9..0000000 --- a/kiwi_simple_metrics/metrics.py +++ /dev/null @@ -1,81 +0,0 @@ -import os -from dataclasses import dataclass, field - -import psutil - -from .settings import SETTINGS, MetricSettings - - -@dataclass(slots=True, frozen=True) -class Report: - result: str - failed: bool = field(default=False, kw_only=True) - - -def _report( - *, - settings: MetricSettings, - name: str, - value: float, -) -> Report: - result = settings.report.format(name=name, value=value) - - if ( - value > settings.threshold and not settings.inverted - or value < settings.threshold and settings.inverted - ): - return Report(result, failed=True) - - else: - return Report(result) - - -def cpu_metric() -> Report | None: - if not SETTINGS.cpu.enabled: - return None - - value = psutil.cpu_percent(interval=1) - return _report( - settings=SETTINGS.cpu, - name=SETTINGS.cpu.name, - value=value, - ) - - -def disk_metric() -> Report | None: - if not SETTINGS.disk.enabled: - return None - - def path_to_free_percent(path: os.PathLike) -> float: - try: - sv = os.statvfs(path) - return sv.f_bavail / sv.f_blocks * 100 - except ZeroDivisionError: - return 0 - - data = sorted([ - (str(path), path_to_free_percent(path)) - for path in SETTINGS.disk.paths - ], key=lambda d: d[1]) - - reports = [_report( - 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 - ), - ) diff --git a/kiwi_simple_metrics/metrics/__init__.py b/kiwi_simple_metrics/metrics/__init__.py new file mode 100644 index 0000000..a2223b7 --- /dev/null +++ b/kiwi_simple_metrics/metrics/__init__.py @@ -0,0 +1,7 @@ +from .cpu import cpu +from .disk import disk + +__all__ = [ + "cpu", + "disk", +] diff --git a/kiwi_simple_metrics/metrics/_report.py b/kiwi_simple_metrics/metrics/_report.py new file mode 100644 index 0000000..38101cb --- /dev/null +++ b/kiwi_simple_metrics/metrics/_report.py @@ -0,0 +1,28 @@ +from dataclasses import dataclass, field +from typing import Self + +from ..settings import MetricSettings + + +@dataclass(slots=True, frozen=True) +class Report: + result: str + failed: bool = field(default=False, kw_only=True) + + @classmethod + def new( + cls, *, + settings: MetricSettings, + name: str, + value: float, + ) -> Self: + result = settings.report.format(name=name, value=value) + + if ( + value > settings.threshold and not settings.inverted + or value < settings.threshold and settings.inverted + ): + return cls(result, failed=True) + + else: + return cls(result) diff --git a/kiwi_simple_metrics/metrics/cpu.py b/kiwi_simple_metrics/metrics/cpu.py new file mode 100644 index 0000000..8ab4a73 --- /dev/null +++ b/kiwi_simple_metrics/metrics/cpu.py @@ -0,0 +1,16 @@ +import psutil + +from ..settings import SETTINGS +from ._report import Report + + +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, + ) diff --git a/kiwi_simple_metrics/metrics/disk.py b/kiwi_simple_metrics/metrics/disk.py new file mode 100644 index 0000000..cdc46b8 --- /dev/null +++ b/kiwi_simple_metrics/metrics/disk.py @@ -0,0 +1,43 @@ +import os + +from ..settings import SETTINGS +from ._report import Report + + +def disk() -> Report | None: + if not SETTINGS.disk.enabled: + return None + + def path_to_free_percent(path: os.PathLike) -> float: + try: + sv = os.statvfs(path) + return sv.f_bavail / sv.f_blocks * 100 + except ZeroDivisionError: + return 0 + + data = sorted([ + (str(path), path_to_free_percent(path)) + for path in SETTINGS.disk.paths + ], key=lambda d: d[1]) + + reports = [Report.new( + 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 + ), + )