satellite.md 25 KB
trixy_core.satellite
Satellite-Management mit Gruppen, Load-Balancing, Health-Monitoring und Failover.
Übersicht
from trixy_core.satellite import (
# Core
Satellite, SatelliteManager, ConnectionState, SatelliteSocket,
# Registration
RegistrationManager,
# API
SatelliteAPI, SatelliteProxy, SatelliteCollection,
get_satellite_api, TTSRequest, StreamSession, StreamState,
# Groups
SatelliteGroup, GroupManager, GroupType, GroupConfig,
Zone, ZoneManager, ZoneConfig,
# Load Balancing
LoadBalancer, LoadBalancingStrategy, SelectionResult,
RoundRobinStrategy, LeastConnectionsStrategy, WeightedStrategy,
# Health
HealthScorer, HealthScore, HealthMetrics, HealthState, HealthThresholds,
# Failover
FailoverManager, FailoverConfig, FailoverState,
BackupAssignment, BackupRegistry,
# Capability Discovery
CapabilityRegistry, Capability, CapabilityInfo, CapabilityState,
CapabilityDiscovery, DiscoveryConfig, CapabilityProbe,
# Metrics
SatelliteMetricsCollector, SatelliteMetrics, MetricValue, MetricType,
MetricsAggregator, AggregationType, TimeWindow, AggregatedMetric,
)
Satellite
Repräsentiert ein verbundenes Satellite-Gerät.
from trixy_core.satellite import Satellite, ConnectionState
# Satellite-Eigenschaften
satellite = manager.get("living_room")
print(satellite.id) # "living_room"
print(satellite.room_id) # "wohnzimmer"
print(satellite.mac_address) # "AA:BB:CC:DD:EE:FF"
print(satellite.alias) # "Wohnzimmer"
print(satellite.ip_address) # "192.168.1.100"
print(satellite.state) # ConnectionState.CONNECTED
print(satellite.is_connected) # True
print(satellite.version) # "1.0.0"
# Konversation
print(satellite.conversation_id) # "conv_123" oder None
print(satellite.has_active_conversation) # True/False
# Zeitstempel
print(satellite.connected_at) # datetime
print(satellite.last_heartbeat) # datetime
Audio-Ausgabe
# TTS (Text-to-Speech)
await satellite.speak(
text="Hallo Welt",
voice="default",
speed=1.0,
volume=1.0,
)
# Raw Audio
await satellite.say(audio_bytes)
# Musik/Streams
await satellite.play(source="music/song.mp3", volume=0.8)
await satellite.stop_playback()
# Verbindung trennen
await satellite.disconnect(reason="Manuell getrennt")
ConnectionState
from trixy_core.satellite import ConnectionState
class ConnectionState(Enum):
DISCONNECTED = "disconnected" # Nicht verbunden
CONNECTING = "connecting" # Verbindung wird aufgebaut
CONNECTED = "connected" # Verbunden
AUTHENTICATED = "authenticated" # Authentifiziert
DISCONNECTING = "disconnecting" # Verbindung wird getrennt
ERROR = "error" # Fehler
SatelliteManager
Verwaltet alle registrierten und verbundenen Satellites.
from trixy_core.satellite import SatelliteManager
manager = SatelliteManager(application)
# Statistiken
print(manager.count) # Gesamt registriert
print(manager.connected_count) # Aktuell verbunden
Abrufen
# Nach ID
satellite = manager.get("living_room")
# Nach MAC-Adresse
satellite = manager.get_by_mac("AA:BB:CC:DD:EE:FF")
# Nach Raum
satellites = manager.get_by_room("wohnzimmer")
# Gefiltert
connected = manager.get_connected()
disconnected = manager.get_disconnected()
all_satellites = manager.get_all()
# Suche mit Kriterien
results = manager.find(
room="wohnzimmer",
alias="Wohnzimmer",
connected=True,
)
# Index-Zugriff
satellite = manager[0] # Nach Position
satellite = manager["living_room"] # Nach ID
satellite = manager["room:wohnzimmer"] # Nach Raum
satellite = manager["mac:AA:BB:CC:DD:EE:FF"] # Nach MAC
Verbindungsmanagement
# Einzelner Satellite
await manager.disconnect("living_room", reason="Wartung")
await manager.reconnect("living_room")
# Alle Satellites
count = await manager.disconnect_all(reason="Server Shutdown")
count = await manager.reconnect_all()
Broadcast
# TTS an alle/bestimmte
await manager.say("Hallo", room="wohnzimmer", voice="default")
# Raw Audio Broadcast
count = await manager.say_raw(audio_bytes, room="wohnzimmer")
# Musik-Stream
session = await manager.play(
source="music/playlist.m3u",
room="wohnzimmer",
volume=0.7,
)
RegistrationManager
Verwaltet Satellite-Registrierungen (60-Sekunden-Fenster).
from trixy_core.satellite import RegistrationManager
reg_manager = RegistrationManager(application)
# Registrierungsmodus aktivieren
reg_manager.enable_registration_mode(timeout_seconds=60)
print(reg_manager.registration_mode) # True
# Registrierung starten
success = reg_manager.begin_registration(
mac_address="AA:BB:CC:DD:EE:FF",
room="wohnzimmer",
alias="Wohnzimmer",
)
# Registrierung abschließen
satellite = reg_manager.complete_registration("AA:BB:CC:DD:EE:FF")
# Registrierung abbrechen
reg_manager.cancel_registration("AA:BB:CC:DD:EE:FF")
# Registrierungsmodus deaktivieren
reg_manager.disable_registration_mode()
# Prüfen und Laden
is_reg = reg_manager.is_registered("AA:BB:CC:DD:EE:FF")
satellite = reg_manager.load_registered("AA:BB:CC:DD:EE:FF")
all_registered = reg_manager.load_all()
# Löschen
reg_manager.delete_registration("AA:BB:CC:DD:EE:FF")
SatelliteAPI
High-Level API für Plugins.
from trixy_core.satellite import get_satellite_api, SatelliteAPI
api = get_satellite_api(application)
Satellite-Zugriff
# Collection
satellites = api.satellites
# Abruf
satellite = satellites["living_room"] # Nach ID
satellite = satellites["room:wohnzimmer"] # Nach Raum-Selektor
satellite = satellites[0] # Nach Index
# Listen
all_proxies = satellites.all
connected = satellites.connected
disconnected = satellites.disconnected
# Nach Raum
room_satellites = satellites.room("wohnzimmer")
# Suche
results = satellites.find(room="wohnzimmer", connected=True)
TTS und Audio
# Text-to-Speech
await api.say(
text="Hallo Welt",
satellites=["living_room", "bedroom"], # oder None für alle
room="wohnzimmer", # oder nach Raum
voice="default",
speed=1.0,
volume=1.0,
wait=True, # Warten auf Fertigstellung
priority=0, # Höher = wichtiger
)
# Raw Audio
count = await api.say_raw(
audio_data=audio_bytes,
satellites=["living_room"],
room=None,
)
Streaming
# Stream starten
session = await api.play(
source="music/song.mp3",
satellites=["living_room"],
room=None,
volume=0.8,
)
# Stream-Management
await api.add_to_stream(session.id, satellite)
await api.remove_from_stream(session.id, satellite)
await api.stop_stream(session.id)
# Stoppen
await api.stop_satellite("living_room")
count = await api.stop_all()
# Lautstärke
await api.set_volume(satellite="living_room", volume=0.5)
SatelliteProxy
Wrapper für einzelne Satellites.
from trixy_core.satellite import SatelliteProxy
proxy = api.satellites["living_room"]
# Eigenschaften
print(proxy.id) # "living_room"
print(proxy.alias) # "Wohnzimmer"
print(proxy.room) # "wohnzimmer"
print(proxy.is_connected) # True
# Zugriff auf rohes Satellite
raw = proxy.raw
# Audio
await proxy.say("Hallo", voice="default", wait=True)
await proxy.say_raw(audio_bytes)
session = await proxy.play("music/song.mp3", volume=0.8)
await proxy.stop()
await proxy.set_volume(0.5)
Gruppen
SatelliteGroup
Logische Gruppierung von Satellites.
from trixy_core.satellite import (
SatelliteGroup, GroupManager, GroupType, GroupConfig
)
# Konfiguration
config = GroupConfig(
auto_add=True, # Automatisch hinzufügen
auto_remove=True, # Automatisch entfernen
max_members=10, # Maximum Mitglieder
priority=0, # Gruppen-Priorität
)
# Gruppe erstellen
group = SatelliteGroup(
id="living_area",
name="Wohnbereich",
group_type=GroupType.ZONE,
config=config,
)
# Mitglieder verwalten
group.add(satellite)
group.remove("living_room")
has_member = group.has("living_room")
# Mitglieder abrufen
members = group.get_members()
filtered = group.filter(lambda s: s.is_connected)
# Iteration
for member in group:
print(member.alias)
print(len(group)) # Mitglieder-Anzahl
GroupManager
from trixy_core.satellite import GroupManager
manager = GroupManager(application)
# Gruppe erstellen
group = manager.create(
group_id="living_area",
name="Wohnbereich",
group_type=GroupType.ZONE,
config=config,
)
# Abrufen
group = manager.get("living_area")
all_groups = manager.get_all()
# Suchen
by_name = manager.find_by_name("Wohnbereich")
containing = manager.find_containing("living_room")
# Löschen
manager.delete("living_area")
GroupType
from trixy_core.satellite import GroupType
class GroupType(Enum):
ROOM = "room" # Nach Raum
ZONE = "zone" # Geografische Zone
CUSTOM = "custom" # Benutzerdefiniert
DYNAMIC = "dynamic" # Dynamisch basierend auf Regeln
Zone
from trixy_core.satellite import Zone, ZoneManager, ZoneConfig
config = ZoneConfig(
location="Erdgeschoss",
boundary={"x_min": 0, "x_max": 10, "y_min": 0, "y_max": 10},
)
zone = Zone(
id="ground_floor",
name="Erdgeschoss",
config=config,
)
# ZoneManager wie GroupManager verwenden
zone_manager = ZoneManager(application)
Load-Balancing
Strategien
from trixy_core.satellite import (
LoadBalancer, LoadBalancingStrategy, SelectionResult,
RoundRobinStrategy, LeastConnectionsStrategy, WeightedStrategy,
)
# Round-Robin
strategy = RoundRobinStrategy()
# Least-Connections
strategy = LeastConnectionsStrategy()
# Weighted
from trixy_core.satellite import WeightConfig
strategy = WeightedStrategy(weights={
"living_room": WeightConfig(weight=2.0),
"bedroom": WeightConfig(weight=1.0),
})
LoadBalancer
from trixy_core.satellite import LoadBalancer
balancer = LoadBalancer(strategy=RoundRobinStrategy())
# Satellite auswählen
satellites = manager.get_connected()
result = balancer.select(satellites, context={"task": "tts"})
if result:
print(result.satellite.id) # Ausgewählter Satellite
print(result.reason) # Auswahlgrund
print(result.score) # Score (falls berechnet)
# Request-Tracking
balancer.on_request_start(result.satellite)
try:
await process(result.satellite)
finally:
balancer.on_request_end(result.satellite, success=True)
# Zurücksetzen
balancer.reset()
Eigene Strategie
from trixy_core.satellite import LoadBalancingStrategy, SelectionResult
class LatencyBasedStrategy(LoadBalancingStrategy):
@property
def name(self) -> str:
return "latency_based"
def select(
self,
satellites: list,
context: dict = None,
) -> SelectionResult | None:
if not satellites:
return None
# Niedrigste Latenz wählen
best = min(satellites, key=lambda s: s.metadata.get("latency", 999))
return SelectionResult(
satellite=best,
reason="Lowest latency",
score=1.0 / (best.metadata.get("latency", 1) + 1),
)
Health-Monitoring
HealthScorer
from trixy_core.satellite import (
HealthScorer, HealthScore, HealthMetrics, HealthState,
)
scorer = HealthScorer()
# Metriken sammeln
metrics = HealthMetrics(
latency_ms=50.0,
packet_loss=0.01,
uptime=86400.0,
error_rate=0.001,
memory_usage=0.6,
cpu_usage=0.3,
last_heartbeat_age=5.0,
connection_quality=0.95,
custom_metrics={"audio_quality": 0.9},
)
# Score berechnen
score = scorer.calculate(satellite, metrics)
print(score.overall) # 0-100
print(score.state) # HealthState.HEALTHY
print(score.is_healthy) # True
print(score.is_usable) # True
print(score.components) # Dict mit Komponenten-Scores
print(score.warnings) # Liste von Warnungen
HealthState
from trixy_core.satellite import HealthState
class HealthState(Enum):
HEALTHY = "healthy" # Alles OK
DEGRADED = "degraded" # Eingeschränkt
UNHEALTHY = "unhealthy" # Nicht OK
CRITICAL = "critical" # Kritisch
UNKNOWN = "unknown" # Unbekannt
Custom Scorer
from trixy_core.satellite import HealthScorer
scorer = HealthScorer()
def audio_quality_scorer(satellite, metrics) -> float:
quality = metrics.custom_metrics.get("audio_quality", 1.0)
return quality * 100 # Normalisiert auf 0-100
scorer.add_custom_scorer("audio", audio_quality_scorer)
scorer.remove_custom_scorer("audio")
Thresholds
from trixy_core.satellite import HealthThresholds, ThresholdConfig
thresholds = HealthThresholds(
latency=ThresholdConfig(warning=100, critical=500),
packet_loss=ThresholdConfig(warning=0.05, critical=0.2),
error_rate=ThresholdConfig(warning=0.01, critical=0.1),
)
Failover
FailoverManager
from trixy_core.satellite import (
FailoverManager, FailoverConfig, FailoverState,
)
config = FailoverConfig(
enabled=True,
auto_failover=True,
failback_delay=60.0, # Sekunden bis Failback
health_threshold=30.0, # Health-Score für Failover
)
failover = FailoverManager(config)
# Primary und Backup registrieren
failover.register_primary(primary_satellite)
failover.register_backup("living_room", backup_satellite)
# Manuelles Failover
backup = failover.trigger_failover("living_room")
if backup:
print(f"Failover zu {backup.id}")
# Backup abrufen
backup = failover.get_backup("living_room")
BackupRegistry
from trixy_core.satellite import BackupAssignment, BackupRegistry
registry = BackupRegistry()
# Zuweisung erstellen
assignment = BackupAssignment(
primary_id="living_room",
backup_id="bedroom",
priority=1,
)
registry.add(assignment)
backup_id = registry.get_backup("living_room")
FailoverState
from trixy_core.satellite import FailoverState
class FailoverState(Enum):
NORMAL = "normal" # Normalbetrieb
FAILOVER_PENDING = "pending" # Failover wird vorbereitet
FAILED_OVER = "failed_over" # Failover aktiv
FAILBACK_PENDING = "failback" # Failback wird vorbereitet
Capability Discovery
Dynamische Fähigkeitserkennung für Satellites.
CapabilityInfo
from trixy_core.satellite.capability import (
CapabilityInfo,
Capability,
CapabilityState,
CapabilityRegistry,
CapabilityDiscovery,
DiscoveryConfig,
)
# Capability definieren
info = CapabilityInfo(
id="audio.input",
name="Mikrofon",
description="Audio-Eingang für Sprachaufnahme",
category="audio",
required_features=["pyaudio"],
)
Capability
from trixy_core.satellite.capability import Capability, CapabilityState
cap = Capability(
info=info,
state=CapabilityState.AVAILABLE,
satellite_id="sat-001",
)
# Eigenschaften
print(cap.id) # "audio.input"
print(cap.state) # CapabilityState.AVAILABLE
print(cap.is_available) # True
print(cap.is_degraded) # False
# State aktualisieren
cap.update_state(
CapabilityState.UNAVAILABLE,
error_message="Gerät nicht gefunden",
)
# Als Dictionary
d = cap.to_dict()
CapabilityState
from trixy_core.satellite.capability import CapabilityState
class CapabilityState(Enum):
UNKNOWN = "unknown" # Nicht geprüft
AVAILABLE = "available" # Verfügbar
UNAVAILABLE = "unavailable" # Nicht verfügbar
DEGRADED = "degraded" # Eingeschränkt
CHECKING = "checking" # Wird geprüft
CapabilityRegistry
from trixy_core.satellite.capability import CapabilityRegistry
registry = CapabilityRegistry()
# Definition registrieren
registry.define_capability(info)
# Capability für Satellite registrieren
cap = Capability(info=info, state=CapabilityState.AVAILABLE)
registry.register("sat-001", cap)
# Abrufen
capability = registry.get("sat-001", "audio.input")
all_caps = registry.get_all("sat-001")
available = registry.get_available("sat-001")
by_category = registry.get_by_category("sat-001", "audio")
# Prüfen
has = registry.has_capability("sat-001", "audio.input")
has_available = registry.has_capability("sat-001", "audio.input", require_available=True)
has_all = registry.has_all_capabilities("sat-001", ["audio.input", "audio.output"])
has_any = registry.has_any_capability("sat-001", ["audio.input", "display"])
# Satellites finden
satellites = registry.find_satellites_with("audio.input")
satellites = registry.find_satellites_with_all(["audio.input", "audio.output"])
# State aktualisieren
registry.update_state(
"sat-001",
"audio.input",
CapabilityState.DEGRADED,
"Niedrige Qualität",
)
# Entfernen
registry.unregister("sat-001", "audio.input") # Eine Capability
registry.unregister("sat-001") # Alle Capabilities
# Callback
def on_change(sat_id: str, cap: Capability):
print(f"{sat_id}: {cap.id} -> {cap.state}")
registry.on_change(on_change)
# Zusammenfassung
summary = registry.get_summary()
print(f"Satellites: {summary['satellite_count']}")
print(f"Capabilities: {summary['total_capabilities']}")
print(f"Verfügbar: {summary['available_capabilities']}")
CapabilityDiscovery
from trixy_core.satellite.capability import (
CapabilityDiscovery,
DiscoveryConfig,
SimpleProbe,
)
config = DiscoveryConfig(
probe_timeout=10.0, # Probe-Timeout
probe_concurrency=5, # Parallele Probes
auto_discover=True, # Auto-Discovery bei Connect
rediscovery_interval=300, # Sekunden bis Re-Discovery
)
# Discovery erstellen
async def send_capability_check(sat_id: str, message: dict) -> dict:
return await satellite.send_and_wait(message)
discovery = CapabilityDiscovery(
registry=registry,
send_func=send_capability_check,
config=config,
)
# Probe registrieren
probe = SimpleProbe(info)
discovery.register_probe(probe)
# Einzelne Capability prüfen
result = await discovery.discover_capability("sat-001", "audio.input")
if result.success:
print(f"State: {result.state}")
print(f"Parameters: {result.parameters}")
# Alle Capabilities eines Satellites
results = await discovery.discover_satellite("sat-001")
# Spezifische Capabilities
results = await discovery.discover_satellite(
"sat-001",
capability_ids=["audio.input", "display"],
)
# Mehrere Satellites
all_results = await discovery.discover_all(["sat-001", "sat-002"])
# Rediscovery prüfen
needs_check = discovery.needs_rediscovery("sat-001")
# Statistiken
stats = discovery.get_stats()
print(f"Probes: {stats['probe_count']}")
Eigene Probe erstellen
from trixy_core.satellite.capability import CapabilityProbe, ProbeResult
class AudioProbe(CapabilityProbe):
def __init__(self):
super().__init__(CapabilityInfo(
id="audio.input",
name="Mikrofon",
category="audio",
))
async def probe(
self,
satellite_id: str,
send_func,
) -> ProbeResult:
try:
response = await send_func(satellite_id, {
"type": "capability_check",
"capability": "audio.input",
})
if response.get("available"):
return ProbeResult(
capability_id=self.info.id,
success=True,
state=CapabilityState.AVAILABLE,
parameters=response.get("parameters", {}),
)
else:
return ProbeResult(
capability_id=self.info.id,
success=True,
state=CapabilityState.UNAVAILABLE,
error_message=response.get("error"),
)
except Exception as e:
return ProbeResult(
capability_id=self.info.id,
success=False,
state=CapabilityState.UNKNOWN,
error_message=str(e),
)
discovery.register_probe(AudioProbe())
Metriken
Umfassende Metriken-Sammlung und Aggregation.
MetricType
from trixy_core.satellite.metrics import (
SatelliteMetricsCollector,
SatelliteMetrics,
MetricValue,
MetricType,
MetricsAggregator,
AggregationType,
TimeWindow,
AggregatedMetric,
)
class MetricType(Enum):
COUNTER = "counter" # Zähler (nur steigend)
GAUGE = "gauge" # Momentanwert
HISTOGRAM = "histogram" # Verteilung
TIMING = "timing" # Zeitdauer
SatelliteMetricsCollector
from trixy_core.satellite.metrics import SatelliteMetricsCollector
collector = SatelliteMetricsCollector()
# Metriken aufzeichnen
collector.record_latency("sat-001", 15.5)
collector.record_message_sent("sat-001", bytes_count=256)
collector.record_message_received("sat-001", bytes_count=512)
collector.record_heartbeat("sat-001")
collector.record_reconnect("sat-001")
collector.record_error("sat-001", "Connection lost")
collector.record_packet_loss("sat-001", percent=2.5)
collector.record_bandwidth("sat-001", kbps=1000.0)
# Benutzerdefinierte Metrik
collector.record_custom(
"sat-001",
name="cpu_usage",
value=45.5,
metric_type=MetricType.GAUGE,
unit="%",
)
# Metriken abrufen
metrics = collector.get_metrics("sat-001")
print(f"Latenz: {metrics.latency_ms}ms")
print(f"Gesendet: {metrics.messages_sent}")
print(f"Fehler: {metrics.error_count}")
# Custom Metriken
cpu_history = collector.get_custom_metric("sat-001", "cpu_usage")
# Alle Satellites
all_metrics = collector.get_all_metrics()
# Zusammenfassung
summary = collector.get_summary()
print(f"Satellites: {summary['satellite_count']}")
print(f"Avg Latenz: {summary['avg_latency_ms']}ms")
print(f"Fehler: {summary['total_errors']}")
# Callback für neue Metriken
def on_metric(sat_id: str, value: MetricValue):
print(f"{sat_id}: {value.name} = {value.value}")
collector.on_metric(on_metric)
# Aufräumen
collector.remove_satellite("sat-001")
collector.clear_all()
SatelliteMetrics
from trixy_core.satellite.metrics import SatelliteMetrics
metrics = SatelliteMetrics(satellite_id="sat-001")
# Latenz aufzeichnen
metrics.record_latency(15.5)
metrics.record_latency(20.0)
metrics.record_latency(10.0)
# Durchschnitt berechnen
avg = metrics.get_avg_latency(window_seconds=300)
print(f"Durchschnitt (5 Min): {avg}ms")
# Fehler aufzeichnen
metrics.record_error("Connection timeout")
print(f"Fehler: {metrics.error_count}")
print(f"Letzter Fehler: {metrics.last_error}")
# Als Dictionary
d = metrics.to_dict()
MetricsAggregator
from trixy_core.satellite.metrics import (
MetricsAggregator,
AggregationType,
TimeWindow,
)
aggregator = MetricsAggregator(collector)
# Einzelne Aggregation
result = aggregator.aggregate(
metric_name="latency",
aggregation=AggregationType.AVG,
window=TimeWindow.FIVE_MINUTES,
)
print(f"Wert: {result.value}")
print(f"Samples: {result.sample_count}")
print(f"Zeitraum: {result.start_time} - {result.end_time}")
# AggregationType
class AggregationType(Enum):
SUM = "sum" # Summe
AVG = "avg" # Durchschnitt
MIN = "min" # Minimum
MAX = "max" # Maximum
COUNT = "count" # Anzahl
MEDIAN = "median" # Median
P95 = "p95" # 95. Perzentil
P99 = "p99" # 99. Perzentil
# TimeWindow
class TimeWindow(Enum):
MINUTE = 60
FIVE_MINUTES = 300
HOUR = 3600
DAY = 86400
# Mehrfache Aggregation
results = aggregator.aggregate_multiple(
metrics=["latency", "error_count"],
aggregations=[AggregationType.AVG, AggregationType.SUM],
)
print(results["latency"]["AVG"])
print(results["error_count"]["SUM"])
# Top Satellites
top = aggregator.get_top_satellites(
metric="latency",
limit=5,
ascending=False, # Höchste zuerst
)
for sat_id, value in top:
print(f"{sat_id}: {value}ms")
# Verteilung
dist = aggregator.get_metric_distribution(
metric="latency",
buckets=10,
)
print(f"Min: {dist['min']}")
print(f"Max: {dist['max']}")
print(f"Count: {dist['count']}")
print(f"Buckets: {dist['buckets']}")
# Satellite-Vergleich
comparison = aggregator.get_satellite_comparison(
metric="latency",
satellite_ids=["sat-001", "sat-002"],
)
for sat_id, data in comparison.items():
print(f"{sat_id}: current={data['current']}, avg={data['avg']}")
# Health Summary
summary = aggregator.get_health_summary()
print(f"Status: {summary['status']}")
print(f"Satellites: {summary['satellite_count']}")
print(f"Metriken: {summary['metrics']}")
# Cache
aggregator.clear_cache()
MetricValue
from trixy_core.satellite.metrics import MetricValue
value = MetricValue(
name="latency",
value=15.5,
metric_type=MetricType.TIMING,
unit="ms",
labels={"endpoint": "/api/data"},
)
print(value.name) # "latency"
print(value.value) # 15.5
print(value.metric_type) # MetricType.TIMING
print(value.unit) # "ms"
print(value.labels) # {"endpoint": "/api/data"}
d = value.to_dict()