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, )