feat(win): T-WIN-007 特征提取 + 流式 DTW 评分引擎
This commit is contained in:
@@ -0,0 +1,150 @@
|
||||
extends TestCase
|
||||
|
||||
## T-WIN-007：DTW 评分引擎
|
||||
|
||||
const PoseFeatures = preload("res://src/score/pose_features.gd")
|
||||
const DtwMatcher = preload("res://src/score/dtw.gd")
|
||||
|
||||
|
||||
func _vector(seed_value: float, dim: int = 108) -> PoseFeatures.FeatureVector:
|
||||
var v := PoseFeatures.FeatureVector.new()
|
||||
for i in dim:
|
||||
v.values[i] = sin(float(i) * 0.1 + seed_value)
|
||||
v.weights[i] = 1.0
|
||||
return v
|
||||
|
||||
|
||||
func _sequence(count: int, seed_value: float) -> Array:
|
||||
var out: Array = []
|
||||
for i in count:
|
||||
out.append(_vector(seed_value + float(i) * 0.01))
|
||||
return out
|
||||
|
||||
|
||||
func test_identical_sequences_have_zero_distance() -> void:
|
||||
var seq := _sequence(20, 0.0)
|
||||
var result := DtwMatcher.compute_full(seq, seq)
|
||||
almost(result["distance"], 0.0, 1e-5, "完全相同序列距离为 0")
|
||||
almost(result["normalized"], 0.0, 1e-6, "归一化距离为 0")
|
||||
eq(result["steps"], 20, "路径长度等于序列长度")
|
||||
|
||||
|
||||
func test_different_sequences_have_positive_distance() -> void:
|
||||
var a := _sequence(20, 0.0)
|
||||
var b := _sequence(20, 5.0)
|
||||
var result := DtwMatcher.compute_full(a, b)
|
||||
ok(result["distance"] > 0.0, "不同序列距离大于 0", "实际 %.6f" % result["distance"])
|
||||
ok(result["normalized"] > 0.0, "归一化距离大于 0")
|
||||
|
||||
|
||||
func test_empty_input_is_handled() -> void:
|
||||
var empty: Array = []
|
||||
var result := DtwMatcher.compute_full(empty, _sequence(5, 0.0))
|
||||
eq(result["steps"], 0, "空查询序列返回空结果")
|
||||
is_true(is_inf(result["distance"]), "空输入距离为 INF")
|
||||
|
||||
var result2 := DtwMatcher.compute_full(_sequence(5, 0.0), empty)
|
||||
eq(result2["steps"], 0, "空参考序列返回空结果")
|
||||
|
||||
|
||||
func test_path_is_monotonic_and_starts_at_origin() -> void:
|
||||
var a := _sequence(15, 0.0)
|
||||
var b := _sequence(15, 1.0)
|
||||
var result := DtwMatcher.compute_full(a, b)
|
||||
var path: Array = result["path"]
|
||||
ok(path.size() > 0, "路径非空")
|
||||
eq(path[0], Vector2i(0, 0), "路径起点为 (0,0)")
|
||||
for k in range(1, path.size()):
|
||||
var prev: Vector2i = path[k - 1]
|
||||
var cur: Vector2i = path[k]
|
||||
var di := cur.x - prev.x
|
||||
var dj := cur.y - prev.y
|
||||
ok(di >= 0 and dj >= 0 and (di + dj) >= 1, "路径第 %d 步单调前进" % k)
|
||||
|
||||
|
||||
func test_band_constraint_limits_warping() -> void:
|
||||
var a := _sequence(30, 0.0)
|
||||
var b := _sequence(30, 1.0)
|
||||
var result := DtwMatcher.compute_full(a, b, 3)
|
||||
for point in result["path"]:
|
||||
ok(absi(point.x - point.y) <= 3, "路径点 (%d,%d) 在带宽 3 内" % [point.x, point.y])
|
||||
|
||||
|
||||
func test_time_warping_absorbs_local_stretch() -> void:
|
||||
# 参考序列把 a 中每一帧重复两次；DTW 应以较小代价吸收这种时间伸缩
|
||||
var base := _sequence(12, 0.0)
|
||||
var stretched: Array = []
|
||||
for v in base:
|
||||
stretched.append(v)
|
||||
stretched.append(v)
|
||||
|
||||
var plain := DtwMatcher.compute_full(base, base)
|
||||
var warped := DtwMatcher.compute_full(base, stretched)
|
||||
ok(
|
||||
warped["normalized"] < plain["normalized"] + 1e-5,
|
||||
"时间伸缩后归一化代价不明显增大",
|
||||
"原始 %.6f，伸缩后 %.6f" % [plain["normalized"], warped["normalized"]]
|
||||
)
|
||||
|
||||
|
||||
func test_streaming_matches_offline_on_identical_input() -> void:
|
||||
var seq := _sequence(25, 0.0)
|
||||
var streaming := DtwMatcher.StreamingDtw.new(seq, DtwMatcher.DEFAULT_BAND)
|
||||
var last: Dictionary = {}
|
||||
for v in seq:
|
||||
last = streaming.push(v)
|
||||
|
||||
almost(last["cost"], 0.0, 1e-5, "流式 DTW 对相同序列代价为 0")
|
||||
eq(streaming.frames_consumed(), 25, "流式 DTW 消费 25 帧")
|
||||
is_true(streaming.is_finished(), "走完全部参考帧后标记完成")
|
||||
almost(streaming.progress(), 1.0, 1e-6, "进度为 1.0")
|
||||
|
||||
|
||||
func test_streaming_reports_progress_and_cursor() -> void:
|
||||
var reference := _sequence(30, 0.0)
|
||||
var streaming := DtwMatcher.StreamingDtw.new(reference, 5)
|
||||
var query := _sequence(10, 0.0)
|
||||
for v in query:
|
||||
streaming.push(v)
|
||||
|
||||
eq(streaming.frames_consumed(), 10, "已消费 10 帧")
|
||||
ok(streaming.reference_cursor() >= 0, "存在参考游标", "实际 %d" % streaming.reference_cursor())
|
||||
ok(streaming.progress() > 0.0 and streaming.progress() < 1.0, "进度落在 (0,1)")
|
||||
is_false(streaming.is_finished(), "参考未走完时不标记完成")
|
||||
|
||||
|
||||
func test_streaming_penalizes_deviation_monotonically() -> void:
|
||||
var reference := _sequence(20, 0.0)
|
||||
|
||||
var close := DtwMatcher.StreamingDtw.new(reference, 5)
|
||||
for v in _sequence(20, 0.05):
|
||||
close.push(v)
|
||||
|
||||
var far := DtwMatcher.StreamingDtw.new(reference, 5)
|
||||
for v in _sequence(20, 3.0):
|
||||
far.push(v)
|
||||
|
||||
ok(
|
||||
close.normalized_cost() < far.normalized_cost(),
|
||||
"偏差小的序列代价更低",
|
||||
"接近 %.6f，偏离 %.6f" % [close.normalized_cost(), far.normalized_cost()]
|
||||
)
|
||||
|
||||
|
||||
func test_streaming_reset_clears_state() -> void:
|
||||
var reference := _sequence(10, 0.0)
|
||||
var streaming := DtwMatcher.StreamingDtw.new(reference, 3)
|
||||
for v in _sequence(5, 0.0):
|
||||
streaming.push(v)
|
||||
streaming.reset()
|
||||
|
||||
eq(streaming.frames_consumed(), 0, "重置后帧计数归零")
|
||||
eq(streaming.reference_cursor(), -1, "重置后游标归零")
|
||||
is_false(streaming.is_finished(), "重置后未完成")
|
||||
|
||||
|
||||
func test_streaming_with_empty_reference() -> void:
|
||||
var streaming := DtwMatcher.StreamingDtw.new([], 5)
|
||||
var result := streaming.push(_vector(0.0))
|
||||
is_true(is_inf(result["cost"]), "空参考序列返回 INF 代价")
|
||||
is_true(streaming.is_finished(), "空参考序列立即标记完成")
|
||||
@@ -0,0 +1,144 @@
|
||||
extends TestCase
|
||||
|
||||
## DTW 特征向量构造 —— 契约 contracts/skeleton-topology.md §5
|
||||
## 每帧 108 维：[33 × (x, y, z), 9 个关节角余弦]
|
||||
|
||||
const KeypointFrame = preload("res://src/net/keypoint_frame.gd")
|
||||
const PoseFeatures = preload("res://src/score/pose_features.gd")
|
||||
|
||||
|
||||
## 构造一个站姿：肩宽 0.2、髋宽 0.1，四肢按解剖学方向摆放
|
||||
func _upright_frame(confidence_value: float = 1.0) -> KeypointFrame.PoseFrame:
|
||||
var frame := KeypointFrame.PoseFrame.new()
|
||||
frame.points.resize(33)
|
||||
frame.confidence.resize(33)
|
||||
frame.points.fill(Vector3(0.5, 0.5, 0.0))
|
||||
frame.confidence.fill(confidence_value)
|
||||
|
||||
frame.points[KeypointFrame.NOSE] = Vector3(0.50, 0.18, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_SHOULDER] = Vector3(0.60, 0.30, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_SHOULDER] = Vector3(0.40, 0.30, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_ELBOW] = Vector3(0.70, 0.45, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_ELBOW] = Vector3(0.30, 0.45, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_WRIST] = Vector3(0.78, 0.60, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_WRIST] = Vector3(0.22, 0.60, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_HIP] = Vector3(0.55, 0.62, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_HIP] = Vector3(0.45, 0.62, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_KNEE] = Vector3(0.56, 0.78, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_KNEE] = Vector3(0.44, 0.78, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_ANKLE] = Vector3(0.57, 0.94, 0.0)
|
||||
frame.points[KeypointFrame.RIGHT_ANKLE] = Vector3(0.43, 0.94, 0.0)
|
||||
return frame
|
||||
|
||||
|
||||
func test_feature_dimension_is_108() -> void:
|
||||
var feature := PoseFeatures.build(_upright_frame())
|
||||
eq(feature.dimension(), 108, "特征维度为 108（33×3 + 9）")
|
||||
eq(PoseFeatures.FEATURE_DIM, 108, "FEATURE_DIM 常量正确")
|
||||
eq(PoseFeatures.ANGLE_COUNT, 9, "关节角数量为 9")
|
||||
|
||||
|
||||
func test_coordinates_are_copied_verbatim() -> void:
|
||||
var frame := _upright_frame()
|
||||
var feature := PoseFeatures.build(frame, false)
|
||||
for i in 33:
|
||||
almost(feature.values[i * 3], frame.points[i].x, 1e-6, "第 %d 点 x 复制正确" % i)
|
||||
almost(feature.values[i * 3 + 1], frame.points[i].y, 1e-6, "第 %d 点 y 复制正确" % i)
|
||||
almost(feature.values[i * 3 + 2], frame.points[i].z, 1e-6, "第 %d 点 z 复制正确" % i)
|
||||
|
||||
|
||||
func test_all_weights_one_when_confidence_high() -> void:
|
||||
var feature := PoseFeatures.build(_upright_frame(1.0))
|
||||
var all_one := true
|
||||
for w in feature.weights:
|
||||
if absf(w - 1.0) > 1e-6:
|
||||
all_one = false
|
||||
break
|
||||
is_true(all_one, "全部可信时权重均为 1")
|
||||
|
||||
|
||||
func test_low_confidence_dimensions_are_zero_weighted() -> void:
|
||||
var frame := _upright_frame(1.0)
|
||||
frame.confidence[KeypointFrame.LEFT_WRIST] = 0.2
|
||||
var feature := PoseFeatures.build(frame)
|
||||
var base := KeypointFrame.LEFT_WRIST * 3
|
||||
almost(feature.weights[base], 0.0, 1e-6, "低置信度点 x 维权重为 0")
|
||||
almost(feature.weights[base + 1], 0.0, 1e-6, "低置信度点 y 维权重为 0")
|
||||
almost(feature.weights[base + 2], 0.0, 1e-6, "低置信度点 z 维权重为 0")
|
||||
almost(feature.weights[KeypointFrame.RIGHT_WRIST * 3], 1.0, 1e-6, "其它点权重不受影响")
|
||||
|
||||
|
||||
func test_angle_of_straight_limb_is_minus_one() -> void:
|
||||
# 左肘角度：左肩(0.60,0.30) — 左肘(0.70,0.45) — 左腕(0.80,0.60) 共线
|
||||
var frame := _upright_frame()
|
||||
frame.points[KeypointFrame.LEFT_WRIST] = Vector3(0.80, 0.60, 0.0)
|
||||
var feature := PoseFeatures.build(frame)
|
||||
var elbow_left_index := 33 * 3 + 2
|
||||
almost(feature.values[elbow_left_index], -1.0, 1e-4, "共线肢体关节角余弦为 -1")
|
||||
almost(feature.weights[elbow_left_index], 1.0, 1e-6, "共线角度权重为 1")
|
||||
|
||||
|
||||
func test_angle_of_right_angle_is_zero() -> void:
|
||||
# 构造左肘为直角：肩(0.60,0.30) — 肘(0.70,0.30) — 腕(0.70,0.40)
|
||||
var frame := _upright_frame()
|
||||
frame.points[KeypointFrame.LEFT_ELBOW] = Vector3(0.70, 0.30, 0.0)
|
||||
frame.points[KeypointFrame.LEFT_WRIST] = Vector3(0.70, 0.40, 0.0)
|
||||
var feature := PoseFeatures.build(frame)
|
||||
almost(feature.values[33 * 3 + 2], 0.0, 1e-4, "直角关节角余弦为 0")
|
||||
|
||||
|
||||
func test_angles_with_untrusted_endpoint_are_zero_weighted() -> void:
|
||||
var frame := _upright_frame()
|
||||
frame.confidence[KeypointFrame.LEFT_HIP] = 0.1
|
||||
var feature := PoseFeatures.build(frame)
|
||||
almost(feature.weights[33 * 3 + 0], 0.0, 1e-6, "左肩角（依赖左髋）权重归零")
|
||||
almost(feature.weights[33 * 3 + 4], 0.0, 1e-6, "左髋角（依赖左肩/左髋）权重归零")
|
||||
almost(feature.weights[33 * 3 + 1], 1.0, 1e-6, "右肩角不受影响")
|
||||
|
||||
|
||||
func test_identical_frames_have_zero_distance() -> void:
|
||||
var a := PoseFeatures.build(_upright_frame())
|
||||
var b := PoseFeatures.build(_upright_frame())
|
||||
almost(PoseFeatures.weighted_distance(a, b), 0.0, 1e-9, "相同帧距离为 0")
|
||||
|
||||
|
||||
func test_distance_ignores_untrusted_dimensions() -> void:
|
||||
var frame_a := _upright_frame()
|
||||
var frame_b := _upright_frame()
|
||||
# 让左腕不可信并在 b 中挪动它：距离应不受影响
|
||||
frame_a.confidence[KeypointFrame.LEFT_WRIST] = 0.1
|
||||
frame_b.confidence[KeypointFrame.LEFT_WRIST] = 0.1
|
||||
frame_b.points[KeypointFrame.LEFT_WRIST] = Vector3(0.95, 0.95, 0.5)
|
||||
almost(PoseFeatures.weighted_distance(PoseFeatures.build(frame_a), PoseFeatures.build(frame_b)), 0.0, 1e-9, "不可信维度不参与距离")
|
||||
|
||||
|
||||
func test_distance_grows_with_deviation() -> void:
|
||||
var base := _upright_frame()
|
||||
var near := _upright_frame()
|
||||
near.points[KeypointFrame.LEFT_WRIST] = Vector3(0.79, 0.61, 0.0)
|
||||
var far := _upright_frame()
|
||||
far.points[KeypointFrame.LEFT_WRIST] = Vector3(0.95, 0.10, 0.0)
|
||||
|
||||
var d_near := PoseFeatures.weighted_distance(PoseFeatures.build(base), PoseFeatures.build(near))
|
||||
var d_far := PoseFeatures.weighted_distance(PoseFeatures.build(base), PoseFeatures.build(far))
|
||||
ok(d_near < d_far, "偏离越大距离越大", "近 %.6f，远 %.6f" % [d_near, d_far])
|
||||
|
||||
|
||||
func test_mirrored_frame_is_normalized_before_feature_extraction() -> void:
|
||||
var frame := _upright_frame()
|
||||
frame.flags = KeypointFrame.FLAG_MIRRORED
|
||||
var mirrored := PoseFeatures.build(frame, true)
|
||||
var unmirrored := PoseFeatures.build(frame, false)
|
||||
# 镜像归一化后左腕 x 应变为 1 - 0.78
|
||||
almost(mirrored.values[KeypointFrame.LEFT_WRIST * 3], 1.0 - 0.78, 1e-4, "镜像帧已翻转 x")
|
||||
almost(unmirrored.values[KeypointFrame.LEFT_WRIST * 3], 0.78, 1e-4, "保留原始坐标时 x 不变")
|
||||
ok(PoseFeatures.weighted_distance(mirrored, unmirrored) > 0.0, "两种解释产生不同特征")
|
||||
|
||||
|
||||
func test_angle_labels_cover_all_nine() -> void:
|
||||
var labels: Array[String] = []
|
||||
for i in 9:
|
||||
labels.append(PoseFeatures.angle_label(i))
|
||||
eq(labels.size(), 9, "9 个关节角均有标签")
|
||||
for label in labels:
|
||||
ok(label != "", "标签非空：%s" % label)
|
||||
Reference in New Issue
Block a user