6e83bbd73f
Serveur FastAPI reçoit le flux JSONL (sim ou ROV réel) sur /ws/ingest, SLAM incrémental, rediffuse carte+poses sur /ws/live, GUI live et export PLY. Déployé Docker sur caddy-net, exposé /moulin-live/. Client PC stream_client.py. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
"""
|
||
stream_client.py — Envoie un flux JSONL vers le serveur moulin-mapper.
|
||
|
||
Usage:
|
||
python stream_client.py --file run_L.jsonl --url ws://127.0.0.1:8211/ws/ingest --token moulin-2026 --speed 1.0
|
||
|
||
--speed 0 : aussi vite que possible (pas de délai)
|
||
--speed 1 : temps réel (deltas de t respectés)
|
||
--stdin : lit depuis stdin au lieu d'un fichier (bridge ROV)
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import time
|
||
import asyncio
|
||
|
||
try:
|
||
import websockets
|
||
except ImportError:
|
||
print("Installe d'abord : pip install websockets")
|
||
sys.exit(1)
|
||
|
||
|
||
async def stream(url: str, token: str, lines, speed: float):
|
||
# Ajoute le token comme query param
|
||
sep = "&" if "?" in url else "?"
|
||
full_url = f"{url}{sep}token={token}"
|
||
|
||
t_prev = None
|
||
sent = 0
|
||
t_file_prev = None
|
||
|
||
print(f"[moulin-stream] connexion → {full_url}")
|
||
|
||
while True:
|
||
try:
|
||
async with websockets.connect(full_url, ping_interval=20, ping_timeout=30) as ws:
|
||
print("[moulin-stream] connecté")
|
||
for line in lines:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
t_rec = rec.get("t", 0.0)
|
||
|
||
# Timing
|
||
if speed > 0 and t_file_prev is not None:
|
||
dt_file = t_rec - t_file_prev
|
||
if dt_file > 0:
|
||
await asyncio.sleep(dt_file / speed)
|
||
|
||
t_file_prev = t_rec
|
||
|
||
await ws.send(json.dumps(rec))
|
||
sent += 1
|
||
|
||
if sent % 100 == 0:
|
||
print(f"\r[moulin-stream] {sent} lignes envoyées, t={t_rec:.1f}s", end="", flush=True)
|
||
|
||
print(f"\n[moulin-stream] terminé — {sent} enregistrements envoyés")
|
||
return
|
||
|
||
except (websockets.exceptions.ConnectionClosed,
|
||
ConnectionRefusedError,
|
||
OSError) as e:
|
||
print(f"\n[moulin-stream] connexion perdue ({e}), reconnexion dans 3s…")
|
||
await asyncio.sleep(3.0)
|
||
# Redémarre depuis le début si on a perdu la connexion
|
||
# (les données sont déjà lues depuis un itérateur — on ne peut pas rembobiner stdin)
|
||
print("[moulin-stream] AVERTISSEMENT: relance depuis le début du fichier")
|
||
break
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Stream JSONL vers moulin-mapper")
|
||
src = parser.add_mutually_exclusive_group(required=True)
|
||
src.add_argument("--file", help="Fichier JSONL à envoyer")
|
||
src.add_argument("--stdin", action="store_true", help="Lit depuis stdin")
|
||
parser.add_argument("--url", default="ws://127.0.0.1:8211/ws/ingest",
|
||
help="URL WebSocket du serveur (ex: wss://lab.freeboxos.fr/moulin-live/ws/ingest)")
|
||
parser.add_argument("--token", default="moulin-2026", help="Token d'authentification")
|
||
parser.add_argument("--speed", type=float, default=1.0,
|
||
help="Multiplicateur de vitesse (0=max, 1=temps réel, 2=2×)")
|
||
args = parser.parse_args()
|
||
|
||
if args.file:
|
||
with open(args.file, "r") as f:
|
||
lines = f.readlines()
|
||
print(f"[moulin-stream] fichier: {args.file} ({len(lines)} lignes)")
|
||
else:
|
||
lines = sys.stdin
|
||
|
||
asyncio.run(stream(args.url, args.token, lines, args.speed))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|