diff --git a/openpilot/common/params_keys.h b/openpilot/common/params_keys.h index 3c39fc5f53..96ee858609 100644 --- a/openpilot/common/params_keys.h +++ b/openpilot/common/params_keys.h @@ -134,6 +134,7 @@ inline static std::unordered_map keys = { {"UsbGpuPresent", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}}, {"UsbGpuCompiled", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}}, {"Version", {PERSISTENT, STRING}}, + {"WgpuEnabled", {CLEAR_ON_MANAGER_START | DEVELOPMENT_ONLY, BOOL}}, // --- sunnypilot params --- // {"ApiCache_DriveStats", {PERSISTENT, JSON}}, diff --git a/openpilot/selfdrive/modeld/SConscript b/openpilot/selfdrive/modeld/SConscript index 5f6281556f..18a5d1091c 100644 --- a/openpilot/selfdrive/modeld/SConscript +++ b/openpilot/selfdrive/modeld/SConscript @@ -37,6 +37,9 @@ available = probe_devices() if 'CUDA' in available: tg_backend = 'CUDA' tg_flags = f'DEV={tg_backend}' +elif 'METAL' in available: + tg_backend = 'METAL' + tg_flags = f'DEV={tg_backend} FLOAT16=1' elif 'QCOM' in available: tg_backend = 'QCOM' tg_flags = f'DEV={tg_backend} IMAGE=1 FLOAT16=1 NOLOCALS=1 JIT_BATCH_SIZE=0 OPENPILOT_HACKS=1' @@ -55,6 +58,7 @@ tg_devices = { # which device to put jit inputs to at runtime } USBGPU = usbgpu_present() # or release # TODO always build big model on release +WGPU = os.getenv('WGPU') == '1' if USBGPU: usbgpu_tg_flags = f'DEBUG=2 DEV=USB+AMD:LLVM WARP_DEV={tg_backend} FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0' # the USB+AMD GPU takes an exclusive flock; serialize all targets that touch it @@ -84,13 +88,13 @@ compile_modeld_script = [ model_w, model_h = MEDMODEL_INPUT_SIZE frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ -for usbgpu in [False, True] if USBGPU else [False]: +for usbgpu in [False, True] if USBGPU or WGPU else [False]: target_pkl_path = File(modeld_pkl_path(usbgpu)).abspath # BIG_INTO_SMALL=1 builds the default target from the big model, e.g. to test it without a USB GPU - file_prefix, cmd_flags = ('big_', usbgpu_tg_flags) if usbgpu else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags) + file_prefix, cmd_flags = ('big_', usbgpu_tg_flags if USBGPU else tg_flags) if usbgpu else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags) driving_onnx_deps = get_existing_chunks(File(f"models/{file_prefix}driving_supercombo.onnx").abspath) camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in CAMERA_CONFIGS) - cmd = (f'{cmd_flags} {mac_brew_string} python3 {modeld_dir}/compile_modeld.py ' + cmd = (f'{cmd_flags} {mac_brew_string} {sys.executable} {modeld_dir}/compile_modeld.py ' f'--model-size {model_w}x{model_h} ' f'--camera-resolutions {camera_res_args} ' f'--onnx {File(f"models/{file_prefix}driving_supercombo.onnx").abspath} ' @@ -104,7 +108,7 @@ for usbgpu in [False, True] if USBGPU else [False]: tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(chunk_targets), chunker_file], [cmd, Action(do_chunk, " [CHUNK] $TARGET")], ) - if usbgpu: + if usbgpu and USBGPU: lenv.SideEffect(usbgpu_lock, node) # get model metadata diff --git a/openpilot/selfdrive/modeld/modeld.py b/openpilot/selfdrive/modeld/modeld.py index 1ca1a71f35..44b34ce393 100755 --- a/openpilot/selfdrive/modeld/modeld.py +++ b/openpilot/selfdrive/modeld/modeld.py @@ -26,6 +26,7 @@ from openpilot.common.file_chunker import open_file_chunked, get_manifest_path from openpilot.selfdrive.modeld.constants import ModelConstants, Plan from openpilot.selfdrive.modeld.helpers import usbgpu_present, modeld_pkl_path, get_tg_input_devices, load_oob from openpilot.selfdrive.modeld.usbgpu_link import wait_usbgpu_link +from openpilot.tools.wgpu.zmq import ZmqPubMaster, ZmqSubMaster from openpilot.sunnypilot.livedelay.helpers import get_lat_delay from openpilot.sunnypilot.modeld_v2.modeld_base import ModelStateBase @@ -79,12 +80,12 @@ class FrameMeta: class ModelState(ModelStateBase): prev_desire: np.ndarray # for tracking the rising edge of the pulse - def __init__(self, cam_w: int, cam_h: int, usbgpu: bool): + def __init__(self, cam_w: int, cam_h: int, usbgpu: bool, big_model: bool = False): ModelStateBase.__init__(self) self.LAT_SMOOTH_SECONDS = LAT_SMOOTH_SECONDS input_devices = get_tg_input_devices(PROCESS_NAME, usbgpu) self.WARP_DEV, self.QUEUE_DEV = input_devices['WARP_DEV'], input_devices['QUEUE_DEV'] - jits = load_oob(open_file_chunked(modeld_pkl_path(usbgpu))) + jits = load_oob(open_file_chunked(modeld_pkl_path(usbgpu or big_model))) metadata = jits['metadata'] self.input_shapes = metadata['input_shapes'] self.vision_input_names = [k for k in self.input_shapes if 'img' in k] @@ -139,12 +140,14 @@ class ModelState(ModelStateBase): return outputs_dict -def main(demo=False): +def main(demo=False, remote_addr: str | None = None, big_model: bool = False): cloudlog.warning("modeld init") _present = usbgpu_present() _compiled = os.path.isfile(get_manifest_path(modeld_pkl_path(usbgpu=True))) USBGPU = _present and _compiled + if big_model and not _compiled: + raise FileNotFoundError(f"big model is not compiled: {modeld_pkl_path(usbgpu=True)}") params = Params() params.put_bool("UsbGpuPresent", _present) params.put_bool("UsbGpuCompiled", _compiled) @@ -178,12 +181,16 @@ def main(demo=False): wait_usbgpu_link() st = time.monotonic() cloudlog.warning("loading model") - model = ModelState(vipc_client_main.width, vipc_client_main.height, USBGPU) + model = ModelState(vipc_client_main.width, vipc_client_main.height, USBGPU, big_model=big_model) cloudlog.warning(f"models loaded in {time.monotonic() - st:.1f}s, modeld starting") # messaging - pm = PubMaster(["modelV2", "drivingModelData", "cameraOdometry", "modelDataV2SP"]) - sm = SubMaster(["deviceState", "carState", "roadCameraState", "liveCalibration", "driverMonitoringState", "carControl", "liveDelay"]) + output_services = ["modelV2", "drivingModelData", "cameraOdometry", "modelDataV2SP"] + pm = ZmqPubMaster(output_services) if remote_addr is not None else PubMaster(output_services) + services = ["deviceState", "carState", "roadCameraState", "liveCalibration", "driverMonitoringState", "carControl", "liveDelay"] + if remote_addr is not None: + services.append("carParams") + sm = ZmqSubMaster(services, remote_addr) if remote_addr is not None else SubMaster(services) publish_state = PublishState() params = Params() @@ -203,6 +210,11 @@ def main(demo=False): if demo: CP = get_demo_car_params() + elif remote_addr is not None: + cloudlog.warning("waiting for remote carParams") + while not sm.seen["carParams"]: + sm.update(100) + CP = sm["carParams"] else: CP = messaging.log_from_bytes(params.get("CarParams", block=True), car.CarParams) cloudlog.info("modeld got CarParams: %s", CP.brand) @@ -332,7 +344,9 @@ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument('--demo', action='store_true', help='A boolean for demo mode.') + parser.add_argument('--remote', metavar='ADDRESS', help='Run against a remote device over the cereal ZMQ bridge.') + parser.add_argument('--big-model', action='store_true', help='Use the locally compiled big driving model.') args = parser.parse_args() - main(demo=args.demo) + main(demo=args.demo, remote_addr=args.remote, big_model=args.big_model) except KeyboardInterrupt: cloudlog.warning("got SIGINT") diff --git a/openpilot/system/manager/process_config.py b/openpilot/system/manager/process_config.py index a72675385d..9210b99bb8 100644 --- a/openpilot/system/manager/process_config.py +++ b/openpilot/system/manager/process_config.py @@ -95,6 +95,9 @@ def is_stock_model(started, params, CP: car.CarParams) -> bool: """Check if the active model runner is stock.""" return bool(get_active_model_runner(params, not started) == custom.ModelManagerSP.Runner.stock) +def not_wgpu(started: bool, params: Params, CP: car.CarParams) -> bool: + return not params.get_bool("WgpuEnabled") + def mapd_ready(started: bool, params: Params, CP: car.CarParams) -> bool: return bool(os.path.exists(Paths.mapd_root())) @@ -128,7 +131,7 @@ procs = [ PythonProcess("micd", "openpilot.system.micd", iscar), PythonProcess("timed", "openpilot.system.timed", always_run, enabled=not PC), - PythonProcess("modeld", "openpilot.selfdrive.modeld.modeld", and_(only_onroad, is_stock_model)), + PythonProcess("modeld", "openpilot.selfdrive.modeld.modeld", and_(and_(only_onroad, is_stock_model), not_wgpu)), PythonProcess("dmonitoringmodeld", "openpilot.selfdrive.modeld.dmonitoringmodeld", driverview, enabled=(WEBCAM or not PC)), PythonProcess("sensord", "openpilot.system.sensord.sensord", only_onroad, enabled=not PC), diff --git a/openpilot/tools/camerastream/compressed_vipc.py b/openpilot/tools/camerastream/compressed_vipc.py index 56ed12889f..19f8849f20 100755 --- a/openpilot/tools/camerastream/compressed_vipc.py +++ b/openpilot/tools/camerastream/compressed_vipc.py @@ -1,5 +1,4 @@ #!/usr/bin/env python3 -import os import argparse import multiprocessing import time @@ -10,6 +9,7 @@ from collections import deque import openpilot.cereal.messaging as messaging from msgq.visionipc import VisionIpcServer, VisionStreamType from openpilot.tools.camerastream.ffmpeg_decoder import Decoder, FFmpegError +from openpilot.tools.wgpu.zmq import ZmqSubMaster, ZmqSubSocket V4L2_BUF_FLAG_KEYFRAME = 8 @@ -30,10 +30,7 @@ def decoder(addr, vipc_server, vst, W, H, debug=False): codec = Decoder("hevc") - os.environ["ZMQ"] = "1" - messaging.reset_context() - sock = messaging.sub_sock(sock_name, None, addr=addr, conflate=False) - cnt = 0 + sock = ZmqSubSocket(sock_name, addr) last_idx = -1 seen_iframe = False @@ -46,8 +43,9 @@ def decoder(addr, vipc_server, vst, W, H, debug=False): time_q.clear() while 1: - msgs = messaging.drain_sock(sock, wait_for_one=True) - for evt in msgs: + msgs = sock.drain(wait_for_one=True) + for raw in msgs: + evt = messaging.log_from_bytes(raw) evta = getattr(evt, evt.which()) if last_idx != -1 and evta.idx.encodeId != (last_idx + 1): if debug: @@ -94,8 +92,9 @@ def decoder(addr, vipc_server, vst, W, H, debug=False): continue frame_start_time = time_q.popleft() - vipc_server.send(vst, img_yuv.data, cnt, int(frame_start_time*1e9), int(time.monotonic()*1e9)) - cnt += 1 + # Preserve the device camera metadata so remote model outputs line up with + # the rest of the device's cereal timeline. + vipc_server.send(vst, img_yuv.data, evta.idx.frameId, evta.idx.timestampSof, evta.idx.timestampEof) pc_latency = (time.monotonic()-frame_start_time)*1000 if debug: @@ -105,14 +104,10 @@ def decoder(addr, vipc_server, vst, W, H, debug=False): class CompressedVipc: def __init__(self, addr, vision_streams, server_name, debug=False): - print("getting frame sizes") - os.environ["ZMQ"] = "1" - messaging.reset_context() - sm = messaging.SubMaster([ENCODE_SOCKETS[s] for s in vision_streams], addr=addr) + print("waiting for remote camera stream metadata", flush=True) + sm = ZmqSubMaster([ENCODE_SOCKETS[s] for s in vision_streams], addr) while min(sm.recv_frame.values()) == 0: sm.update(100) - os.environ.pop("ZMQ") - messaging.reset_context() self.vipc_server = VisionIpcServer(server_name) for vst in vision_streams: @@ -121,9 +116,10 @@ class CompressedVipc: self.vipc_server.start_listener() self.procs = [] + process_context = multiprocessing.get_context("fork") for vst in vision_streams: ed = sm[ENCODE_SOCKETS[vst]] - p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug)) + p = process_context.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug)) p.start() self.procs.append(p) diff --git a/openpilot/tools/camerastream/ffmpeg_decoder.py b/openpilot/tools/camerastream/ffmpeg_decoder.py index 13cdc12df1..226a91c664 100644 --- a/openpilot/tools/camerastream/ffmpeg_decoder.py +++ b/openpilot/tools/camerastream/ffmpeg_decoder.py @@ -46,10 +46,23 @@ def _bind(fn, restype, *argtypes): return fn +def _library_path(name: str, major: int) -> str: + candidates = ( + f"lib{name}.so.{major}", + f"lib{name}.{major}.dylib", + f"lib{name}.dylib", + ) + for candidate in candidates: + path = os.path.join(ffmpeg.LIB_DIR, candidate) + if os.path.isfile(path): + return path + raise FileNotFoundError(f"FFmpeg library not found in {ffmpeg.LIB_DIR}: {', '.join(candidates)}") + + def _load_libraries(): - avutil = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavutil.so.59"), mode=ctypes.RTLD_GLOBAL) - avcodec = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavcodec.so.61"), mode=ctypes.RTLD_GLOBAL) - swscale = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libswscale.so.8"), mode=ctypes.RTLD_GLOBAL) + avutil = ctypes.CDLL(_library_path("avutil", 59), mode=ctypes.RTLD_GLOBAL) + avcodec = ctypes.CDLL(_library_path("avcodec", 61), mode=ctypes.RTLD_GLOBAL) + swscale = ctypes.CDLL(_library_path("swscale", 8), mode=ctypes.RTLD_GLOBAL) c_int, c_char_p, c_void_p, c_size_t = ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_size_t c_uint8_p = ctypes.POINTER(ctypes.c_uint8) diff --git a/openpilot/tools/wgpu/README.md b/openpilot/tools/wgpu/README.md new file mode 100644 index 0000000000..08b5dc3605 --- /dev/null +++ b/openpilot/tools/wgpu/README.md @@ -0,0 +1,59 @@ +# Wireless modeld proof of concept + +This runs driving `modeld` on a laptop and returns its cereal outputs to a comma +device over the existing Wi-Fi network. It reuses the existing HEVC camera +stream, VisionIPC decoder, and cereal ZMQ bridge. + +This is for offroad/bench testing only. Wi-Fi has no deterministic latency or +availability guarantee. The device restores local `modeld` when the device-side +helper exits, but that is not a seamless onroad failover. + +## Build + +Use the same commit on the laptop and comma device. Build the cereal bridge on +the device: + +```sh +scons -u openpilot/cereal/messaging/bridge +``` + +Build and test the normal model on the laptop first: + +```sh +PATH="$PWD/.venv/bin:$PATH" scons -u +``` + +To compile the big external-GPU model for the laptop's local tinygrad backend: + +```sh +PATH="$PWD/.venv/bin:$PATH" WGPU=1 scons -u \ + openpilot/selfdrive/modeld/models/big_driving_tinygrad.pkl.chunkmanifest +``` + +On macOS, the build selects tinygrad's Metal backend when it is available. +On an 8-GPU-core M5 MacBook Air, the small model's compiled policy pass measured +about 6–11 ms, while the big model measured about 79–81 ms. The latter already +misses the 50 ms model cadence before network and codec latency, so start with +the small model on that class of laptop. + +## Run + +Find the laptop's LAN IP address that the comma device can reach. While the +device is offroad, run: + +```sh +cd /data/openpilot +python3 -m openpilot.tools.wgpu.device LAPTOP_IP +``` + +Keep that terminal open. On the laptop, run: + +```sh +cd /path/to/openpilot +python3 -m openpilot.tools.wgpu.host COMMA_IP +``` + +Add `--big-model` after `COMMA_IP` to use the locally compiled big model. + +The first remote `carParams` packet can take up to 50 seconds. Stop either side +with Ctrl+C. Stop the device helper before changing branches or rebooting. diff --git a/openpilot/tools/wgpu/device.py b/openpilot/tools/wgpu/device.py new file mode 100644 index 0000000000..7be206f2e0 --- /dev/null +++ b/openpilot/tools/wgpu/device.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +import argparse +import signal +import subprocess +import time +from pathlib import Path + +from openpilot.common.params import Params + + +MODEL_OUTPUTS = "modelV2,drivingModelData,cameraOdometry,modelDataV2SP" +ROOT = Path(__file__).resolve().parents[3] +BRIDGE = ROOT / "openpilot/cereal/messaging/bridge" + + +def stop_process(proc: subprocess.Popen) -> None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + +def handle_sigterm(*_) -> None: + raise KeyboardInterrupt + + +def main() -> None: + parser = argparse.ArgumentParser(description="Route modeld traffic between this device and a wireless host.") + parser.add_argument("host", help="Laptop IP address reachable from this device") + args = parser.parse_args() + + if not BRIDGE.is_file(): + raise FileNotFoundError(f"build the cereal bridge first: {BRIDGE}") + + params = Params() + if not params.get_bool("IsOffroad"): + raise RuntimeError("start the wgpu bridge while offroad") + + procs: list[subprocess.Popen] = [] + try: + params.put_bool("WgpuEnabled", True, block=True) + procs = [ + subprocess.Popen([str(BRIDGE)]), + subprocess.Popen([str(BRIDGE), args.host, MODEL_OUTPUTS]), + ] + print(f"wgpu enabled; forwarding camera/state to {args.host}") + print("keep this process running; Ctrl+C restores local modeld") + while all(proc.poll() is None for proc in procs): + time.sleep(0.25) + failed = next(proc for proc in procs if proc.poll() is not None) + raise RuntimeError(f"bridge exited with status {failed.returncode}") + finally: + params.put_bool("WgpuEnabled", False, block=True) + for proc in procs: + if proc.poll() is None: + stop_process(proc) + print("wgpu disabled; local modeld restored") + + +if __name__ == "__main__": + signal.signal(signal.SIGTERM, handle_sigterm) + try: + main() + except KeyboardInterrupt: + pass diff --git a/openpilot/tools/wgpu/host.py b/openpilot/tools/wgpu/host.py new file mode 100644 index 0000000000..dbbbe63399 --- /dev/null +++ b/openpilot/tools/wgpu/host.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +import argparse +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +CAMERASTREAM = ROOT / "openpilot/tools/camerastream/compressed_vipc.py" + + +def stop_process(proc: subprocess.Popen) -> None: + try: + os.killpg(proc.pid, signal.SIGTERM) + except ProcessLookupError: + proc.wait() + return + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(proc.pid, signal.SIGKILL) + proc.wait() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run modeld on this host for a remote comma device.") + parser.add_argument("device", help="comma device hostname or IP address") + parser.add_argument("--big-model", action="store_true", help="use the locally compiled big driving model") + args = parser.parse_args() + + camera = subprocess.Popen([sys.executable, str(CAMERASTREAM), args.device, "--cams", "0,2"], start_new_session=True) + model_args = [sys.executable, "-m", "openpilot.selfdrive.modeld.modeld", "--remote", args.device] + if args.big_model: + model_args.append("--big-model") + model = subprocess.Popen(model_args, cwd=ROOT, start_new_session=True) + + procs = [camera, model] + try: + while all(proc.poll() is None for proc in procs): + time.sleep(0.25) + failed = next(proc for proc in procs if proc.poll() is not None) + raise RuntimeError(f"wgpu host process exited with status {failed.returncode}") + finally: + for proc in procs: + if proc.poll() is None: + stop_process(proc) + print("wgpu host stopped") + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + pass diff --git a/openpilot/tools/wgpu/zmq.py b/openpilot/tools/wgpu/zmq.py new file mode 100644 index 0000000000..0300dcb7cb --- /dev/null +++ b/openpilot/tools/wgpu/zmq.py @@ -0,0 +1,90 @@ +import time + +import zmq + +import openpilot.cereal.messaging as messaging + + +def service_port(endpoint: str) -> int: + # Keep this in sync with cereal/messaging/bridge_zmq.cc. + value = 0xcbf29ce484222325 + for char in endpoint.encode(): + value ^= char + value = (value * 0x100000001b3) & 0xffffffffffffffff + return 8023 + (value % (65535 - 8023)) + + +class ZmqSubSocket: + def __init__(self, endpoint: str, address: str, conflate: bool = False): + self.context = zmq.Context() + self.socket = self.context.socket(zmq.SUB) + self.socket.setsockopt(zmq.SUBSCRIBE, b"") + self.socket.setsockopt(zmq.RECONNECT_IVL_MAX, 500) + if conflate: + self.socket.setsockopt(zmq.CONFLATE, 1) + self.socket.connect(f"tcp://{address}:{service_port(endpoint)}") + + def receive(self, non_blocking: bool = False) -> bytes | None: + try: + return self.socket.recv(flags=zmq.NOBLOCK if non_blocking else 0) + except zmq.Again: + return None + + def drain(self, wait_for_one: bool = False) -> list[bytes]: + messages = [] + if wait_for_one: + message = self.receive() + if message is not None: + messages.append(message) + while (message := self.receive(non_blocking=True)) is not None: + messages.append(message) + return messages + + +class ZmqSubMaster: + def __init__(self, services: list[str], address: str): + self.services = services + self.sockets = {service: ZmqSubSocket(service, address, conflate=True) for service in services} + self.poller = zmq.Poller() + self.socket_to_service = {} + for service, sub in self.sockets.items(): + self.poller.register(sub.socket, zmq.POLLIN) + self.socket_to_service[sub.socket] = service + + self.data = {service: getattr(messaging.new_message(service).as_reader(), service) for service in services} + self.seen = dict.fromkeys(services, False) + self.updated = dict.fromkeys(services, False) + self.recv_frame = dict.fromkeys(services, 0) + self.frame = -1 + + def __getitem__(self, service: str): + return self.data[service] + + def update(self, timeout: int = 100) -> None: + self.frame += 1 + self.updated = dict.fromkeys(self.services, False) + for socket, _ in self.poller.poll(timeout): + service = self.socket_to_service[socket] + raw = self.sockets[service].receive(non_blocking=True) + if raw is None: + continue + event = messaging.log_from_bytes(raw) + self.data[service] = getattr(event, service) + self.seen[service] = True + self.updated[service] = True + self.recv_frame[service] = self.frame + + +class ZmqPubMaster: + def __init__(self, services: list[str]): + context = zmq.Context.instance() + self.sockets = {} + for service in services: + socket = context.socket(zmq.PUB) + socket.bind(f"tcp://*:{service_port(service)}") + self.sockets[service] = socket + # Give already-running bridge subscribers time to finish their handshake. + time.sleep(0.1) + + def send(self, service: str, message) -> None: + self.sockets[service].send(message.to_bytes(), flags=zmq.NOBLOCK)