100 lines
2.2 KiB
Python
100 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from app.models.definitions import MetricDefinition, MetricValue
|
|
|
|
|
|
|
|
def to_float(value: float | str | None) -> float | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (float, int)):
|
|
return float(value)
|
|
try:
|
|
return float(str(value).replace(",", "."))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
|
|
def round_value(value: float | None, precision: int) -> float | None:
|
|
if value is None:
|
|
return None
|
|
return round(value, precision)
|
|
|
|
|
|
|
|
def compare_delta_pct(current: float | None, previous: float | None) -> float | None:
|
|
if current is None or previous in (None, 0):
|
|
return None
|
|
return round(((current - previous) / previous) * 100.0, 2)
|
|
|
|
|
|
|
|
def build_status(metric_id: str, numeric: float | None) -> str:
|
|
if numeric is None:
|
|
return "neutral"
|
|
|
|
if metric_id == "inverter_temp":
|
|
if numeric < 55:
|
|
return "ok"
|
|
if numeric < 70:
|
|
return "warn"
|
|
return "critical"
|
|
|
|
return "ok"
|
|
|
|
|
|
|
|
def metric_value(
|
|
metric: MetricDefinition,
|
|
value: float | str | None,
|
|
*,
|
|
timestamp: datetime | None = None,
|
|
) -> MetricValue:
|
|
rendered = value
|
|
numeric = None
|
|
if metric.kind != "text":
|
|
numeric = to_float(value)
|
|
rendered = round_value(numeric, metric.precision)
|
|
|
|
return MetricValue(
|
|
metric_id=metric.id,
|
|
label=metric.label,
|
|
unit=metric.unit,
|
|
value=rendered,
|
|
timestamp=timestamp,
|
|
precision=metric.precision,
|
|
kind=metric.kind,
|
|
status=build_status(metric.id, numeric),
|
|
)
|
|
|
|
|
|
|
|
def custom_metric_value(
|
|
metric_id: str,
|
|
label: str,
|
|
value: float | str | None,
|
|
*,
|
|
unit: str = "",
|
|
precision: int = 2,
|
|
timestamp: datetime | None = None,
|
|
status: str = "neutral",
|
|
kind: str = "gauge",
|
|
) -> MetricValue:
|
|
rendered = value
|
|
if kind != "text":
|
|
numeric = to_float(value)
|
|
rendered = round_value(numeric, precision)
|
|
return MetricValue(
|
|
metric_id=metric_id,
|
|
label=label,
|
|
unit=unit,
|
|
value=rendered,
|
|
timestamp=timestamp,
|
|
precision=precision,
|
|
kind=kind,
|
|
status=status,
|
|
)
|