File size: 2,270 Bytes
977b06e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | # Copyright 2025 starVLA community. All rights reserved.
# Licensed under the MIT License, Version 1.0 (the "License");
# Implemented by [Jinhui YE / HKUST University] in [2025].
import logging
import socket
import argparse
from deployment.model_server.tools.websocket_policy_server import WebsocketPolicyServer
from deployment.model_server.policy_wrapper import PolicyServerWrapper
import os
def main(args) -> None:
# Example usage:
# policy = YourPolicyClass() # Replace with your actual policy class
# server = WebsocketPolicyServer(policy, host="localhost", port=10091)
# server.serve_forever()
policy = PolicyServerWrapper(
ckpt_path=args.ckpt_path,
device="cuda",
use_bf16=args.use_bf16,
)
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
logging.info("Creating server (host: %s, ip: %s)", hostname, local_ip)
# start websocket server
server = WebsocketPolicyServer(
policy=policy,
host="0.0.0.0",
port=args.port,
idle_timeout=args.idle_timeout,
metadata={**policy.metadata, "policy_seed": args.seed},
)
logging.info("server running ...")
server.serve_forever()
def build_argparser():
parser = argparse.ArgumentParser()
parser.add_argument("--ckpt_path", type=str, default="Qwen/Qwen2.5-VL-3B-Instruct")
parser.add_argument("--port", type=int, default=10093)
parser.add_argument("--use_bf16", action="store_true")
parser.add_argument("--seed", type=int, default=0)
parser.add_argument("--idle_timeout" , type=int, default=1800, help="Idle timeout in seconds, -1 means never close")
return parser
def start_debugpy_once():
"""start debugpy once"""
import debugpy
if getattr(start_debugpy_once, "_started", False):
return
debugpy.listen(("0.0.0.0", 10095))
print("🔍 Waiting for VSCode attach on 0.0.0.0:10095 ...")
debugpy.wait_for_client()
start_debugpy_once._started = True
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, force=True)
parser = build_argparser()
args = parser.parse_args()
if os.getenv("DEBUG", False):
print("🔍 DEBUGPY is enabled")
start_debugpy_once()
main(args)
|