fix(postprocess): 逐帧补偿一段动作内的单调尺度漂移

backend/packages/ai_engine/src/windup_ai_engine/postprocess/pack.py

@@ -8,6 +8,7 @@
 
 import logging
 
+import numpy as np
 from PIL import Image
 
 _logger = logging.getLogger(__name__)
@@ -60,6 +61,47 @@ def span(counts: np.ndarray, base: float) -> float:
     return (span(rows, float(np.median(nz_rows))), span(cols, float(cols.max())))
 
 
+# 单调漂移的判定门槛(整段首尾相对变化)。低于它的不动 —— 真实身高起伏实测约 4%,
+# 把那也当漂移消掉,就成了原设计担心的"蹲下的帧被放大"。
+DRIFT_MIN_RATIO = 0.08
+
+
+def scale_drift(spans: list[float | None]) -> tuple[list[float], float]:
+    """把逐帧本体高里的**单调趋势**分离出来,返回(逐帧补偿系数, 首尾相对变化)。
+
+    存在的理由是整段共用一个缩放系数会原样保留 i2v 的推镜:实测线上两段真实产出,
+    本体高从 137→165(+20%)与 70→158(+127%),几乎无回落。统一缩放对整段乘同一个数,
+    趋势不受影响,于是角色在一个动作内单调变大。
+
+    只除趋势、不逐帧归一:后者会把走路自然的身高起伏(约 4%)一起压平,蹲下的帧被放大、
+    伸展的帧被缩小 —— 那正是本模块最初拒绝逐帧归一的原因。对本体高做一次线性拟合,
+    补偿拟合值、保留残差,两个目标就不再冲突(实测修后趋势归零,残差 1.5%–6.7%)。
+
+    ``None`` = 空帧(量不到本体):不参与拟合、系数取 1.0,其余帧照常补偿。
+
+    返回的系数以 1.0 为中心(除以均值),所以整段的**平均**尺寸不变,跨动作口径不受影响。
+    """
+    n = len(spans)
+    # 自变量用**真实帧号**而不是压缩后的序号:主要是系数必须落回对应的帧,否则空洞之后
+    # 整体错位一帧;顺带也不让空洞压短趋势的时间轴(32 帧缺 1 实测斜率差 3.7%,
+    # 落到逐帧系数上 <0.3%)。
+    x = np.array([i for i, s in enumerate(spans) if s is not None], dtype=float)
+    a = np.array([s for s in spans if s is not None], dtype=float)
+    if len(a) < 4:                            # 观测不足:三点拟不出可信趋势,拟合反而制造漂移
+        return [1.0] * n, 0.0
+    k, b = np.polyfit(x, a, 1)
+    trend = k * x + b
+    if trend.min() <= 0:                      # 拟合出非正值:数据不适合线性描述,不动
+        return [1.0] * n, 0.0
+    ratio = float(trend[-1] / trend[0] - 1.0)
+    if abs(ratio) < DRIFT_MIN_RATIO:
+        return [1.0] * n, ratio
+    comp = [1.0] * n
+    for i, c in zip(x.astype(int), trend / trend.mean(), strict=True):
+        comp[i] = float(c)
+    return comp, ratio
+
+
 def align_bottom_center(
     frames: list[Image.Image],
     cell: int = CELL,
@@ -119,7 +161,9 @@ def align_bottom_center(
     if not heights:
         return [Image.new("RGBA", (cw, ch), (0, 0, 0, 0)) for _ in frames]
     # 定标一律按**本体**跨度,不按包围盒:后者被延展物撑大,而延展物幅度随动作变。
-    spans = [s for s in (core_span(f) for f in frames) if s is not None]
+    # 逐帧补偿要按帧号索引系数,故先留一份与 frames 等长、空帧为 None 的原始表。
+    core_spans = [core_span(f) for f in frames]
+    spans = [s for s in core_spans if s is not None]
     # 腾空模式:以最低脚线(数值最大 = 站在地上)为地面基准,保留每帧的抬升量
     ground = max(b[3] for b in boxes if b) if preserve_lift else 0
     # 定标要把抬升量算进去,否则跳到最高时头顶会顶出画布被切掉
@@ -161,22 +205,36 @@ def align_bottom_center(
     # 跳到 0.961,跨过了任何合理的窗口。一个永不成立的分支比没有分支更坏。
     #
     # 关键是**不静默**:裁掉多少写进日志,让丢像素可见,而不是靠人看图发现。
+
+    # 逐帧补偿单调漂移。整段共用的 scale 只决定平均尺寸,趋势项由这里除掉;
+    # 补偿系数以 1.0 为中心,故平均尺寸与跨动作口径都不变。
+    # 空帧照常传给 scale_drift(它按帧号拟合、空位给 1.0)—— 少一个观测不该让整段不补。
+    per_frame = [1.0] * len(frames)
+    comp, ratio = scale_drift([s[0] if s is not None else None for s in core_spans])
+    if any(c != 1.0 for c in comp):
+        per_frame = [1.0 / c for c in comp]
+        _logger.info(
+            "整段尺度单调漂移 %.1f%%(i2v 推镜),已逐帧补偿;补偿区间 %.3f–%.3f",
+            ratio * 100, min(per_frame), max(per_frame),
+        )
+
     if max_full * scale > cw:
         _logger.info(
             "保尺寸一致而不压缩:整帧需 %.0fpx、画布 %dpx,两侧各溢出约 %.0fpx",
             max_full * scale, cw, (max_full * scale - cw) / 2,
         )
 
     out = []
-    for f, box in zip(frames, boxes):
+    for idx, (f, box) in enumerate(zip(frames, boxes)):
         if box is None:
             out.append(Image.new("RGBA", (cw, ch), (0, 0, 0, 0)))
             continue
         crop = f.crop(box)
-        w = max(1, round(crop.width * scale))
-        h = max(1, round(crop.height * scale))
+        fs = scale * per_frame[idx]
+        w = max(1, round(crop.width * fs))
+        h = max(1, round(crop.height * fs))
         crop = crop.resize((w, h), Image.NEAREST)
-        lift = round((ground - box[3]) * scale) if preserve_lift else 0
+        lift = round((ground - box[3]) * fs) if preserve_lift else 0
         canvas = Image.new("RGBA", (cw, ch), (0, 0, 0, 0))
         canvas.alpha_composite(crop, (cw // 2 - w // 2, int(ch * foot_line) - h - lift))
         out.append(canvas)

backend/tests/test_pack_align.py

@@ -244,3 +244,132 @@ def test_clipping_is_logged_not_silent(caplog):
         align_bottom_center(src, cell=256, cell_h=256)
     assert any("溢出" in r.message for r in caplog.records), \
         f"裁切没有上报，日志：{[r.message for r in caplog.records]}"
+
+
+# ── 一段动作内的单调漂移(#307)──────────────────────────────────────────────
+#
+# 线上真实产出实测:walk 的本体高 137→165(+20%)、custom 70→158(+127%),几乎无回落。
+# 整段共用一个缩放系数只决定平均尺寸,趋势原样保留,于是角色在一个动作内单调变大。
+
+
+# 任务 94(walk,32 帧)的逐帧本体高,直接取自线上产物。
+_REAL_WALK_SPANS = [
+    132, 133, 136, 139, 141, 142, 139, 138, 141, 146, 146, 148, 146, 145, 149, 155,
+    155, 153, 151, 154, 157, 161, 160, 161, 159, 160, 162, 170, 169, 167, 168, 168,
+]
+
+
+def test_monotonic_drift_is_removed_on_real_data():
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    comp, ratio = scale_drift(_REAL_WALK_SPANS)
+    assert ratio > 0.15, "这段真实数据本身就有 20% 漂移,判不出来说明门槛错了"
+    fixed = np.asarray(_REAL_WALK_SPANS, float) / np.asarray(comp)
+    head, tail = fixed[:8].mean(), fixed[-8:].mean()
+    assert abs(tail / head - 1) < 0.03, f"补偿后首尾仍差 {(tail/head-1)*100:.1f}%"
+
+
+def test_natural_bob_is_preserved_not_flattened():
+    """只除趋势、不逐帧归一 —— 走路自然的身高起伏必须留着。
+
+    逐帧归一会把蹲下的帧放大、伸展的帧缩小,那正是本模块最初拒绝它的原因。
+    """
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    comp, _ = scale_drift(_REAL_WALK_SPANS)
+    fixed = np.asarray(_REAL_WALK_SPANS, float) / np.asarray(comp)
+    spread = fixed.std() / fixed.mean()
+    assert spread > 0.005, "起伏被压平了,退化成逐帧归一"
+    assert spread < 0.10, f"残差 {spread*100:.1f}% 过大,趋势没除干净"
+
+
+def test_steady_sequence_is_left_alone():
+    """没有漂移就不该动。真实身高起伏约 4%,把那当漂移消掉是过度矫正。"""
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    steady = [100, 104, 98, 102, 101, 99, 103, 100] * 4
+    comp, ratio = scale_drift(steady)
+    assert abs(ratio) < 0.08
+    assert all(c == 1.0 for c in comp)
+
+
+def test_average_size_is_unchanged_so_cross_action_scale_still_holds():
+    """补偿系数以 1.0 为中心:整段平均尺寸不变,#280 的跨动作口径不受影响。"""
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    comp, _ = scale_drift(_REAL_WALK_SPANS)
+    assert abs(float(np.mean(comp)) - 1.0) < 0.01
+
+
+def test_too_few_frames_are_left_alone():
+    """三帧拟合不出可信趋势,拟合了反而制造漂移。"""
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    comp, ratio = scale_drift([100, 130, 160])
+    assert comp == [1.0, 1.0, 1.0] and ratio == 0.0
+
+
+def _drifting_bodies(n=16, lo=60, hi=140):
+    """本体高从 lo 单调涨到 hi 的合成序列,形状与线上观测到的推镜一致。"""
+    def body(h: int) -> Image.Image:
+        a = np.zeros((256, 256, 4), np.uint8)
+        w = max(2, h // 3)
+        a[200 - h:200, 128 - w // 2:128 + w // 2, 3] = 255
+        return Image.fromarray(a)
+
+    return [body(int(round(v))) for v in np.linspace(lo, hi, n)]
+
+
+def test_drift_is_still_compensated_when_a_frame_is_empty():
+    """中间夹一帧全透明,其余帧的漂移照样要补掉。
+
+    空帧只是**缺一个观测**。整段跳过补偿会让其余帧静默留着漂移 —— 本 PR 要修的问题
+    原样回来,且无声无息。
+    """
+    from windup_ai_engine.postprocess.pack import align_bottom_center, core_span
+
+    src = _drifting_bodies()
+    src[8] = Image.new("RGBA", (256, 256), (0, 0, 0, 0))
+
+    out = align_bottom_center(src, cell=256)
+    assert core_span(out[8]) is None, "空帧必须原样透明输出"
+
+    got = [core_span(f)[0] for i, f in enumerate(out) if i != 8]
+    head, tail = float(np.mean(got[:4])), float(np.mean(got[-4:]))
+    assert abs(tail / head - 1) < 0.08, (
+        f"有空帧时补偿被整段跳过,出帧仍在单调变大:首 {head:.0f} → 尾 {tail:.0f}"
+        f"({(tail/head-1)*100:+.0f}%)"
+    )
+
+
+def test_empty_frames_do_not_shift_the_trend_timeline():
+    """空帧不参与拟合,系数取 1.0,其余帧的系数与它不在时一致。"""
+    from windup_ai_engine.postprocess.pack import scale_drift
+
+    full = _REAL_WALK_SPANS
+    holed = list(full)
+    holed[8] = None
+
+    ref, _ = scale_drift(full)
+    comp, ratio = scale_drift(holed)
+    assert ratio > 0.15, "少一个观测不该让 20% 的漂移判不出来"
+    assert comp[8] == 1.0, "空帧的系数应为 1.0"
+    for i, (c, r) in enumerate(zip(comp, ref, strict=True)):
+        if i != 8:
+            assert abs(c - r) < 0.01, f"第 {i} 帧系数被空洞带偏: {c:.3f} vs {r:.3f}"
+
+
+def test_align_actually_applies_the_compensation():
+    """钉的是"补偿真的接上了",不是"函数算得对"。
+
+    只测 ``scale_drift`` 的话,把 ``align_bottom_center`` 里那一行乘法删掉,用例照样全绿
+    （变异测试逮到过）—— 那正是本仓最忌讳的"看起来成功的错结果"。
+    """
+    from windup_ai_engine.postprocess.pack import align_bottom_center, core_span
+
+    out = align_bottom_center(_drifting_bodies(), cell=256)
+    got = [core_span(f)[0] for f in out]
+    head, tail = float(np.mean(got[:4])), float(np.mean(got[-4:]))
+    assert abs(tail / head - 1) < 0.08, (
+        f"出帧后仍在单调变大:首 {head:.0f} → 尾 {tail:.0f}({(tail/head-1)*100:+.0f}%)"
+    )