-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathMultiFloorBackHomeAction.py
More file actions
85 lines (65 loc) · 2.61 KB
/
Copy pathMultiFloorBackHomeAction.py
File metadata and controls
85 lines (65 loc) · 2.61 KB
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
"""
MultiFloorBackHomeAction — 跨楼层自主回桩
功能:自动规划跨楼层回桩路径,单楼层时等同于 GoHomeAction
依赖:需要配置多楼层地图(slamware.agent.multi_floor 能力)
"""
import time
import requests
# ── 配置 ──────────────────────────────────────────────
ROBOT_IP = "10.160.129.252"
PORT = 1448
BASE_URL = f"http://{ROBOT_IP}:{PORT}/api/core/motion/v1/actions"
# 绕过系统代理(避免 127.0.0.1:7890 超时)
SESSION = __import__("requests").Session()
SESSION.trust_env = False
# 上桩模式:"dock"=需要完成上桩 "no_dock"=到达附近即可
FLAGS = "dock"
# 上桩失败后是否返回上桩点等待
BACK_TO_LANDING = True
# 上桩重试次数
CHARGING_RETRY_COUNT = 3
POLL_INTERVAL_S = 2.0
# ──────────────────────────────────────────────────────
def build_payload():
return {
"action_name": "agent.actions.MultiFloorBackHomeAction",
"options": {
"gohome_options": {
"flags": FLAGS,
"back_to_landing": BACK_TO_LANDING,
"charging_retry_count": CHARGING_RETRY_COUNT,
"move_options": {"mode": 0},
}
},
}
def poll_until_done(action_id):
url = f"{BASE_URL}/{action_id}"
while True:
resp = SESSION.get(url, timeout=5)
data = resp.json()
status = data["state"]["status"]
result = data["state"]["result"]
reason = data["state"].get("reason", "")
if status == 4:
if result == 0:
print(f"[完成] action_id={action_id} 跨楼层回桩成功")
else:
print(f"[失败] action_id={action_id} result={result} reason={reason}")
return result == 0
status_desc = {0: "初始化中", 1: "回桩导航中"}.get(status, str(status))
print(f"[{status_desc}] action_id={action_id} ...")
time.sleep(POLL_INTERVAL_S)
def main():
print(f"[MultiFloorBackHomeAction] flags={FLAGS} retry={CHARGING_RETRY_COUNT}")
payload = build_payload()
resp = SESSION.post(BASE_URL, json=payload, timeout=5)
if resp.status_code != 200:
print(f"[错误] HTTP {resp.status_code}: {resp.text}")
return
data = resp.json()
action_id = data["action_id"]
stage = data.get("stage", "")
print(f"[已创建] action_id={action_id} stage={stage}")
poll_until_done(action_id)
if __name__ == "__main__":
main()