import atexit
import base64
import json
import math
import os
import queue
import shlex
import threading
import time
import uuid
try:
import paramiko
except ImportError as exc:
raise ImportError("缺少 paramiko，请先执行：pip install paramiko") from exc
_PROTOCOL_PREFIX = "@@XMWROBOT@@"
_REMOTE_BRIDGE_SOURCE = r'''
import glob
import json
import os
import select
import sys
import time
PREFIX = "@@XMWROBOT@@"
PERIOD = 0.01
def emit(data):
print(PREFIX + json.dumps(data, ensure_ascii=False, separators=(",", ":")), flush=True)
def find_sdk_root():
home = os.path.expanduser("~")
candidates = []
configured = os.environ.get("XMWROBOT_SDK_ROOT", "").strip()
if configured:
candidates.append(configured)
candidates.extend([
os.path.join(home, "work"),
os.path.join(home, "noetix_sdk_bumi-main"),
os.path.join(home, "noetix_sdk_bumi"),
os.path.join(home, "work", "noetix_sdk_bumi-main"),
os.path.join(home, "work", "noetix_sdk_bumi"),
])
candidates.extend(glob.glob(os.path.join(home, "**", "noetix_sdk_bumi*"), recursive=True))
seen = set()
for root in candidates:
root = os.path.abspath(os.path.expanduser(root))
if root in seen:
continue
seen.add(root)
if os.path.isfile(os.path.join(root, "config", "dds.xml")):
return root
raise RuntimeError("没有找到新版 Bumi SDK")
def load_sdk():
root = find_sdk_root()
os.chdir(root)
os.environ["CYCLONEDDS_URI"] = "file://" + os.path.join(root, "config", "dds.xml")
search_dirs = [
root,
os.path.join(root, "build"),
os.path.join(root, "build", "lib"),
os.path.join(root, "examples_py"),
]
for so_path in glob.glob(os.path.join(root, "**", "highcontrol_py*.so"), recursive=True):
search_dirs.append(os.path.dirname(so_path))
for item in search_dirs:
if os.path.isdir(item) and item not in sys.path:
sys.path.insert(0, item)
from highcontrol_py import ControlCmd, HighController
controller = HighController.instance()
controller.init()
return controller, ControlCmd
def stop(controller, ControlCmd, count=8):
for _ in range(max(1, int(count))):
controller.publish_cmd(0.0, 0.0, 0.0, ControlCmd.DEFAULT, 0)
time.sleep(PERIOD)
def move_forward(controller, ControlCmd, speed, duration):
speed = max(0.1, min(0.5, float(speed)))
duration = max(0.1, min(8.0, float(duration)))
started = time.monotonic()
while True:
elapsed = time.monotonic() - started
if elapsed >= duration:
break
remaining = duration - elapsed
if elapsed < 0.15:
factor = max(0.45, min(1.0, 0.45 + elapsed / 0.15 * 0.55))
elif remaining < 0.22:
factor = max(0.0, remaining / 0.22)
else:
factor = 1.0
controller.publish_cmd(speed * factor, 0.0, 0.0, ControlCmd.DEFAULT, 0)
time.sleep(PERIOD)
stop(controller, ControlCmd, 10)
try:
controller, ControlCmd = load_sdk()
emit({"event": "ready", "ok": True})
except Exception as exc:
emit({"event": "ready", "ok": False, "error": str(exc)})
raise
pending_action = None
pending_index = 0
running = True
while running:
started = time.monotonic()
try:
readable, _, _ = select.select([sys.stdin], [], [], 0)
if readable:
line = sys.stdin.readline()
if line == "":
running = False
else:
request_id = ""
try:
message = json.loads(line)
request_id = str(message.get("id", ""))
command = str(message.get("command", "")).lower()
try:
stop(controller, ControlCmd, 10)
except Exception:
pass
'''
class RobotError(RuntimeError):
pass
class RobotConnectionError(RobotError):
pass
class RobotCommandError(RobotError):
pass
class _Robot:
_ACTIONS = {
"wave": "SWING",
"swing": "SWING",
"挥手": "SWING",
"handshake": "SHAKE",
"shake": "SHAKE",
"握手": "SHAKE",
"cheer": "CHEER",
"欢呼": "CHEER",
"tear": "TEAR",
"wipe_tears": "TEAR",
"擦眼泪": "TEAR",
"dance": "DANCE",
"dance1": "DANCE",
"dance2": "DANCE1",
"dance3": "DANCE2",
"stand_up": "FALLTOSTAND",
"falltostand": "FALLTOSTAND",
"倒地起身": "FALLTOSTAND",
"lie_down": "STANDTOFALL",
"standtofall": "STANDTOFALL",
"站立倒地": "STANDTOFALL",
}
systemctl --user stop bumi-onboard-voice.service >/dev/null 2>&1 || true
for pattern in '(^|/)operation([space:]|\.py)' '\.bumi_qwen_voice_highcontrol\.py' 'Bumi_机器人本体千问语音控制'; do
pgrep -u "(id−u)"−f"(id -u)" -f "(id−u)"−f"pattern" 2>/dev/null | while read -r pid; do
[ "pid"="pid" = "pid"=""]∣∣kill−TERM"" ] || kill -TERM ""]∣∣kill−TERM"pid" >/dev/null 2>&1 || true
done
done
sleep 0.3
"""
try:
_, stdout, stderr = self._ssh.exec_command("bash -lc " + shlex.quote(script), timeout=5)
stdout.channel.recv_exit_status()
stderr.read()
except Exception:
pass
def connect(ip, ssh_port=None, username=None, password=None):
return _Robot(ip, ssh_port=ssh_port, username=username, password=password)
all = ["connect", "RobotError", "RobotConnectionError", "RobotCommandError"]
