kyleandClaude Opus 4.8
54a01625ce
Standalone Open3D viewer that listens on a TCP/Unix socket and renders NDJSON point-cloud streams from any producer. Decoupled server/protocol/store layers (no Open3D dependency, testable headless) plus a lazy Open3D render loop. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
"""Stream-socket server that feeds a CloudStore from NDJSON producers.
|
|
|
|
Listens on one or more addresses (TCP and/or Unix), accepts any number of
|
|
producers concurrently, and applies their messages to a shared store. All
|
|
networking is on background threads; the render loop reads the store separately.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import socket
|
|
import threading
|
|
|
|
from .protocol import LineDecoder, parse_address
|
|
from .store import CloudStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CloudServer:
|
|
def __init__(self, store: CloudStore, addresses: list[str]) -> None:
|
|
self._store = store
|
|
self._addresses = addresses
|
|
self._listeners: list[socket.socket] = []
|
|
self._unix_paths: list[str] = []
|
|
self._stop = threading.Event()
|
|
# Actual bound endpoints (resolves ephemeral ":0" ports for callers/tests).
|
|
self.bound: list[tuple[str, str, int | None]] = []
|
|
|
|
def start(self) -> None:
|
|
for addr in self._addresses:
|
|
kind, a, b = parse_address(addr)
|
|
if kind == "tcp":
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
sock.bind((a, b))
|
|
host, port = sock.getsockname()[:2]
|
|
self.bound.append(("tcp", host, port))
|
|
else:
|
|
# Replace a stale socket file from a prior run.
|
|
if os.path.exists(a):
|
|
os.unlink(a)
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.bind(a)
|
|
self._unix_paths.append(a)
|
|
self.bound.append(("unix", a, None))
|
|
sock.listen(8)
|
|
sock.settimeout(0.5)
|
|
self._listeners.append(sock)
|
|
threading.Thread(target=self._accept_loop, args=(sock, addr),
|
|
name=f"accept[{addr}]", daemon=True).start()
|
|
logger.info("listening on %s", addr)
|
|
|
|
def _accept_loop(self, listener: socket.socket, addr: str) -> None:
|
|
while not self._stop.is_set():
|
|
try:
|
|
conn, _ = listener.accept()
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
break
|
|
threading.Thread(target=self._serve_client, args=(conn, addr),
|
|
name=f"client[{addr}]", daemon=True).start()
|
|
|
|
def _serve_client(self, conn: socket.socket, addr: str) -> None:
|
|
logger.info("producer connected on %s", addr)
|
|
conn.settimeout(1.0)
|
|
decoder = LineDecoder()
|
|
try:
|
|
while not self._stop.is_set():
|
|
try:
|
|
data = conn.recv(65536)
|
|
except socket.timeout:
|
|
continue
|
|
if not data:
|
|
break
|
|
for msg in decoder.feed(data):
|
|
self._apply(msg)
|
|
except Exception:
|
|
logger.exception("error serving producer on %s", addr)
|
|
finally:
|
|
conn.close()
|
|
logger.info("producer disconnected (%s)", addr)
|
|
|
|
def _apply(self, msg: dict) -> None:
|
|
kind = msg.get("type")
|
|
source = msg.get("source", "default")
|
|
if kind == "points":
|
|
self._store.add_points(source, msg.get("points", []),
|
|
msg.get("color"))
|
|
elif kind == "hello":
|
|
self._store.hello(source, msg.get("name"))
|
|
elif kind == "clear":
|
|
self._store.clear(source)
|
|
elif kind == "pose":
|
|
self._store.set_pose(source, msg.get("position", [0, 0, 0]),
|
|
msg.get("yaw"))
|
|
else:
|
|
logger.debug("ignoring unknown message type %r", kind)
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
for sock in self._listeners:
|
|
try:
|
|
sock.close()
|
|
except OSError:
|
|
pass
|
|
for path in self._unix_paths:
|
|
try:
|
|
os.unlink(path)
|
|
except OSError:
|
|
pass
|