opendance-windows/build/integration/probe-ws.gd

extends SceneTree

## WebSocket 契约一致性探针（T-WIN-020）
##
## 用法（由 build/accept-ws.sh 调用，也可手工执行）：
##   godot --headless --main-pack <包内 OpenDance.pck> \
##         --script /绝对路径/build/integration/probe-ws.gd \
##         -- --port 18765 --duration-ms 60000 --report /tmp/ws-report.json
##
## 为什么需要它：`tests/test_pose_server.gd` 用的是**同引擎内的客户端**
## （Godot 自己写的 WebSocketPeer），而真实对端是 **Android（Kotlin/OkHttp）**。
## 同引擎内的自测会共享同样的假设（超时判定、分包、文本/二进制边界、关闭语义），
## 因此**必须有一个独立实现的客户端**来对照 —— 这就是本探针 + Node 客户端的组合：
##   * 本探针：把**发布包里的** PoseServer 跑起来（只加载包内代码），
##     把收到的连接/控制/帧/断开事件汇总成 JSON 报告；
##   * `build/integration/ws-client.mjs`：Node 端按契约 v0.2.0 逐条跑场景并断言。
##
## 探针自身不做判定（只收集事实），判定在 Node 客户端与 accept-ws.sh 里，避免
## 「用被测实现来判定被测实现」。

const DEFAULT_PORT := 18765
const DEFAULT_DURATION_MS := 60_000

var _server: Node = null
var _started := false
var _done := false
var _start_ms := 0
var _port := DEFAULT_PORT
var _duration_ms := DEFAULT_DURATION_MS
var _report_path := "/tmp/opendance-ws-report.json"
## 可选：客户端跑完后 touch 这个文件，探针提前收尾（比固定时长更快、更稳）
var _stop_file := ""

var _events: Array = []
var _connected: Array = []
var _disconnected: Array = []
var _replaced: Array = []
var _controls := {}
## 每条连接的 hello 标识（client_id → {device, deviceId}），
## 用来在报告核对阶段回答「某类对端有没有被误踢」——断开原因里没有设备名，
## 只有 client_id 可用，所以必须在这里建立映射。
var _hellos: Array = []
var _frames := 0
var _last_frame_id := -1
var _frame_id_regressions := 0
var _first_frame_ms := 0
var _last_frame_ms := 0


func _initialize() -> void:
	var args := OS.get_cmdline_user_args()
	var i := 0
	while i < args.size():
		match args[i]:
			"--port":
				_port = int(args[i + 1]); i += 2
			"--duration-ms":
				_duration_ms = int(args[i + 1]); i += 2
			"--report":
				_report_path = args[i + 1]; i += 2
			"--stop-file":
				_stop_file = args[i + 1]; i += 2
			_:
				i += 1


func _process(_delta: float) -> bool:
	if not _started:
		_started = true
		_start_ms = Time.get_ticks_msec()
		_start_server()
	if _done:
		return true
	var elapsed := Time.get_ticks_msec() - _start_ms
	if elapsed >= _duration_ms:
		_finish()
	elif _stop_file != "" and FileAccess.file_exists(_stop_file):
		_finish()
	return _done


func _start_server() -> void:
	var script: GDScript = load("res://src/net/pose_server.gd")
	if script == null:
		print("[ws] ✗ 无法加载 res://src/net/pose_server.gd")
		_finish(2)
		return
	_server = script.new()
	_server.name = "ProbePoseServer"
	root.add_child(_server)

	_server.listening.connect(_on_listening)
	_server.listen_failed.connect(_on_listen_failed)
	_server.client_connected.connect(_on_client_connected)
	_server.client_disconnected.connect(_on_client_disconnected)
	_server.device_replaced.connect(_on_device_replaced)
	_server.frame_received.connect(_on_frame_received)
	_server.control_received.connect(_on_control_received)

	var err: int = _server.listen("127.0.0.1", _port)
	if err != OK:
		print("[ws] ✗ listen(%d) 失败：%s" % [_port, error_string(err)])
		_finish(2)


func _on_listening(port: int) -> void:
	# accept-ws.sh 靠这一行判断「服务端已就绪」
	print("[ws] listening on 127.0.0.1:%d（契约 §1 端点）" % port)


func _on_listen_failed(port: int, reason: String) -> void:
	print("[ws] ✗ 监听 %d 失败：%s" % [port, reason])


func _now_ms() -> int:
	return Time.get_ticks_msec() - _start_ms


func _on_client_connected(client_id: int, device: String) -> void:
	_connected.append({"id": client_id, "device": device, "atMs": _now_ms()})
	_events.append({"atMs": _now_ms(), "kind": "connected", "id": client_id, "device": device})


func _on_client_disconnected(client_id: int, reason: String) -> void:
	_disconnected.append({"id": client_id, "reason": reason, "atMs": _now_ms()})
	_events.append({"atMs": _now_ms(), "kind": "disconnected", "id": client_id, "reason": reason})


func _on_device_replaced(device_id: String) -> void:
	_replaced.append(device_id)
	_events.append({"atMs": _now_ms(), "kind": "device_replaced", "deviceId": device_id})


func _on_frame_received(frame) -> void:
	_frames += 1
	if _first_frame_ms == 0:
		_first_frame_ms = _now_ms()
	_last_frame_ms = _now_ms()
	if _last_frame_id >= 0 and frame.frame_id <= _last_frame_id:
		_frame_id_regressions += 1
	_last_frame_id = frame.frame_id


func _on_control_received(client_id: int, message: Dictionary) -> void:
	var type := str(message.get("type", "(无 type)"))
	_controls[type] = int(_controls.get(type, 0)) + 1
	_events.append({"atMs": _now_ms(), "kind": "control", "id": client_id, "type": type})
	if type == "hello":
		_hellos.append({
			"id": client_id,
			"device": str(message.get("device", "")),
			"deviceId": str(message.get("deviceId", "")),
			"atMs": _now_ms(),
		})


func _finish(code: int = 0) -> void:
	if _done:
		return
	_done = true
	var report := {
		"port": _port,
		"durationMs": _now_ms(),
		"server": _server.snapshot() if _server != null else {},
		"connected": _connected,
		"disconnected": _disconnected,
		"deviceReplaced": _replaced,
		"controls": _controls,
		"hellos": _hellos,
		"framesReceived": _frames,
		"lastFrameId": _last_frame_id,
		# 注意：这是**跨连接**的全局计数。frame_id 是「每条连接各自从 0 单调递增」，
		# 新连接重新从 0 开始完全合规，因此该字段只作参考，不能用来判定丢帧/回退。
		"frameIdRegressionsAcrossConnections": _frame_id_regressions,
		"firstFrameAtMs": _first_frame_ms,
		"lastFrameAtMs": _last_frame_ms,
		"events": _events,
	}
	if _server != null:
		_server.stop()
	var file := FileAccess.open(_report_path, FileAccess.WRITE)
	if file == null:
		print("[ws] ✗ 无法写入报告 %s" % _report_path)
	else:
		file.store_string(JSON.stringify(report, "  "))
		file.close()
		print("[ws] 报告已写入 %s（连接 %d、帧 %d、控制帧 %s、替换 %d）" % [
			_report_path, _connected.size(), _frames, JSON.stringify(_controls), _replaced.size()])
	quit(code)