# -*- coding: utf-8 -*- """ Health-Scoring für Satellites. Berechnet zusammengesetzte Gesundheits-Scores basierend auf verschiedenen Metriken wie Latenz, Paketverlust, Uptime etc. """ import time from collections import deque from dataclasses import dataclass, field from datetime import datetime from enum import IntEnum, auto from typing import Any, Callable, TYPE_CHECKING if TYPE_CHECKING: from trixy_core.satellite.satellite import Satellite class HealthState(IntEnum): """Gesundheitszustand eines Satellites.""" HEALTHY = auto() # Alles in Ordnung DEGRADED = auto() # Eingeschränkt, aber funktionsfähig UNHEALTHY = auto() # Problematisch CRITICAL = auto() # Kritisch, sollte nicht verwendet werden UNKNOWN = auto() # Keine Daten verfügbar @dataclass class HealthMetrics: """ Gesammelte Gesundheitsmetriken. Attributes: latency_ms: Durchschnittliche Latenz in Millisekunden packet_loss: Paketverlustrate (0.0-1.0) uptime: Uptime-Prozentsatz (0.0-1.0) error_rate: Fehlerrate (0.0-1.0) memory_usage: Speicherauslastung (0.0-1.0) cpu_usage: CPU-Auslastung (0.0-1.0) last_heartbeat_age: Zeit seit letztem Heartbeat in Sekunden connection_quality: Verbindungsqualität (0.0-1.0) custom_metrics: Zusätzliche benutzerdefinierte Metriken """ latency_ms: float = 0.0 packet_loss: float = 0.0 uptime: float = 1.0 error_rate: float = 0.0 memory_usage: float = 0.0 cpu_usage: float = 0.0 last_heartbeat_age: float = 0.0 connection_quality: float = 1.0 custom_metrics: dict[str, float] = field(default_factory=dict) timestamp: datetime = field(default_factory=datetime.now) @dataclass class HealthScore: """ Berechneter Gesundheits-Score. Attributes: overall: Gesamtscore (0.0-100.0) state: Abgeleiteter Zustand components: Einzelne Komponenten-Scores metrics: Die zugrundeliegenden Metriken warnings: Warnungen calculated_at: Berechnungszeitpunkt """ overall: float state: HealthState components: dict[str, float] metrics: HealthMetrics warnings: list[str] = field(default_factory=list) calculated_at: datetime = field(default_factory=datetime.now) @property def is_healthy(self) -> bool: """Prüft, ob der Satellite gesund ist.""" return self.state in (HealthState.HEALTHY, HealthState.DEGRADED) @property def is_usable(self) -> bool: """Prüft, ob der Satellite verwendbar ist.""" return self.state != HealthState.CRITICAL def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "overall": self.overall, "state": self.state.name, "components": self.components, "warnings": self.warnings, "is_healthy": self.is_healthy, "is_usable": self.is_usable, "calculated_at": self.calculated_at.isoformat(), } class HealthScorer: """ Berechnet Gesundheits-Scores für Satellites. Kombiniert verschiedene Metriken zu einem Gesamtscore. Gewichtung der Metriken ist konfigurierbar. Example: scorer = HealthScorer() # Metriken aktualisieren scorer.update_metrics("satellite_1", HealthMetrics( latency_ms=50.0, packet_loss=0.01, uptime=0.999, error_rate=0.001 )) # Score berechnen score = scorer.calculate_score("satellite_1") print(f"Gesamtscore: {score.overall}") print(f"Zustand: {score.state.name}") # Alle Satellites bewerten for satellite_id, score in scorer.get_all_scores().items(): print(f"{satellite_id}: {score.overall}") """ # Standard-Gewichte für Metriken DEFAULT_WEIGHTS = { "latency": 0.20, "packet_loss": 0.15, "uptime": 0.15, "error_rate": 0.15, "heartbeat": 0.15, "connection_quality": 0.10, "resources": 0.10, } # Schwellenwerte für Zustandsbestimmung DEFAULT_THRESHOLDS = { HealthState.HEALTHY: 80.0, HealthState.DEGRADED: 60.0, HealthState.UNHEALTHY: 40.0, HealthState.CRITICAL: 0.0, } def __init__( self, weights: dict[str, float] | None = None, thresholds: dict[HealthState, float] | None = None, history_size: int = 100, ) -> None: """ Initialisiert den HealthScorer. Args: weights: Gewichtung der Metriken thresholds: Schwellenwerte für Zustände history_size: Anzahl der zu speichernden historischen Werte """ self._weights = weights or self.DEFAULT_WEIGHTS.copy() self._thresholds = thresholds or self.DEFAULT_THRESHOLDS.copy() self._history_size = history_size self._metrics: dict[str, HealthMetrics] = {} self._history: dict[str, deque[HealthScore]] = {} self._custom_scorers: dict[str, Callable[[HealthMetrics], float]] = {} def update_metrics(self, satellite_id: str, metrics: HealthMetrics) -> None: """ Aktualisiert die Metriken für einen Satellite. Args: satellite_id: Satellite-ID metrics: Neue Metriken """ metrics.timestamp = datetime.now() self._metrics[satellite_id] = metrics def update_metric( self, satellite_id: str, metric_name: str, value: float ) -> None: """ Aktualisiert eine einzelne Metrik. Args: satellite_id: Satellite-ID metric_name: Name der Metrik value: Neuer Wert """ if satellite_id not in self._metrics: self._metrics[satellite_id] = HealthMetrics() metrics = self._metrics[satellite_id] if hasattr(metrics, metric_name): setattr(metrics, metric_name, value) else: metrics.custom_metrics[metric_name] = value metrics.timestamp = datetime.now() def add_custom_scorer( self, name: str, scorer: Callable[[HealthMetrics], float], weight: float = 0.1 ) -> None: """ Fügt einen benutzerdefinierten Scorer hinzu. Args: name: Name des Scorers scorer: Funktion die Metriken -> Score (0-100) umwandelt weight: Gewicht des Scorers """ self._custom_scorers[name] = scorer self._weights[name] = weight def calculate_score(self, satellite_id: str) -> HealthScore: """ Berechnet den Gesundheits-Score. Args: satellite_id: Satellite-ID Returns: HealthScore """ metrics = self._metrics.get(satellite_id) if metrics is None: return HealthScore( overall=0.0, state=HealthState.UNKNOWN, components={}, metrics=HealthMetrics(), warnings=["Keine Metriken verfügbar"], ) components = {} warnings = [] # Latenz-Score (niedrigere Latenz = höherer Score) if "latency" in self._weights: latency_score = self._calculate_latency_score(metrics.latency_ms) components["latency"] = latency_score if latency_score < 50: warnings.append(f"Hohe Latenz: {metrics.latency_ms:.0f}ms") # Paketverlust-Score if "packet_loss" in self._weights: loss_score = (1.0 - metrics.packet_loss) * 100 components["packet_loss"] = loss_score if metrics.packet_loss > 0.05: warnings.append(f"Hoher Paketverlust: {metrics.packet_loss*100:.1f}%") # Uptime-Score if "uptime" in self._weights: uptime_score = metrics.uptime * 100 components["uptime"] = uptime_score if metrics.uptime < 0.99: warnings.append(f"Niedrige Uptime: {metrics.uptime*100:.1f}%") # Fehlerrate-Score if "error_rate" in self._weights: error_score = (1.0 - metrics.error_rate) * 100 components["error_rate"] = error_score if metrics.error_rate > 0.05: warnings.append(f"Hohe Fehlerrate: {metrics.error_rate*100:.1f}%") # Heartbeat-Score if "heartbeat" in self._weights: heartbeat_score = self._calculate_heartbeat_score(metrics.last_heartbeat_age) components["heartbeat"] = heartbeat_score if heartbeat_score < 50: warnings.append(f"Heartbeat veraltet: {metrics.last_heartbeat_age:.0f}s") # Verbindungsqualität if "connection_quality" in self._weights: conn_score = metrics.connection_quality * 100 components["connection_quality"] = conn_score # Ressourcen-Score if "resources" in self._weights: resource_score = self._calculate_resource_score( metrics.cpu_usage, metrics.memory_usage ) components["resources"] = resource_score if resource_score < 50: warnings.append("Hohe Ressourcenauslastung") # Benutzerdefinierte Scorer for name, scorer in self._custom_scorers.items(): try: score = scorer(metrics) components[name] = max(0, min(100, score)) except Exception: components[name] = 50.0 # Default bei Fehler # Gewichteter Gesamtscore overall = self._calculate_weighted_score(components) # Zustand bestimmen state = self._determine_state(overall) score = HealthScore( overall=overall, state=state, components=components, metrics=metrics, warnings=warnings, ) # In Historie speichern self._add_to_history(satellite_id, score) return score def _calculate_latency_score(self, latency_ms: float) -> float: """Berechnet den Latenz-Score.""" if latency_ms <= 0: return 100.0 elif latency_ms <= 50: return 100.0 elif latency_ms <= 100: return 90.0 - (latency_ms - 50) * 0.2 elif latency_ms <= 200: return 80.0 - (latency_ms - 100) * 0.3 elif latency_ms <= 500: return 50.0 - (latency_ms - 200) * 0.1 else: return max(0.0, 20.0 - (latency_ms - 500) * 0.02) def _calculate_heartbeat_score(self, age_seconds: float) -> float: """Berechnet den Heartbeat-Score.""" if age_seconds <= 5: return 100.0 elif age_seconds <= 10: return 90.0 elif age_seconds <= 30: return 70.0 - (age_seconds - 10) * 1.0 elif age_seconds <= 60: return 50.0 - (age_seconds - 30) * 0.5 else: return max(0.0, 35.0 - (age_seconds - 60) * 0.3) def _calculate_resource_score(self, cpu: float, memory: float) -> float: """Berechnet den Ressourcen-Score.""" avg_usage = (cpu + memory) / 2 return (1.0 - avg_usage) * 100 def _calculate_weighted_score(self, components: dict[str, float]) -> float: """Berechnet den gewichteten Gesamtscore.""" total_weight = 0.0 weighted_sum = 0.0 for name, score in components.items(): weight = self._weights.get(name, 0.0) weighted_sum += score * weight total_weight += weight if total_weight > 0: return weighted_sum / total_weight return 0.0 def _determine_state(self, overall: float) -> HealthState: """Bestimmt den Zustand basierend auf dem Score.""" if overall >= self._thresholds[HealthState.HEALTHY]: return HealthState.HEALTHY elif overall >= self._thresholds[HealthState.DEGRADED]: return HealthState.DEGRADED elif overall >= self._thresholds[HealthState.UNHEALTHY]: return HealthState.UNHEALTHY else: return HealthState.CRITICAL def _add_to_history(self, satellite_id: str, score: HealthScore) -> None: """Fügt einen Score zur Historie hinzu.""" if satellite_id not in self._history: self._history[satellite_id] = deque(maxlen=self._history_size) self._history[satellite_id].append(score) def get_history(self, satellite_id: str) -> list[HealthScore]: """Gibt die Score-Historie zurück.""" return list(self._history.get(satellite_id, [])) def get_average_score(self, satellite_id: str, samples: int = 10) -> float: """ Berechnet den Durchschnittsscore über die letzten N Samples. Args: satellite_id: Satellite-ID samples: Anzahl Samples Returns: Durchschnittsscore """ history = self._history.get(satellite_id, []) if not history: return 0.0 recent = list(history)[-samples:] return sum(s.overall for s in recent) / len(recent) def get_all_scores(self) -> dict[str, HealthScore]: """Berechnet und gibt alle Scores zurück.""" return { satellite_id: self.calculate_score(satellite_id) for satellite_id in self._metrics.keys() } def get_healthy_satellites( self, min_state: HealthState = HealthState.DEGRADED ) -> list[str]: """ Gibt alle Satellites mit mindestens dem angegebenen Zustand zurück. Args: min_state: Minimaler akzeptabler Zustand Returns: Liste der Satellite-IDs """ result = [] for satellite_id in self._metrics.keys(): score = self.calculate_score(satellite_id) if score.state <= min_state: result.append(satellite_id) return result def remove_satellite(self, satellite_id: str) -> None: """Entfernt einen Satellite aus dem Scoring.""" self._metrics.pop(satellite_id, None) self._history.pop(satellite_id, None) def clear(self) -> None: """Löscht alle Daten.""" self._metrics.clear() self._history.clear()