Code source de mirza.server.widgetServer
"""
Serveur ASGI FastAPI et passerelle WebSocket pour l'interface HUD React de Mirza.
Expose les routes REST pour les metriques, la memoire et le KV Cache,
distribue les fichiers statiques du widget React compile, et maintient
le canal WebSocket bidirectionnel pour le micro-streaming binaire et le routage multi-clients.
"""
from __future__ import annotations
import asyncio
import os
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import UTC, datetime
import psutil
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from ..core.logger import getMirzaLogger
from ..core.memory.thermal import getThermalSnapshot
from ..core.memoryManager import MemoryManager
from ..core.networkUtils import (
getLocalIp,
getTailscaleInfo,
getTailscaleIp,
pcm16ToWavBytes,
)
from ..core.telemetry import TelemetryManager
from ..pipeline.conversationOrchestrator import ConversationOrchestrator
from .eventBroadcaster import EventBroadcaster
from .routes.memories import createMemoriesRouter
from .routes.system import createSystemRouter
from .routes.websocket import createWebSocketRouter
from .sslManager import SslManager
logger = getMirzaLogger("mirza.server.widget")
# Re-exports pour la compatibilite des appelants aval.
__all__ = [
"WidgetServer",
"createWidgetApp",
"getLocalIp",
"getTailscaleInfo",
"getTailscaleIp",
"pcm16ToWavBytes",
]
[docs]
def createWidgetApp(orchestrator: ConversationOrchestrator, broadcaster: EventBroadcaster) -> FastAPI:
"""
Instancie et configure l'application FastAPI et les routes WebSocket de Mirza.
Parameters
----------
orchestrator : ConversationOrchestrator
Orchestrateur central gerant VAD, ASR, LLM et TTS.
broadcaster : EventBroadcaster
Diffuseur d'evenements temps reel pour les clients connectes.
Returns
-------
FastAPI
Application ASGI configuree prete a etre servie par Uvicorn.
"""
def emitLog(source: str, message: str, level: str = "INFO") -> None:
timestamp = datetime.now(UTC).strftime("%H:%M:%S")
broadcaster.broadcastEvent("log", {
"time": timestamp,
"source": source,
"level": level,
"message": message,
})
def onState(state: str) -> None:
broadcaster.broadcastEvent("stateChange", {"state": state})
emitLog("STATE", f"Changement d'etat : {state}", "INFO")
def onAudio(rms: float, prob: float) -> None:
broadcaster.broadcastEvent("audioLevel", {"rms": round(rms, 4), "speechProbability": round(prob, 3)})
def onTranscribe(text: str) -> None:
broadcaster.broadcastEvent("transcription", {"text": text})
emitLog("STT", f"Transcription : \"{text}\"", "INFO")
def onToken(token: str) -> None:
broadcaster.broadcastEvent("token", {"token": token})
def onThought(token: str) -> None:
broadcaster.broadcastEvent("thoughtToken", {"token": token})
def onComplete(resp: str) -> None:
broadcaster.broadcastEvent("responseComplete", {"text": resp})
emitLog("LLM", f"Reponse terminee ({len(resp.split())} mots)", "INFO")
def onMetrics(metrics: dict) -> None:
ttft = metrics.get('ttftMs', 0)
tps = metrics.get('tokensPerSec', 0)
kv = metrics.get('kvCacheTokens', 0)
TelemetryManager.record_ttft(ttft)
TelemetryManager.record_speed(tps)
TelemetryManager.update_kv_cache(kv)
broadcaster.broadcastEvent("metrics", metrics)
emitLog("METRICS", f"TTFT: {ttft}ms | Vitesse: {tps} tok/s | KV: {kv}/{metrics.get('kvCacheMax')}", "INFO")
def onSpeechAudio(wavBytes: bytes, clientTargetId: str | None = None) -> None:
import base64
audioB64 = base64.b64encode(wavBytes).decode("ascii")
payload: dict[str, str] = {
"audioBase64": audioB64,
"format": "wav",
}
if clientTargetId:
payload["clientTargetId"] = clientTargetId
broadcaster.broadcastEvent("speechAudio", payload)
orchestrator.onStateChange = onState
orchestrator.onAudioLevel = onAudio
orchestrator.onTranscription = onTranscribe
orchestrator.onTokenStream = onToken
orchestrator.onThoughtStream = onThought
orchestrator.onAssistantResponseComplete = onComplete
orchestrator.onMetrics = onMetrics
orchestrator.onSpeechAudioGenerated = onSpeechAudio
orchestrator.onAgentDoc = lambda text: broadcaster.broadcastEvent("agentDoc", {"text": text})
orchestrator.onCodeGenerating = lambda active: broadcaster.broadcastEvent("codeGenerating", {"active": active})
orchestrator.onMemoryUpdated = lambda status: broadcaster.broadcastEvent("memoryStatus", status)
orchestrator.onActiveContextUpdated = lambda ctx: broadcaster.broadcastEvent("activeContext", ctx)
orchestrator.onLog = emitLog
if orchestrator.agentManager is not None:
orchestrator.agentManager.broadcaster = broadcaster
async def periodicHardwareTelemetry() -> None:
"""
Diffuse les metriques reelles materielles (RAM Systeme, Memoire Mirza, CPU %) vers le HUD en temps reel.
"""
while True:
await asyncio.sleep(2.0)
try:
snap = MemoryManager.getMemorySnapshot()
cpuPct = psutil.cpu_percent(interval=None)
gpuPct = MemoryManager.getGpuUtilization()
systemUsed = round(snap["systemUsedGb"], 1)
systemTotal = round(snap["systemTotalGb"], 1)
systemPct = round((snap["systemUsedGb"] / snap["systemTotalGb"]) * 100, 1)
mirzaUsed = round(snap["vramActiveGb"], 1)
mirzaTotal = round(snap["systemTotalGb"], 1)
mirzaPct = min(100.0, round((snap["vramActiveGb"] / snap["systemTotalGb"]) * 100, 1))
TelemetryManager.update_hardware_metrics(vram=mirzaUsed, ram=systemUsed, cpu=cpuPct, gpu=gpuPct)
# None quand aucun capteur n'est lisible : le HUD affiche "--".
thermal = getThermalSnapshot()
broadcaster.broadcastEvent("hardwareMetrics", {
"systemUsedGb": systemUsed,
"systemTotalGb": systemTotal,
"systemPercent": systemPct,
"mirzaUsedGb": mirzaUsed,
"mirzaTotalGb": mirzaTotal,
"mirzaPercent": mirzaPct,
"cpuPercent": round(cpuPct, 1),
"gpuPercent": gpuPct,
"cpuTempC": thermal["cpuC"],
"gpuTempC": thermal["gpuC"],
})
except asyncio.CancelledError:
break
except Exception as exc:
logger.debug("Erreur telemetrie materielle: %s", exc)
@asynccontextmanager
async def appLifespan(app: FastAPI) -> AsyncGenerator[None]:
"""
Demarre la boucle de telemetrie materielle a l'ouverture de l'application.
Parameters
----------
app : FastAPI
Application dont le cycle de vie est gere.
Yields
------
None
Relais du controle tant que l'application reste active.
"""
broadcaster.setEventLoop(asyncio.get_running_loop())
telemetryTask = asyncio.create_task(periodicHardwareTelemetry())
emitLog("SYSTEM", "Serveur FastAPI et telemetrie materielle initialises", "INFO")
try:
yield
finally:
telemetryTask.cancel()
try:
await telemetryTask
except asyncio.CancelledError:
pass
logger.info("Cycle de vie du serveur termine avec succes")
app = FastAPI(title="Mirza Widget Backend", version="1.0.0", lifespan=appLifespan)
# CORS limite a localhost, au LAN prive et au reseau Tailscale.
app.add_middleware(
CORSMiddleware,
allow_origin_regex=r"^https?://(localhost|127\.0\.0\.1|192\.168\.\d+\.\d+|10\.\d+\.\d+\.\d+|172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+|100\.\d+\.\d+\.\d+|.*\.ts\.net)(:\d+)?$",
allow_credentials=True,
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
allow_headers=["*"],
)
staticDir = os.path.join(os.path.dirname(__file__), "static")
if os.path.exists(staticDir):
@app.middleware("http")
async def _staticCachePolicy(request, call_next):
"""
Empeche le navigateur de servir un widget perime apres un build.
Les fichiers de ``assets`` portent leur contenu dans leur nom, donc
un nom nouveau est un contenu nouveau et se cache indefiniement.
``index.html`` n'est qu'un pointeur vers le bundle hache : sans en-tete
explicite, le navigateur le garde sur une heuristique de date de
modification et continue de charger un bundle qui n'existe plus.
Parameters
----------
request : Request
Requete HTTP entrante.
call_next : Callable
Rappel invoquant le middleware suivant.
Returns
-------
Response
Reponse decoratee de l'en-tete de cache adapte au chemin.
"""
response = await call_next(request)
path = request.url.path
if path.startswith("/assets/"):
response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
elif path in ("/", "/index.html"):
response.headers["Cache-Control"] = "no-store, must-revalidate"
return response
app.mount("/static", StaticFiles(directory=staticDir), name="static")
assetsDir = os.path.join(staticDir, "assets")
if os.path.exists(assetsDir):
app.mount("/assets", StaticFiles(directory=assetsDir), name="assets")
systemRouter = createSystemRouter(orchestrator, broadcaster, emitLog)
memoriesRouter = createMemoriesRouter(orchestrator, broadcaster, emitLog)
websocketRouter = createWebSocketRouter(orchestrator, broadcaster, emitLog)
app.include_router(systemRouter)
app.include_router(memoriesRouter)
app.include_router(websocketRouter)
@app.get("/", response_model=None)
async def getWidgetUi() -> FileResponse | dict[str, str]:
"""
Sert le widget compile, ou un etat minimal si le build est absent.
Returns
-------
FileResponse | dict[str, str]
Le fichier ``index.html`` du widget, sinon un dictionnaire d'etat.
"""
indexPath = os.path.join(staticDir, "index.html")
if os.path.exists(indexPath):
return FileResponse(indexPath)
return {"status": "Mirza backend online"}
return app
[docs]
class WidgetServer:
"""
Serveur Web unifie distribuant le widget interactif React et l'API WebSocket.
Gere les liaisons full-duplex pour le streaming audio mobile, la telemetrie,
et la gestion des certificats SSL/TLS sur reseau local et Tailscale.
Parameters
----------
orchestrator : ConversationOrchestrator
Reference vers l'orchestrateur conversationnel de Mirza.
host : str
Adresse d'ecoute reseau (defaut: "0.0.0.0").
port : int
Port d'ecoute TCP (defaut: 8765).
enableSsl : bool
Active HTTPS et WSS avec certificats auto-signes (defaut: False).
Attributes
----------
orchestrator : ConversationOrchestrator
Orchestrateur de conversation utilise par l'application.
broadcaster : EventBroadcaster
Diffuseur d'evenements temps reel de l'application.
app : FastAPI
Application ASGI exposee par le serveur.
"""
[docs]
def __init__(
self,
orchestrator: ConversationOrchestrator,
host: str = "0.0.0.0",
port: int = 8765,
enableSsl: bool = False,
) -> None:
self.orchestrator = orchestrator
self.host = host
self.port = port
self.enableSsl = enableSsl
self.broadcaster = EventBroadcaster()
self.app = createWidgetApp(self.orchestrator, self.broadcaster)
[docs]
def run(self) -> None:
"""
Demarre le serveur Uvicorn avec support HTTPS / WSS en mode bloquant.
"""
localIp = getLocalIp()
tailscaleIp, tailscaleDns = getTailscaleInfo()
protocol = "https" if self.enableSsl else "http"
wsProtocol = "wss" if self.enableSsl else "ws"
sslCert = None
sslKey = None
if self.enableSsl:
additional = []
if tailscaleIp:
additional.append(tailscaleIp)
if tailscaleDns:
additional.append(tailscaleDns)
sslCert, sslKey = SslManager.ensureCertificates(
localIp=localIp,
additionalIps=additional if additional else None,
)
print("=" * 68)
print(" SERVEUR MIRZA EN LIGNE (LAN & SMARTPHONE AUDIO READY)")
print("=" * 68)
print(f" Acces Mac Local : {protocol}://127.0.0.1:{self.port}")
print(f" Acces Smartphone LAN : {protocol}://{localIp}:{self.port}")
if tailscaleIp:
print(f" Acces Tailscale (IP) : {protocol}://{tailscaleIp}:{self.port}")
if tailscaleDns:
print(f" Acces Tailscale (DNS) : {protocol}://{tailscaleDns}:{self.port}")
print(f" Canal WebSocket WSS : {wsProtocol}://{tailscaleIp or localIp}:{self.port}/ws")
print("=" * 68)
if self.enableSsl and sslCert and sslKey:
uvicorn.run(
self.app,
host=self.host,
port=self.port,
ssl_certfile=sslCert,
ssl_keyfile=sslKey,
log_level="warning",
)
else:
uvicorn.run(
self.app,
host=self.host,
port=self.port,
log_level="warning",
)