84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""test_session.py — 端到端测试：板子端算 / 电脑端算 / 会话隔离。"""
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
import _mock_server
|
|
import session_client as sc
|
|
|
|
|
|
def main():
|
|
srv = _mock_server.MockServer(port=5599)
|
|
srv.start()
|
|
time.sleep(0.4)
|
|
try:
|
|
# ---------------- 会话 1：板子端算（点云不出板）
|
|
c1 = sc.BoardClient('127.0.0.1', 5599)
|
|
h, _ = c1.request('hello')
|
|
assert h['ok'] and h['board_can_compute'], h
|
|
|
|
h, img = c1.request('capture')
|
|
assert h['ok'] and len(img) > 1000, h
|
|
|
|
r, _ = c1.request('click', u=800.0, v=648.0, radius=10, fit_plane=True)
|
|
assert r['ok'] and r['found'], r
|
|
pm_board = np.asarray(r['p_map'], dtype=np.float64)
|
|
print('board-mode p_map =', np.round(pm_board, 4))
|
|
# 合成点放在 3 m 处
|
|
assert abs(np.linalg.norm(pm_board) - 3.0) < 0.06, pm_board
|
|
|
|
st, _ = c1.request('status')
|
|
assert st['ok'] and st['has_target'] and st['delta'], st
|
|
print('board-mode status =', st['delta'])
|
|
assert abs(st['delta'][0] - (0.2507 - pm_board[0])) < 1e-6
|
|
|
|
# ---------------- 会话 2：电脑端算（拉点云 + 回传 p_map）
|
|
c2 = sc.BoardClient('127.0.0.1', 5599)
|
|
c2.request('hello')
|
|
h2, _ = c2.request('capture')
|
|
hc, cblob = c2.request('cloud')
|
|
xyz = sc.decode_cloud(hc, cblob)
|
|
idx = sc.target_math.TargetIndex(xyz)
|
|
res, pm_pc = sc.compute_target(xyz, h2['odom'], 800.0, 648.0,
|
|
10.0, True, index=idx)
|
|
assert res['found'], res
|
|
print('pc-mode p_map =', np.round(pm_pc, 4))
|
|
assert np.allclose(pm_pc, pm_board, atol=1e-6)
|
|
|
|
r2, _ = c2.request('set_target', u=800, v=648,
|
|
p_cam=res['p_cam'].tolist(),
|
|
p_lidar=res['p_lidar'].tolist(),
|
|
p_map=pm_pc.tolist())
|
|
assert r2['ok'], r2
|
|
st2, _ = c2.request('status')
|
|
assert st2['has_target'], st2
|
|
|
|
# ---------------- 会话隔离：c1 的目标还在，c2 的 reset 不影响 c1
|
|
st1, _ = c1.request('status')
|
|
assert st1['has_target'], st1
|
|
c2.request('reset')
|
|
st2b, _ = c2.request('status')
|
|
assert not st2b['has_target'], st2b
|
|
st1b, _ = c1.request('status')
|
|
assert st1b['has_target'], st1b
|
|
print('session isolation: OK')
|
|
|
|
# ---------------- 同一帧覆盖点选（索引复用）
|
|
t0 = time.perf_counter()
|
|
for du in (-8, -4, 0, 4, 8):
|
|
rr, _ = c1.request('click', u=800.0 + du, v=648.0, radius=10, fit_plane=True)
|
|
assert rr['ok']
|
|
print('5 clicks on same frame: %.1f ms total' % ((time.perf_counter() - t0) * 1000))
|
|
|
|
c1.close()
|
|
c2.close()
|
|
print('PASS')
|
|
finally:
|
|
srv.stop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|