151 lines
5.1 KiB
GDScript
151 lines
5.1 KiB
GDScript
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(), "空参考序列立即标记完成")
|