@@ -2059,6 +2084,27 @@
}
return;
}
+ if (msg.type === 'node_stats') {
+ const p = msg.payload || {};
+ if (_nodeStatsTarget && p.instanceId === _nodeStatsTarget) {
+ const el = document.getElementById('node-stats-smi');
+ if (el) el.textContent = p.nvidiaSmi || '(leer)';
+ }
+ return;
+ }
+ if (msg.type === 'node_stats_history') {
+ const p = msg.payload || {};
+ if (_nodeStatsTarget && p.instanceId === _nodeStatsTarget) renderNodeStatsHistory(p);
+ return;
+ }
+ if (msg.type === 'node_stats_reset_done') {
+ const p = msg.payload || {};
+ if (_nodeStatsTarget && p.instanceId === _nodeStatsTarget) {
+ renderNodeStatsHistory({ samples: [], tokenCapable: false });
+ send({ action: 'node_stats_history_request', targetInstance: _nodeStatsTarget });
+ }
+ return;
+ }
if (msg.type === 'sat_devices') {
if (msg.satellite) { satDevices[msg.satellite] = { devices: msg.devices || [], location: msg.location, ts: Date.now() }; }
satScanning = null;
@@ -4329,12 +4375,71 @@
'' + stat + '' +
'
';
}).join('');
+ // Auslastungs-Button pro Node โ Ziel bevorzugt die llm-Instanz (sieht alle
+ // GPUs + Tokens), sonst irgendein online Worker des Node.
+ const group = byNode[node];
+ const rep = group.find(w => w.service === 'llm' && w.online) || group.find(w => w.online) || group[0];
+ const btn = rep ? '
' : '';
return '
' +
- '
๐ฅ๏ธ ' + escapeHtml(node) + '
' +
+ '
๐ฅ๏ธ ' + escapeHtml(node) + btn + '
' +
rows + '
';
}).join('');
}
+ // โโ Auslastungs-Monitor (Stage E) โโโโโโโโโโโโโโโโโโโโโโโ
+ let _nodeStatsTarget = '';
+ function openNodeStats(instanceId, node) {
+ _nodeStatsTarget = instanceId;
+ document.getElementById('node-stats-node').textContent = node + ' (' + instanceId + ')';
+ document.getElementById('node-stats-smi').textContent = '(warte auf Boxโฆ)';
+ document.getElementById('node-stats-graph-gpu').innerHTML = '';
+ document.getElementById('node-stats-graph-tokens').innerHTML = '';
+ document.getElementById('node-stats-modal').style.display = 'flex';
+ send({ action: 'node_stats_history_request', targetInstance: instanceId });
+ send({ action: 'node_stats_stream_start', targetInstance: instanceId });
+ }
+ function closeNodeStats() {
+ if (_nodeStatsTarget) send({ action: 'node_stats_stream_stop', targetInstance: _nodeStatsTarget });
+ _nodeStatsTarget = '';
+ document.getElementById('node-stats-modal').style.display = 'none';
+ }
+ function resetNodeStats() {
+ if (!_nodeStatsTarget) return;
+ send({ action: 'node_stats_reset', targetInstance: _nodeStatsTarget });
+ }
+ // Simple Inline-SVG-Sparkline (kein Fremd-Lib, CSP-sicher).
+ function sparkline(values, opts) {
+ opts = opts || {};
+ const w = 320, h = 90, pad = 4, color = opts.color || '#0096FF';
+ if (!values.length) return '
(keine Daten)
';
+ const max = Math.max(opts.max || 0, ...values, 1);
+ const n = values.length;
+ const pts = values.map((v, i) => {
+ const x = pad + (n === 1 ? 0 : (i / (n - 1)) * (w - 2 * pad));
+ const y = h - pad - (Math.max(0, v) / max) * (h - 2 * pad);
+ return x.toFixed(1) + ',' + y.toFixed(1);
+ }).join(' ');
+ return '
';
+ }
+ function renderNodeStatsHistory(p) {
+ const samples = p.samples || [];
+ const gpu = samples.map(s => Number(s.gpu) || 0);
+ document.getElementById('node-stats-graph-gpu').innerHTML =
+ sparkline(gpu, { color: '#3FFF9F', max: 100, fmt: v => Math.round(v) + '%' });
+ const wrap = document.getElementById('node-stats-tokens-wrap');
+ if (p.tokenCapable) {
+ wrap.style.display = '';
+ const tok = samples.map(s => Number(s.tokens) || 0);
+ document.getElementById('node-stats-graph-tokens').innerHTML =
+ sparkline(tok, { color: '#FFD60A' });
+ } else {
+ wrap.style.display = 'none';
+ }
+ }
+
// โโ Satelliten-Ansicht โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
let satellites = []; // [{id, location, caps, control, online}]
let satDevices = {}; // id โ {devices, location, ts}
diff --git a/diagnostic/server.js b/diagnostic/server.js
index 6a8be8c..b56965e 100644
--- a/diagnostic/server.js
+++ b/diagnostic/server.js
@@ -1188,6 +1188,9 @@ function connectRVS(forcePlain) {
} else if (msg.type === "llm_provision_result") {
// Ergebnis eines Modell-Downloads/Aktivierens โ an Browser (Katalog-Status).
broadcast({ type: "llm_provision_result", payload: msg.payload || {} });
+ } else if (msg.type === "node_stats" || msg.type === "node_stats_history" || msg.type === "node_stats_reset_done") {
+ // Auslastungs-Monitor (Stage E): Box-Antworten an die Browser durchreichen.
+ broadcast({ type: msg.type, payload: msg.payload || {} });
} else if (msg.type === "audio_pcm" && msg.payload && _previewPending.size > 0) {
// PCM-Chunks einer laufenden Voice-Preview โ sammeln + WAV bauen
_handlePreviewChunk(msg.payload);
@@ -2936,6 +2939,10 @@ wss.on("connection", (ws) => {
model: msg.model || "", targetInstance: msg.targetInstance || "",
}, "llm_response", ws);
log("info", "llm", `Test-Chat โ ${msg.model || "?"} @ ${msg.targetInstance || "(broadcast)"}`);
+ } else if (msg.action === "node_stats_stream_start" || msg.action === "node_stats_stream_stop"
+ || msg.action === "node_stats_history_request" || msg.action === "node_stats_reset") {
+ // Auslastungs-Monitor (Stage E): an die Box (targetInstance) durchreichen.
+ sendToRVS_raw({ type: msg.action, payload: { targetInstance: msg.targetInstance || "" }, timestamp: Date.now() });
} else if (msg.action === "restart_session") {
handleRestartSession(ws);
// โโ Einstellungen โโ
diff --git a/xtts/docker-compose.yml b/xtts/docker-compose.yml
index 106e172..72a53bd 100644
--- a/xtts/docker-compose.yml
+++ b/xtts/docker-compose.yml
@@ -149,12 +149,15 @@ services:
build: ./llm-adapter
container_name: aria-llm-adapter
profiles: ["llm"]
+ runtime: nvidia # nur fuer nvidia-smi (Auslastungs-Monitor) โ kein Compute
depends_on:
- llama-swap
volumes:
- ./models:/models # generierte Config + Registry + Cache
- ./llama-swap:/llamaswap:ro # Basis-Template (config.yaml)
environment:
+ - NVIDIA_VISIBLE_DEVICES=all # alle Karten sichtbar (nur nvidia-smi)
+ - NVIDIA_DRIVER_CAPABILITIES=utility # utility = nvidia-smi, KEIN VRAM/Compute
- NODE_NAME=${NODE_NAME:-node}
- RVS_HOST=${RVS_HOST}
- RVS_PORT=${RVS_PORT:-443}
diff --git a/xtts/f5tts/Dockerfile b/xtts/f5tts/Dockerfile
index ac81f0d..88239ae 100644
--- a/xtts/f5tts/Dockerfile
+++ b/xtts/f5tts/Dockerfile
@@ -20,6 +20,7 @@ COPY requirements.txt .
RUN printf 'torch==2.6.0\ntorchaudio==2.6.0\n' > /tmp/torch-constraint.txt && \
pip3 install --no-cache-dir -c /tmp/torch-constraint.txt -r requirements.txt
+COPY node_stats.py .
COPY bridge.py .
CMD ["python3", "bridge.py"]
diff --git a/xtts/f5tts/bridge.py b/xtts/f5tts/bridge.py
index 768b7f7..21be3e6 100644
--- a/xtts/f5tts/bridge.py
+++ b/xtts/f5tts/bridge.py
@@ -69,6 +69,11 @@ INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
_tts_busy = False # True waehrend eine Synthese laeuft (busy-Report im ping)
+# โโ Auslastungs-Monitor (Stage E) โโโโโโโโโโโโโโโโโโโโโโโโโโ
+import node_stats
+STATS_PATH = os.getenv("STATS_PATH", f"/root/.cache/huggingface/aria_stats_{WORKER_SERVICE}.json")
+_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH, logger=logger)
+
DEFAULT_F5TTS_MODEL = "F5TTS_v1_Base"
DEFAULT_F5TTS_CKPT_FILE = "" # leer = Default-Checkpoint von HF
DEFAULT_F5TTS_VOCAB_FILE = "" # leer = Default-Vocab vom Modell
@@ -918,6 +923,9 @@ async def run_loop(runner: F5Runner) -> None:
tgt = payload.get("targetInstance")
if tgt and tgt != INSTANCE_ID:
continue
+ # Auslastungs-Monitor (node_stats_*) abfangen.
+ if await _stats.handle(ws, mtype, payload, _send):
+ continue
if mtype == "xtts_request":
try:
@@ -1040,6 +1048,7 @@ async def main() -> None:
sys.exit(1)
VOICES_DIR.mkdir(parents=True, exist_ok=True)
runner = F5Runner()
+ asyncio.create_task(_stats.run_sampler()) # Auslastungs-Sampler (Stage E)
await run_loop(runner)
diff --git a/xtts/f5tts/node_stats.py b/xtts/f5tts/node_stats.py
new file mode 100644
index 0000000..f5a7de3
--- /dev/null
+++ b/xtts/f5tts/node_stats.py
@@ -0,0 +1,167 @@
+"""
+ARIA Node-Stats โ Auslastungs-Monitor pro Box (GPU + optional Tokens).
+
+Identische Kopie in jedem Worker-Build-Context (f5tts/whisper/voxtral/llm-adapter),
+weil jeder Worker ein eigener Docker-Build-Context ist.
+
+Aufgaben:
+- Sampler-Loop (alle SAMPLE_SEC): nvidia-smi-Auslastung + Token-Delta โ Ringpuffer
+ (persistent als JSON auf der Box). Laeuft unabhaengig vom Modal.
+- Live-Stream: bei node_stats_stream_start jede Sekunde rohes nvidia-smi + Werte
+ senden (bis stop / Auto-Timeout).
+- History-Request + Reset (Besen).
+
+Reicht `handle(ws, mtype, payload)` in die Worker-Message-Loop ein; gibt True
+zurueck, wenn die Nachricht eine node_stats_*-Nachricht war.
+"""
+import asyncio
+import json
+import os
+import time
+
+SAMPLE_SEC = int(os.getenv("STATS_SAMPLE_SEC", "15"))
+HISTORY_CAP = int(os.getenv("STATS_HISTORY_CAP", "500")) # ~2h bei 15s
+STREAM_MAX_SEC = int(os.getenv("STATS_STREAM_MAX_SEC", "300"))
+
+
+async def _run_cmd(*args, timeout=8) -> str:
+ """Fuehrt ein Kommando aus, gibt stdout (str) zurueck; '' bei Fehler."""
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *args,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+ return (out or b"").decode("utf-8", "replace")
+ except Exception:
+ return ""
+
+
+class NodeStats:
+ def __init__(self, instance_id: str, node_name: str, history_path: str,
+ token_getter=None, logger=None):
+ self.instance_id = instance_id
+ self.node_name = node_name
+ self.history_path = history_path
+ self.token_getter = token_getter # callable -> kumulative Token-Zahl (oder None)
+ self.log = logger
+ self.samples = self._load()
+ self._last_tokens = self._tokens_now()
+ self._stream_task = None
+
+ # โโ Persistenz โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ def _load(self) -> list:
+ try:
+ with open(self.history_path) as f:
+ data = json.load(f)
+ return data if isinstance(data, list) else []
+ except Exception:
+ return []
+
+ def _persist(self) -> None:
+ try:
+ os.makedirs(os.path.dirname(self.history_path) or ".", exist_ok=True)
+ tmp = self.history_path + ".tmp"
+ with open(tmp, "w") as f:
+ json.dump(self.samples[-HISTORY_CAP:], f)
+ os.replace(tmp, self.history_path)
+ except Exception:
+ pass
+
+ def _tokens_now(self) -> int:
+ try:
+ return int(self.token_getter()) if self.token_getter else 0
+ except Exception:
+ return 0
+
+ # โโ nvidia-smi โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _query_gpu(self) -> dict:
+ """Aggregierte GPU-Werte ueber alle sichtbaren Karten."""
+ out = await _run_cmd(
+ "nvidia-smi",
+ "--query-gpu=utilization.gpu,memory.used,memory.total",
+ "--format=csv,noheader,nounits")
+ utils, used, total = [], 0, 0
+ for line in out.strip().splitlines():
+ parts = [p.strip() for p in line.split(",")]
+ if len(parts) < 3:
+ continue
+ try:
+ utils.append(float(parts[0]))
+ used += float(parts[1])
+ total += float(parts[2])
+ except ValueError:
+ continue
+ gpu = round(sum(utils) / len(utils), 1) if utils else 0.0
+ return {"gpu": gpu, "memUsed": int(used), "memTotal": int(total)}
+
+ async def _nvidia_smi_text(self) -> str:
+ txt = await _run_cmd("nvidia-smi")
+ return txt or "nvidia-smi nicht verfuegbar"
+
+ # โโ Sampler (Verlauf) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def run_sampler(self) -> None:
+ while True:
+ try:
+ g = await self._query_gpu()
+ now_tok = self._tokens_now()
+ dtok = max(0, now_tok - self._last_tokens)
+ self._last_tokens = now_tok
+ self.samples.append({
+ "ts": int(time.time()),
+ "gpu": g["gpu"], "memUsed": g["memUsed"],
+ "memTotal": g["memTotal"], "tokens": dtok,
+ })
+ if len(self.samples) > HISTORY_CAP:
+ self.samples = self.samples[-HISTORY_CAP:]
+ self._persist()
+ except Exception as e:
+ if self.log:
+ self.log.debug("node_stats sample fehlgeschlagen: %s", e)
+ await asyncio.sleep(SAMPLE_SEC)
+
+ # โโ Live-Stream โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _stream(self, ws, send) -> None:
+ t0 = time.time()
+ try:
+ while time.time() - t0 < STREAM_MAX_SEC:
+ g = await self._query_gpu()
+ smi = await self._nvidia_smi_text()
+ await send(ws, "node_stats", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "nvidiaSmi": smi, **g,
+ })
+ await asyncio.sleep(1)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ return
+
+ # โโ Dispatch โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def handle(self, ws, mtype: str, payload: dict, send) -> bool:
+ if mtype == "node_stats_stream_start":
+ if self._stream_task and not self._stream_task.done():
+ self._stream_task.cancel()
+ self._stream_task = asyncio.create_task(self._stream(ws, send))
+ return True
+ if mtype == "node_stats_stream_stop":
+ if self._stream_task:
+ self._stream_task.cancel()
+ self._stream_task = None
+ return True
+ if mtype == "node_stats_history_request":
+ await send(ws, "node_stats_history", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "samples": self.samples[-HISTORY_CAP:],
+ "tokenCapable": self.token_getter is not None,
+ "sampleSec": SAMPLE_SEC,
+ })
+ return True
+ if mtype == "node_stats_reset":
+ self.samples = []
+ self._persist()
+ await send(ws, "node_stats_reset_done",
+ {"instanceId": self.instance_id, "node": self.node_name})
+ return True
+ return False
diff --git a/xtts/llm-adapter/Dockerfile b/xtts/llm-adapter/Dockerfile
index 493d942..efb51a9 100644
--- a/xtts/llm-adapter/Dockerfile
+++ b/xtts/llm-adapter/Dockerfile
@@ -3,6 +3,7 @@ FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
+COPY node_stats.py .
COPY adapter.py .
CMD ["python", "-u", "adapter.py"]
diff --git a/xtts/llm-adapter/adapter.py b/xtts/llm-adapter/adapter.py
index caaaf76..eebbbee 100644
--- a/xtts/llm-adapter/adapter.py
+++ b/xtts/llm-adapter/adapter.py
@@ -75,6 +75,13 @@ BASE_CONFIG_PATH = os.getenv("LLAMA_BASE_CONFIG", "/llamaswap/config.yaml")
GEN_CONFIG_PATH = os.getenv("LLAMA_GEN_CONFIG", "/models/llama-swap.config.yaml")
REGISTRY_PATH = os.getenv("LLM_REGISTRY", "/models/aria_models.json")
+# โโ Auslastungs-Monitor (Stage E) โโโโโโโโโโโโโโโโโโโโโโโโโโ
+import node_stats
+STATS_PATH = os.getenv("STATS_PATH", "/models/aria_stats.json")
+_total_tokens = 0 # kumulativ, fuer den Token-Graph
+_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH,
+ token_getter=lambda: _total_tokens, logger=logger)
+
def _load_registry() -> list:
try:
@@ -189,11 +196,17 @@ async def _call_llama(messages: list, *, max_tokens: int, temperature: float,
r.raise_for_status()
data = r.json()
msg = (data.get("choices") or [{}])[0].get("message", {}) or {}
+ usage = data.get("usage") or {}
+ try:
+ global _total_tokens
+ _total_tokens += int(usage.get("total_tokens") or 0)
+ except Exception:
+ pass
return {
"ok": True,
"content": msg.get("content") or "",
"tool_calls": msg.get("tool_calls") or None,
- "usage": data.get("usage"),
+ "usage": usage,
}
except Exception as e:
logger.warning("llama.cpp-Call fehlgeschlagen: %s", e)
@@ -387,6 +400,9 @@ async def _run() -> None:
# Reihenfolge ab, falls es kurz vor uns startet).
_generate_config()
+ # Auslastungs-Sampler (GPU + Tokens) laeuft unabhaengig vom RVS.
+ asyncio.create_task(_stats.run_sampler())
+
use_tls = RVS_TLS
retry_s = 2
tls_fallback_tried = False
@@ -410,14 +426,17 @@ async def _run() -> None:
except Exception:
continue
mtype = msg.get("type")
- if mtype not in ("llm_request", "llm_provision_model", "llm_remove_model"):
- continue
payload = msg.get("payload", {}) or {}
# Redundanz-Routing: gezielt an eine andere Instanz adressiert
# โ ignorieren. Ohne targetInstance โ wie bisher (jeder nimmt).
tgt = payload.get("targetInstance")
if tgt and tgt != INSTANCE_ID:
continue
+ # Auslastungs-Monitor (node_stats_*) abfangen.
+ if await _stats.handle(ws, mtype, payload, _send):
+ continue
+ if mtype not in ("llm_request", "llm_provision_model", "llm_remove_model"):
+ continue
if mtype == "llm_provision_model":
asyncio.create_task(_handle_provision(ws, payload))
elif mtype == "llm_remove_model":
diff --git a/xtts/llm-adapter/node_stats.py b/xtts/llm-adapter/node_stats.py
new file mode 100644
index 0000000..f5a7de3
--- /dev/null
+++ b/xtts/llm-adapter/node_stats.py
@@ -0,0 +1,167 @@
+"""
+ARIA Node-Stats โ Auslastungs-Monitor pro Box (GPU + optional Tokens).
+
+Identische Kopie in jedem Worker-Build-Context (f5tts/whisper/voxtral/llm-adapter),
+weil jeder Worker ein eigener Docker-Build-Context ist.
+
+Aufgaben:
+- Sampler-Loop (alle SAMPLE_SEC): nvidia-smi-Auslastung + Token-Delta โ Ringpuffer
+ (persistent als JSON auf der Box). Laeuft unabhaengig vom Modal.
+- Live-Stream: bei node_stats_stream_start jede Sekunde rohes nvidia-smi + Werte
+ senden (bis stop / Auto-Timeout).
+- History-Request + Reset (Besen).
+
+Reicht `handle(ws, mtype, payload)` in die Worker-Message-Loop ein; gibt True
+zurueck, wenn die Nachricht eine node_stats_*-Nachricht war.
+"""
+import asyncio
+import json
+import os
+import time
+
+SAMPLE_SEC = int(os.getenv("STATS_SAMPLE_SEC", "15"))
+HISTORY_CAP = int(os.getenv("STATS_HISTORY_CAP", "500")) # ~2h bei 15s
+STREAM_MAX_SEC = int(os.getenv("STATS_STREAM_MAX_SEC", "300"))
+
+
+async def _run_cmd(*args, timeout=8) -> str:
+ """Fuehrt ein Kommando aus, gibt stdout (str) zurueck; '' bei Fehler."""
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *args,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+ return (out or b"").decode("utf-8", "replace")
+ except Exception:
+ return ""
+
+
+class NodeStats:
+ def __init__(self, instance_id: str, node_name: str, history_path: str,
+ token_getter=None, logger=None):
+ self.instance_id = instance_id
+ self.node_name = node_name
+ self.history_path = history_path
+ self.token_getter = token_getter # callable -> kumulative Token-Zahl (oder None)
+ self.log = logger
+ self.samples = self._load()
+ self._last_tokens = self._tokens_now()
+ self._stream_task = None
+
+ # โโ Persistenz โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ def _load(self) -> list:
+ try:
+ with open(self.history_path) as f:
+ data = json.load(f)
+ return data if isinstance(data, list) else []
+ except Exception:
+ return []
+
+ def _persist(self) -> None:
+ try:
+ os.makedirs(os.path.dirname(self.history_path) or ".", exist_ok=True)
+ tmp = self.history_path + ".tmp"
+ with open(tmp, "w") as f:
+ json.dump(self.samples[-HISTORY_CAP:], f)
+ os.replace(tmp, self.history_path)
+ except Exception:
+ pass
+
+ def _tokens_now(self) -> int:
+ try:
+ return int(self.token_getter()) if self.token_getter else 0
+ except Exception:
+ return 0
+
+ # โโ nvidia-smi โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _query_gpu(self) -> dict:
+ """Aggregierte GPU-Werte ueber alle sichtbaren Karten."""
+ out = await _run_cmd(
+ "nvidia-smi",
+ "--query-gpu=utilization.gpu,memory.used,memory.total",
+ "--format=csv,noheader,nounits")
+ utils, used, total = [], 0, 0
+ for line in out.strip().splitlines():
+ parts = [p.strip() for p in line.split(",")]
+ if len(parts) < 3:
+ continue
+ try:
+ utils.append(float(parts[0]))
+ used += float(parts[1])
+ total += float(parts[2])
+ except ValueError:
+ continue
+ gpu = round(sum(utils) / len(utils), 1) if utils else 0.0
+ return {"gpu": gpu, "memUsed": int(used), "memTotal": int(total)}
+
+ async def _nvidia_smi_text(self) -> str:
+ txt = await _run_cmd("nvidia-smi")
+ return txt or "nvidia-smi nicht verfuegbar"
+
+ # โโ Sampler (Verlauf) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def run_sampler(self) -> None:
+ while True:
+ try:
+ g = await self._query_gpu()
+ now_tok = self._tokens_now()
+ dtok = max(0, now_tok - self._last_tokens)
+ self._last_tokens = now_tok
+ self.samples.append({
+ "ts": int(time.time()),
+ "gpu": g["gpu"], "memUsed": g["memUsed"],
+ "memTotal": g["memTotal"], "tokens": dtok,
+ })
+ if len(self.samples) > HISTORY_CAP:
+ self.samples = self.samples[-HISTORY_CAP:]
+ self._persist()
+ except Exception as e:
+ if self.log:
+ self.log.debug("node_stats sample fehlgeschlagen: %s", e)
+ await asyncio.sleep(SAMPLE_SEC)
+
+ # โโ Live-Stream โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _stream(self, ws, send) -> None:
+ t0 = time.time()
+ try:
+ while time.time() - t0 < STREAM_MAX_SEC:
+ g = await self._query_gpu()
+ smi = await self._nvidia_smi_text()
+ await send(ws, "node_stats", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "nvidiaSmi": smi, **g,
+ })
+ await asyncio.sleep(1)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ return
+
+ # โโ Dispatch โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def handle(self, ws, mtype: str, payload: dict, send) -> bool:
+ if mtype == "node_stats_stream_start":
+ if self._stream_task and not self._stream_task.done():
+ self._stream_task.cancel()
+ self._stream_task = asyncio.create_task(self._stream(ws, send))
+ return True
+ if mtype == "node_stats_stream_stop":
+ if self._stream_task:
+ self._stream_task.cancel()
+ self._stream_task = None
+ return True
+ if mtype == "node_stats_history_request":
+ await send(ws, "node_stats_history", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "samples": self.samples[-HISTORY_CAP:],
+ "tokenCapable": self.token_getter is not None,
+ "sampleSec": SAMPLE_SEC,
+ })
+ return True
+ if mtype == "node_stats_reset":
+ self.samples = []
+ self._persist()
+ await send(ws, "node_stats_reset_done",
+ {"instanceId": self.instance_id, "node": self.node_name})
+ return True
+ return False
diff --git a/xtts/voxtral/Dockerfile b/xtts/voxtral/Dockerfile
index 976507e..503496b 100644
--- a/xtts/voxtral/Dockerfile
+++ b/xtts/voxtral/Dockerfile
@@ -21,6 +21,6 @@ COPY requirements.txt .
RUN printf 'torch==2.6.0\ntorchaudio==2.6.0\n' > /tmp/torch-constraint.txt && \
pip3 install --no-cache-dir -c /tmp/torch-constraint.txt -r requirements.txt
-COPY bridge.py speaker_id.py ./
+COPY bridge.py speaker_id.py node_stats.py ./
CMD ["python3", "bridge.py"]
diff --git a/xtts/voxtral/bridge.py b/xtts/voxtral/bridge.py
index f1cc02b..29dc5f6 100644
--- a/xtts/voxtral/bridge.py
+++ b/xtts/voxtral/bridge.py
@@ -68,6 +68,11 @@ WORKER_SERVICE = "voxtral"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
+# โโ Auslastungs-Monitor (Stage E) โโโโโโโโโโโโโโโโโโโโโโโโโโ
+import node_stats
+STATS_PATH = os.getenv("STATS_PATH", f"/root/.cache/huggingface/aria_stats_{WORKER_SERVICE}.json")
+_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH, logger=logger)
+
STREAM_TRANSCRIBE_INTERVAL_MS = int(os.getenv("STREAM_TRANSCRIBE_INTERVAL_MS", "1000"))
STREAM_DEFAULT_ENDPOINT_MS = 2400
STREAM_DEFAULT_HARD_CAP_MS = 300000
@@ -757,6 +762,9 @@ async def run_loop(sessions: SessionManager) -> None:
tgt = payload.get("targetInstance")
if tgt and tgt != INSTANCE_ID:
continue
+ # Auslastungs-Monitor (node_stats_*) abfangen.
+ if await _stats.handle(ws, mtype, payload, _send):
+ continue
if mtype == "stt_stream_start":
sessions.start_session(payload)
elif mtype == "stt_audio_chunk":
@@ -856,6 +864,7 @@ async def main() -> None:
await loop.run_in_executor(None, runner.load) # Modell laden (blockierend)
sessions = SessionManager(runner)
logger.info("Voxtral-Bridge startet โ Modell=%s", VOXTRAL_MODEL)
+ asyncio.create_task(_stats.run_sampler()) # Auslastungs-Sampler (Stage E)
await asyncio.gather(run_loop(sessions), sessions.run_endpointer())
diff --git a/xtts/voxtral/node_stats.py b/xtts/voxtral/node_stats.py
new file mode 100644
index 0000000..f5a7de3
--- /dev/null
+++ b/xtts/voxtral/node_stats.py
@@ -0,0 +1,167 @@
+"""
+ARIA Node-Stats โ Auslastungs-Monitor pro Box (GPU + optional Tokens).
+
+Identische Kopie in jedem Worker-Build-Context (f5tts/whisper/voxtral/llm-adapter),
+weil jeder Worker ein eigener Docker-Build-Context ist.
+
+Aufgaben:
+- Sampler-Loop (alle SAMPLE_SEC): nvidia-smi-Auslastung + Token-Delta โ Ringpuffer
+ (persistent als JSON auf der Box). Laeuft unabhaengig vom Modal.
+- Live-Stream: bei node_stats_stream_start jede Sekunde rohes nvidia-smi + Werte
+ senden (bis stop / Auto-Timeout).
+- History-Request + Reset (Besen).
+
+Reicht `handle(ws, mtype, payload)` in die Worker-Message-Loop ein; gibt True
+zurueck, wenn die Nachricht eine node_stats_*-Nachricht war.
+"""
+import asyncio
+import json
+import os
+import time
+
+SAMPLE_SEC = int(os.getenv("STATS_SAMPLE_SEC", "15"))
+HISTORY_CAP = int(os.getenv("STATS_HISTORY_CAP", "500")) # ~2h bei 15s
+STREAM_MAX_SEC = int(os.getenv("STATS_STREAM_MAX_SEC", "300"))
+
+
+async def _run_cmd(*args, timeout=8) -> str:
+ """Fuehrt ein Kommando aus, gibt stdout (str) zurueck; '' bei Fehler."""
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *args,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+ return (out or b"").decode("utf-8", "replace")
+ except Exception:
+ return ""
+
+
+class NodeStats:
+ def __init__(self, instance_id: str, node_name: str, history_path: str,
+ token_getter=None, logger=None):
+ self.instance_id = instance_id
+ self.node_name = node_name
+ self.history_path = history_path
+ self.token_getter = token_getter # callable -> kumulative Token-Zahl (oder None)
+ self.log = logger
+ self.samples = self._load()
+ self._last_tokens = self._tokens_now()
+ self._stream_task = None
+
+ # โโ Persistenz โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ def _load(self) -> list:
+ try:
+ with open(self.history_path) as f:
+ data = json.load(f)
+ return data if isinstance(data, list) else []
+ except Exception:
+ return []
+
+ def _persist(self) -> None:
+ try:
+ os.makedirs(os.path.dirname(self.history_path) or ".", exist_ok=True)
+ tmp = self.history_path + ".tmp"
+ with open(tmp, "w") as f:
+ json.dump(self.samples[-HISTORY_CAP:], f)
+ os.replace(tmp, self.history_path)
+ except Exception:
+ pass
+
+ def _tokens_now(self) -> int:
+ try:
+ return int(self.token_getter()) if self.token_getter else 0
+ except Exception:
+ return 0
+
+ # โโ nvidia-smi โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _query_gpu(self) -> dict:
+ """Aggregierte GPU-Werte ueber alle sichtbaren Karten."""
+ out = await _run_cmd(
+ "nvidia-smi",
+ "--query-gpu=utilization.gpu,memory.used,memory.total",
+ "--format=csv,noheader,nounits")
+ utils, used, total = [], 0, 0
+ for line in out.strip().splitlines():
+ parts = [p.strip() for p in line.split(",")]
+ if len(parts) < 3:
+ continue
+ try:
+ utils.append(float(parts[0]))
+ used += float(parts[1])
+ total += float(parts[2])
+ except ValueError:
+ continue
+ gpu = round(sum(utils) / len(utils), 1) if utils else 0.0
+ return {"gpu": gpu, "memUsed": int(used), "memTotal": int(total)}
+
+ async def _nvidia_smi_text(self) -> str:
+ txt = await _run_cmd("nvidia-smi")
+ return txt or "nvidia-smi nicht verfuegbar"
+
+ # โโ Sampler (Verlauf) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def run_sampler(self) -> None:
+ while True:
+ try:
+ g = await self._query_gpu()
+ now_tok = self._tokens_now()
+ dtok = max(0, now_tok - self._last_tokens)
+ self._last_tokens = now_tok
+ self.samples.append({
+ "ts": int(time.time()),
+ "gpu": g["gpu"], "memUsed": g["memUsed"],
+ "memTotal": g["memTotal"], "tokens": dtok,
+ })
+ if len(self.samples) > HISTORY_CAP:
+ self.samples = self.samples[-HISTORY_CAP:]
+ self._persist()
+ except Exception as e:
+ if self.log:
+ self.log.debug("node_stats sample fehlgeschlagen: %s", e)
+ await asyncio.sleep(SAMPLE_SEC)
+
+ # โโ Live-Stream โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _stream(self, ws, send) -> None:
+ t0 = time.time()
+ try:
+ while time.time() - t0 < STREAM_MAX_SEC:
+ g = await self._query_gpu()
+ smi = await self._nvidia_smi_text()
+ await send(ws, "node_stats", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "nvidiaSmi": smi, **g,
+ })
+ await asyncio.sleep(1)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ return
+
+ # โโ Dispatch โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def handle(self, ws, mtype: str, payload: dict, send) -> bool:
+ if mtype == "node_stats_stream_start":
+ if self._stream_task and not self._stream_task.done():
+ self._stream_task.cancel()
+ self._stream_task = asyncio.create_task(self._stream(ws, send))
+ return True
+ if mtype == "node_stats_stream_stop":
+ if self._stream_task:
+ self._stream_task.cancel()
+ self._stream_task = None
+ return True
+ if mtype == "node_stats_history_request":
+ await send(ws, "node_stats_history", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "samples": self.samples[-HISTORY_CAP:],
+ "tokenCapable": self.token_getter is not None,
+ "sampleSec": SAMPLE_SEC,
+ })
+ return True
+ if mtype == "node_stats_reset":
+ self.samples = []
+ self._persist()
+ await send(ws, "node_stats_reset_done",
+ {"instanceId": self.instance_id, "node": self.node_name})
+ return True
+ return False
diff --git a/xtts/whisper/Dockerfile b/xtts/whisper/Dockerfile
index da0ad6b..4ca4939 100644
--- a/xtts/whisper/Dockerfile
+++ b/xtts/whisper/Dockerfile
@@ -17,6 +17,6 @@ RUN pip3 install --no-cache-dir torch==2.3.1 torchaudio==2.3.1 \
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
-COPY bridge.py speaker_id.py ./
+COPY bridge.py speaker_id.py node_stats.py ./
CMD ["python3", "bridge.py"]
diff --git a/xtts/whisper/bridge.py b/xtts/whisper/bridge.py
index 29d0252..2fb4ba5 100644
--- a/xtts/whisper/bridge.py
+++ b/xtts/whisper/bridge.py
@@ -67,6 +67,11 @@ WORKER_SERVICE = "whisper"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
+# โโ Auslastungs-Monitor (Stage E) โโโโโโโโโโโโโโโโโโโโโโโโโโ
+import node_stats
+STATS_PATH = os.getenv("STATS_PATH", f"/root/.cache/huggingface/aria_stats_{WORKER_SERVICE}.json")
+_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH, logger=logger)
+
ALLOWED_MODELS = {"tiny", "base", "small", "medium", "large-v3"}
# Streaming-Parameter (Defaults โ koennen pro Session vom App-Payload ueberschrieben werden)
@@ -904,6 +909,9 @@ async def run_loop(runner: WhisperRunner, sessions: SessionManager) -> None:
tgt = payload.get("targetInstance")
if tgt and tgt != INSTANCE_ID:
continue
+ # Auslastungs-Monitor (node_stats_*) abfangen.
+ if await _stats.handle(ws, mtype, payload, _send):
+ continue
if mtype == "stt_request":
req_id = payload.get("requestId", "?")
@@ -1096,6 +1104,7 @@ async def main() -> None:
# Endpointer-Loop nebenbei laufen lassen โ er pruefst _ws is None und
# schlaeft solange das nicht gesetzt ist.
asyncio.create_task(sessions.run_endpointer())
+ asyncio.create_task(_stats.run_sampler()) # Auslastungs-Sampler (Stage E)
await run_loop(runner, sessions)
diff --git a/xtts/whisper/node_stats.py b/xtts/whisper/node_stats.py
new file mode 100644
index 0000000..f5a7de3
--- /dev/null
+++ b/xtts/whisper/node_stats.py
@@ -0,0 +1,167 @@
+"""
+ARIA Node-Stats โ Auslastungs-Monitor pro Box (GPU + optional Tokens).
+
+Identische Kopie in jedem Worker-Build-Context (f5tts/whisper/voxtral/llm-adapter),
+weil jeder Worker ein eigener Docker-Build-Context ist.
+
+Aufgaben:
+- Sampler-Loop (alle SAMPLE_SEC): nvidia-smi-Auslastung + Token-Delta โ Ringpuffer
+ (persistent als JSON auf der Box). Laeuft unabhaengig vom Modal.
+- Live-Stream: bei node_stats_stream_start jede Sekunde rohes nvidia-smi + Werte
+ senden (bis stop / Auto-Timeout).
+- History-Request + Reset (Besen).
+
+Reicht `handle(ws, mtype, payload)` in die Worker-Message-Loop ein; gibt True
+zurueck, wenn die Nachricht eine node_stats_*-Nachricht war.
+"""
+import asyncio
+import json
+import os
+import time
+
+SAMPLE_SEC = int(os.getenv("STATS_SAMPLE_SEC", "15"))
+HISTORY_CAP = int(os.getenv("STATS_HISTORY_CAP", "500")) # ~2h bei 15s
+STREAM_MAX_SEC = int(os.getenv("STATS_STREAM_MAX_SEC", "300"))
+
+
+async def _run_cmd(*args, timeout=8) -> str:
+ """Fuehrt ein Kommando aus, gibt stdout (str) zurueck; '' bei Fehler."""
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *args,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+ return (out or b"").decode("utf-8", "replace")
+ except Exception:
+ return ""
+
+
+class NodeStats:
+ def __init__(self, instance_id: str, node_name: str, history_path: str,
+ token_getter=None, logger=None):
+ self.instance_id = instance_id
+ self.node_name = node_name
+ self.history_path = history_path
+ self.token_getter = token_getter # callable -> kumulative Token-Zahl (oder None)
+ self.log = logger
+ self.samples = self._load()
+ self._last_tokens = self._tokens_now()
+ self._stream_task = None
+
+ # โโ Persistenz โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ def _load(self) -> list:
+ try:
+ with open(self.history_path) as f:
+ data = json.load(f)
+ return data if isinstance(data, list) else []
+ except Exception:
+ return []
+
+ def _persist(self) -> None:
+ try:
+ os.makedirs(os.path.dirname(self.history_path) or ".", exist_ok=True)
+ tmp = self.history_path + ".tmp"
+ with open(tmp, "w") as f:
+ json.dump(self.samples[-HISTORY_CAP:], f)
+ os.replace(tmp, self.history_path)
+ except Exception:
+ pass
+
+ def _tokens_now(self) -> int:
+ try:
+ return int(self.token_getter()) if self.token_getter else 0
+ except Exception:
+ return 0
+
+ # โโ nvidia-smi โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _query_gpu(self) -> dict:
+ """Aggregierte GPU-Werte ueber alle sichtbaren Karten."""
+ out = await _run_cmd(
+ "nvidia-smi",
+ "--query-gpu=utilization.gpu,memory.used,memory.total",
+ "--format=csv,noheader,nounits")
+ utils, used, total = [], 0, 0
+ for line in out.strip().splitlines():
+ parts = [p.strip() for p in line.split(",")]
+ if len(parts) < 3:
+ continue
+ try:
+ utils.append(float(parts[0]))
+ used += float(parts[1])
+ total += float(parts[2])
+ except ValueError:
+ continue
+ gpu = round(sum(utils) / len(utils), 1) if utils else 0.0
+ return {"gpu": gpu, "memUsed": int(used), "memTotal": int(total)}
+
+ async def _nvidia_smi_text(self) -> str:
+ txt = await _run_cmd("nvidia-smi")
+ return txt or "nvidia-smi nicht verfuegbar"
+
+ # โโ Sampler (Verlauf) โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def run_sampler(self) -> None:
+ while True:
+ try:
+ g = await self._query_gpu()
+ now_tok = self._tokens_now()
+ dtok = max(0, now_tok - self._last_tokens)
+ self._last_tokens = now_tok
+ self.samples.append({
+ "ts": int(time.time()),
+ "gpu": g["gpu"], "memUsed": g["memUsed"],
+ "memTotal": g["memTotal"], "tokens": dtok,
+ })
+ if len(self.samples) > HISTORY_CAP:
+ self.samples = self.samples[-HISTORY_CAP:]
+ self._persist()
+ except Exception as e:
+ if self.log:
+ self.log.debug("node_stats sample fehlgeschlagen: %s", e)
+ await asyncio.sleep(SAMPLE_SEC)
+
+ # โโ Live-Stream โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def _stream(self, ws, send) -> None:
+ t0 = time.time()
+ try:
+ while time.time() - t0 < STREAM_MAX_SEC:
+ g = await self._query_gpu()
+ smi = await self._nvidia_smi_text()
+ await send(ws, "node_stats", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "nvidiaSmi": smi, **g,
+ })
+ await asyncio.sleep(1)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ return
+
+ # โโ Dispatch โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ async def handle(self, ws, mtype: str, payload: dict, send) -> bool:
+ if mtype == "node_stats_stream_start":
+ if self._stream_task and not self._stream_task.done():
+ self._stream_task.cancel()
+ self._stream_task = asyncio.create_task(self._stream(ws, send))
+ return True
+ if mtype == "node_stats_stream_stop":
+ if self._stream_task:
+ self._stream_task.cancel()
+ self._stream_task = None
+ return True
+ if mtype == "node_stats_history_request":
+ await send(ws, "node_stats_history", {
+ "instanceId": self.instance_id, "node": self.node_name,
+ "samples": self.samples[-HISTORY_CAP:],
+ "tokenCapable": self.token_getter is not None,
+ "sampleSec": SAMPLE_SEC,
+ })
+ return True
+ if mtype == "node_stats_reset":
+ self.samples = []
+ self._persist()
+ await send(ws, "node_stats_reset_done",
+ {"instanceId": self.instance_id, "node": self.node_name})
+ return True
+ return False