Skip to content
Open
22 changes: 15 additions & 7 deletions backend/packages/ai_engine/src/windup_ai_engine/_subject.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
import numpy as np
from PIL import Image

__all__ = ["bg_color", "subject_bbox", "subject_mask"]
__all__ = ["bbox_of", "bg_color", "subject_bbox", "subject_mask"]

ALPHA_THR = 128 # alpha 高于此值算不透明(与 postprocess.pack 求包围盒的口径一致)
BG_TOL = 60 # 与背景色的 RGB 绝对差之和,超过才算主体
Expand Down Expand Up @@ -52,6 +52,19 @@ def subject_mask(
return np.abs(rgb - _bg_median(rgb)).sum(axis=2) > bg_tol


def bbox_of(mask: np.ndarray) -> tuple[tuple[int, int, int, int], int] | None:
"""已有掩码时的包围盒 + 像素数。

单独拆出来是为了让**同时要掩码和包围盒**的调用方(母版预检要在掩码上再数横向连通段)
只算一次掩码;若让它自己从掩码求包围盒,那份口径就与本模块分叉了。
"""
ys, xs = np.where(mask)
if not len(ys):
return None
box = (int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1)
return box, int(mask.sum())


def subject_bbox(
img: Image.Image, alpha_thr: int = ALPHA_THR, bg_tol: int = BG_TOL
) -> tuple[tuple[int, int, int, int], int] | None:
Expand All @@ -60,9 +73,4 @@ def subject_bbox(
包围盒与像素数一起返回:两者判的不是同一件事 —— 包围盒管"主体有多大",
像素数管"包围盒里是不是真有东西"(散落的几粒噪点能把包围盒撑满整幅)。
"""
m = subject_mask(img, alpha_thr, bg_tol)
ys, xs = np.where(m)
if not len(ys):
return None
box = (int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1)
return box, int(m.sum())
return bbox_of(subject_mask(img, alpha_thr, bg_tol))
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,12 @@
ProgressPort,
)
from windup_ai_engine.postprocess import align_bottom_center, frame_durations
from windup_ai_engine.slicing import dead_frame_indices, loop_seam, motion_scale
from windup_ai_engine.slicing import (
dead_frame_indices,
limb_motion,
loop_seam,
motion_scale,
)
from windup_ai_engine.strategy.base import (
ROUTE_MATRIX,
DerivationStrategy,
Expand Down Expand Up @@ -102,44 +107,87 @@ def generate(

# ② 选路线(架构决策矩阵)。装配表里没有 = 该路线未实现,在边界上炸,
# 不要让"看着成功、内容是空"的结果流到 server 去落库。
#
# 三渲二**不经过这里**:它由 server 读 DB 判断该造型有没有 3D 资产后直接调
# generate_rendered(#122)。ROUTE_MATRIX 因此仍然只映射"由动作物理性质唯一
# 决定"的那两条,它的隐含前提不被破坏(见 strategy.base 模块注释)。
route = ROUTE_MATRIX[action.action]
# .value 而不是枚举本身:Python 3.11+ 的 str-mixin 枚举 __format__ 会给出
# "ActionType.WALK",这串字最终是用户看到的进度文案(3.12.13 实测)。
progress.step("route", _TICK_ROUTE, _TOTAL, f"{action.action.value} → {route.value}")
strategy = self._pick(route, action)

# ③ 生成帧(交给 strategy —— 串联)
frames = strategy.derive(
card, action, master, _BandProgress(progress, _DERIVE_FROM, _DERIVE_TO)
)
return self._finish(frames, action, route, progress, canvas)

def generate_rendered(
self,
card: CharacterCard,
action: ActionSpec,
rigged_model: bytes,
progress: ProgressPort,
canvas: tuple[int, int] | None = None,
) -> GeneratedAction:
"""三渲二入口。见 ``ports.CharacterGeneratorPort.generate_rendered``。

**没有母版预检那一道**,不是漏了:``check_master`` 判的是尺寸 / 比例 / 空图,
而这条路线的输入是 3D 模型,那些判据一条都不适用。模型自身的预检在 server
建资产那一步做 —— 那才是花钱的地方,这里已经不花钱了。
"""
route = GenRoute.RENDER_3D
progress.step("route", _TICK_ROUTE, _TOTAL, f"{action.action.value} → {route.value}")
strategy = self._pick(route, action)
frames = strategy.derive(
card, action, rigged_model, _BandProgress(progress, _DERIVE_FROM, _DERIVE_TO)
)
return self._finish(frames, action, route, progress, canvas)

# ── 两个入口共用的部分 ────────────────────────────────────────────────
def _pick(self, route: GenRoute, action: ActionSpec) -> DerivationStrategy:
strategy = self._by_route.get(route)
if strategy is None:
raise NotImplementedError(
f"动作 {action.action.value} 分流到 {route.value},但未注入该路线的 strategy。"
f"已装配:{sorted(r.value for r in self._by_route)}。"
)
return strategy

# ③ 生成帧(交给 strategy —— 串联)
frames = strategy.derive(
card, action, master, _BandProgress(progress, _DERIVE_FROM, _DERIVE_TO)
)
def _finish(
self,
frames: list[bytes],
action: ActionSpec,
route: GenRoute,
progress: ProgressPort,
canvas: tuple[int, int] | None,
) -> GeneratedAction:
"""出帧之后的公共尾段:帧数对账 → 脚线对齐 → 量成色 → 出参。

# ③.5 帧数必须与契约相符。A2 把 n_frames 从 len(poses) 的推导值改成调用方直接声明的
两个入口共用同一份,不是为了少写几行:这几道闸是**对所有路线**的约束,
复制一份就会有一天只在一条路线上被改。
"""
# 帧数必须与契约相符。A2 把 n_frames 从 len(poses) 的推导值改成调用方直接声明的
# 承诺,而抽帧那两个函数都会**静默少给**:slicing.pick_cycle / pick_oneshot 在
# `len(dense) <= n`(或动作区间比 n 短)时 return frames/span,长度不足且不报错
# (2026-08-08 读码复核)。少给的后果不是崩溃而是"短一截的动作":时长表由
# frame_durations(…, len(frames)) 现算,长度自洽,server 看不出异常,用户拿到
# 一段步子没走完的循环。故在此对账 —— 钱已经花了,但至少不让错产物流下去。
# 放在 generator 而不是某个 strategy 里:这样将来任何新路线都受同一条约束。
if len(frames) != action.n_frames:
raise ValueError(
f"{route.value} 要 {action.n_frames} 帧,实际产出 {len(frames)} 帧。"
"抽帧源帧数不足(i2v 视频太短 / 动作区间过窄)时会静默少给,"
"请调小 n_frames 或加长视频。"
)

# 最后一公里:脚线对齐成原地序列帧(直接对齐到调用方要的画布尺寸)
# 最后一公里:脚线对齐成原地序列帧(直接对齐到调用方要的画布尺寸)
aligned = self._lastmile(frames, progress, canvas)

# 量交付成色。在**对齐之后**量,量的是用户真正会看到的那组帧:抠图 / 像素化 /
# 量交付成色。在**对齐之后**量,量的是用户真正会看到的那组帧:抠图 / 像素化 /
# 对齐都会改像素,在中间任何一步量出来的数都描述不了交付物。
quality = self._assess(aligned, action)

# ⑥ 出参:帧 + 逐帧时长 + 成色(上传 / 落库在 server 侧)
progress.step(
"package", _TICK_PACKAGE, _TOTAL,
f"{len(aligned)} 帧 + 逐帧时长(动量 {quality.motion_scale:.2f},"
Expand All @@ -165,6 +213,9 @@ def _assess(self, frames: list[Image.Image], action: ActionSpec) -> ActionQualit
motion_scale=motion_scale(frames),
dead_frames=dead_frame_indices(frames),
loop_seam=loop_seam(frames) if is_cyclic(action) else None,
# 分区动量:整幅指标的盲区补充。自动绑骨漏认一条肢体时那块网格每帧同姿势,
# 而 motion_scale 与死帧全部正常。
limbs=limb_motion(frames),
)

def _lastmile(
Expand Down
Loading
Loading