add docs+code

880dff9 verified about 2 months ago

17.1 kB

"""Prompt 臂（Incantation 移植）：动作走文本，逐 latent 帧（逐 cell）的 prompt 控制。零新增参数、零钩子。
上游机制（vender/Incantation，只读）
--------------------------------------
Incantation 的动作接口**纯粹是自然语言**，模型里没有任何动作专用模块：
* 模型结构 `modules/causal_model.py`：`CausalWanModel` 与 Wan2.2-TI2V-5B 的 `WanModel` 参数名逐一对应
  （:9 "Weight-compatible with WanModel"、:47-48、:238、:284-286、:386-389），子模块只有
  patch_embedding / text_embedding / time_embedding / time_projection / blocks / head（:434-450），
  没有 action embedding、action cross-attn、adaLN 动作支路。`modules/attention.py` 只有注意力后端、
  RoPE、Norm，同样没有动作模块。它与教师的唯一差异是自注意力改成分块因果 + KV cache（:93-133、:135-224），
  那是流式生成的效率设计，不是动作通路。
* 动作 → 文本 `inference.py:76-159 PromptGenerator`：动作脚本（`"Move forward"` 或 `["Move forward", 秒]`）
  按 `TIME_STEP = 0.25 s`（:85）展开成时间线，每个 latent 帧一条 prompt：
  `_steps = max(1, round(duration / 0.25))`（:133-134），没给时长的用 `DURATION_MAP`（:87-131，缺省 1.0 s）；
  两个实体的时间线取最长者、短的用最后一个动作补齐（:152-154）；模板
  `f"person_1 performs {p}. boss_1 performs {b}."`（:157）。0.25 s = 16 fps 下 4 帧 = 恰好 1 个 latent 帧，
  与本项目"cell k = 帧 4k−3…4k"的粒度完全一致。
* 文本编码 `inference.py:247-260 encode_prompts`：每条**唯一**串过一次 UMT5 得 [512, 4096]（去重，:250-258），
  再按帧展开成 list（:259）。
* 逐帧对齐 `inference.py:318-353 generate`：第 fi 个 latent 帧用 `prompt_embeds_list[pi]` 做 context（:320-321），
  每帧开始把 `crossattn_cache["is_init"]` 清掉（:323-324）重新算该帧文本的 k/v。也就是说**每帧 token 只
  cross-attend 自己那帧的文本**——没有 attention mask 拼接，也没有跨帧的文本序列；逐帧对齐靠"一次只生成一帧"实现。
* 逐帧文本掩码 `causal_model.py:340-344`（`cross_out * cross_attn_mask`）+ :673-686 / :760-772
  （`frame_mask = (t > 0)`）：只有噪声帧（t > 0）收文本，干净帧（t = 0，含写 KV cache 的那一遍）不收，
  注释说明是为了"文本信息不泄漏进自注意力 KV cache"（:760-764）。
* 已发布的训练前向 `causal_model.py:589-712 _forward_train` 收的 context 是**每个样本一条**（:650-655），
  并不逐帧；逐帧文本只在流式推理里出现。训练配方（Self-Forcing 式蒸馏、逐帧 rollout）未开源
  （README "Release Status"）。
本实现的对应关系
----------------
本仓库 `arope_forward` 原生支持逐 cell 文本（`actionrope/arope.py::_embed_context / _block_forward`：
context `[b, f, L, 4096]` ⇒ q 重排成 `(b f) (h w) c`，k/v 为 `(b f) L c`，每帧 token 只 attend 自己 cell 的文本；
唯一串只算一次 text_embedding，`context_ids` 去重），这正是 Incantation "逐 latent 帧一条 prompt、每帧只看自己
文本" 的双向并行版本。所以这个臂**不需要任何钩子**（全部继承基类的恒等实现），只声明它要的数据契约：
* `text_mode = "scene_action"`：数据集不剥动作从句，动作词按 `prompts.mirror_action` 翻回真实屏幕方向
  （`text_table_eybx_mirror.pt`）。这就是它的动作通道。
* `per_cell_context = True`：context 必须是 `[b, f, 512, 4096]`（dataset 默认如此）。`extra_context` 里做核对，
  拿到整条 context 直接报错，避免"以为逐 cell 其实整条"的静默错误（`require_per_cell=False` 可放行）。
* `uses_offset_px = False`：`arope_forward(offset_px=None)`（普通 RoPE）、无 mask 通道、loss 权重全 1。
  `action_inputs` 里的 `offset_px / offset_tok / delta_tok / action_idx` 一个都不用——Incantation 也没有任何
  非文本的动作输入；`action_idx` 的信息已经以动作从句的形式在文本里（infer 侧由 `velocity_to_label(mirror=False)`
  得到同一个标签再拼进 prompt）。
一致处 / 差异处（逐条）
----------------------
1. [一致] 动作只走文本，模型零改动、零新增参数（Incantation 的 student 与教师权重同构）。
2. [一致] 粒度：1 条 prompt / 1 个 latent 帧（0.25 s = 4 视频帧）= 本项目的 1 个 cell。
3. [一致] 逐帧对齐方式：每帧 token 只 cross-attend 自己那帧的文本 k/v，不是把各帧文本拼成长序列，
   也不是 attention mask。Incantation 靠逐帧流式生成天然实现，这里靠 `(b f)` 分组的 cross-attn 并行实现；
   所有 cell 文本相同时两者都退化成整条条件。
4. [一致] 文本编码去重：唯一串只编码一次（`encode_prompts` ↔ `_embed_context` + `context_ids`）。
5. [一致] 时间线展开规则：`round(秒 / 0.25)` 个 cell，缺省时长 1.0 s，脚本短于视频用最后一个动作补齐
   （见下面 `script_to_timeline`）。
6. [差异] prompt 模板：Incantation 是多实体槽位 `person_1 performs {p}. boss_1 performs {b}.`，不含场景
   （场景由首帧给）；本数据集是单实体 + 场景从句 `In a 2.5D top down view, {scene}, the player is {action}.`
   （`data/code/scenes.py::PROMPT_TMPL`）。不能换成 Incantation 模板：模型在本数据集模板上训练、推理时
   查的是预计算文本表（不挂 T5），且转场语料要求场景从句逐 cell 变化。动作词表也不同
   （Incantation 是 Elden Ring 的 11 个玩家动作 + 31 个 boss 动作；这里是 9 个方向标签）。
7. [差异，未复刻] `cross_attn_mask = (t > 0)`：Incantation 的干净帧不收文本。本实现里帧 0（TI2V 条件帧，t=0）
   照常 cross-attend cell 0 的文本。原因：(a) 它是因果 student 的 KV-cache 设计（防文本泄漏进 cache），
   不是动作接口本身；本模型双向、无 cache，不存在泄漏问题；(b) 钩子接口里 cross-attn 输出没有单独的
   挂点（`block_mid` 在 cross-attn 之前、`block_post` 在 ffn 之后），且 `cross_attn.o` 带 bias，无法靠构造
   context 让输出恰为 0，精确复刻要改 `actionrope/arope.py`（加一个逐帧 cross-attn 掩码参数）；
   (c) 复刻会让"装上瞬间输出与原模型相同"不成立，并且与其它四臂（帧 0 都 attend 场景文本）不对等。
   若 integration 决定要它，见文末 `CROSS_ATTN_MASK_NOTE`。
8. [差异] 自注意力：Incantation 是分块因果（帧 f 只看 ≤ f 的帧）+ KV cache 流式；本实现与其它四臂一样是
   Wan 原生的双向全注意力，一次并行出 21 个 cell。这是生成范式的差异，五臂共享，不属于动作通路。
9. [差异] Incantation 已发布的训练前向是整条 context；本实现训练即逐 cell（与其推理一致，比其发布代码更"逐帧"）。
信号 / 参数量 / 初始化
---------------------
* 信号：文本（动作从句），不读 `action_inputs`。
* 新增参数：0（Incantation 同样为 0：其 student 的 state_dict 键集与教师一致）。
* 初始化：无新参数，装上瞬间与原模型**逐位**相同（`arope_forward(arm=PromptArm())` 只多了一次形状核对）。
* 旧报告口径："Wan + Prompt" = `train.py --arm plain` 的行为，本臂与之完全相同，只是走 ActionArm 接口。
推理时"动作脚本 → 逐 cell prompt"
--------------------------------
对照 `vender/Incantation/prompts/prompts.txt`（手工模式：一行一条 prompt、逐帧）与
`examples/actions_example.json`（动作模式：`[["Move forward", 1.5], "Roll forward", ...]`），本臂建议：
* 动作模式（推荐，与 `infer.py --actions "right:1.0:10,left:1.0:11"` 同一套速度/方向约定）：
  `script_to_cell_prompts(scene, [["right", 2.5], "left", ["idle", 1.0]], trigger=None)` ⇒ 21 条
  `PROMPT_TMPL` 串；`script_to_action_spec(script, mult)` ⇒ `--actions` 字符串，让 `offset_px`
  （只用于实测位移对比、不进模型）与文本说的方向一致。动作词 = 真实屏幕方向（`mirror=False` 口径）。
* 手工模式：一行一条、共 21 行（cell 0 是条件帧也要给），串必须能在 `text_table_eybx_mirror.pt` 查到，
  否则回退加载 UMT5（`infer.py::TextEncoder`）。
* CFG 负条件：逐 cell 裸动作串 `In a 2.5D top down view, the player is {action}.`
  （训练 dropout 用的就是它，`infer.py::BARE_TMPL`），不是 `NULL_PROMPT`（不在 eybx 表里）。
"""
from __future__ import annotations

import os
import sys

import torch

from baseline.base import ActionArm

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if os.path.join(ROOT, "data", "code") not in sys.path:
    sys.path.insert(0, os.path.join(ROOT, "data", "code"))
from scenes import ACTIONS, PROMPT_TMPL, SCENE_NL  # noqa: E402

# ---- Incantation 的时间线常量（inference.py:85, :133-134, :140） ----
TIME_STEP_S = 0.25          # 1 条 prompt 覆盖 0.25 s = 1 个 latent 帧 = 本项目 1 个 cell
DEFAULT_DURATION_S = 1.0    # DURATION_MAP 查不到时的缺省时长（inference.py:140）
CELLS = 21

# 动作脚本里的方向名 → scenes.ACTIONS 下标（与 actionrope/infer.py::DIR_NAMES 一致；不 import infer 是因为
# 它 import 时会带起 opencv / diffsynth 等推理依赖）
LABEL_NAMES = ["idle", "up", "up-left", "left", "down-left", "down", "down-right", "right", "up-right"]
DIR_NAMES = {
    "idle": 0, "up": 1, "up-left": 2, "left": 3, "down-left": 4,
    "down": 5, "down-right": 6, "right": 7, "up-right": 8,
    "ul": 2, "dl": 4, "dr": 6, "ur": 8, "u": 1, "d": 5, "l": 3, "r": 7,
}

CROSS_ATTN_MASK_NOTE = (
    "Incantation 的 (t > 0) 逐帧 cross-attn 掩码未复刻。要复刻需在 actionrope/arope.py::_block_forward 的 "
    "cross_attn 之后乘一个 [b, S, 1] 的逐帧掩码（帧 0 为 0），并在 arope_forward 加参数透传；"
    "这会改变帧 0 token 的表示（进而经自注意力影响全部帧），装上瞬间就不再与原模型逐位相同。"
)


class PromptArm(ActionArm):
    """零钩子：动作全在逐 cell 文本里，模型本身一个参数都不加。"""

    name = "prompt"
    # ---- 数据 / 前向契约（integration 按这几个属性配 dataset 与 arope_forward）----
    text_mode = "scene_action"     # 动作从句保留并翻回真实方向（text_table_eybx_mirror.pt）
    per_cell_context = True        # context 必须是 [b, f, 512, 4096]
    uses_offset_px = False         # arope_forward(offset_px=None)：普通 RoPE
    mask_channel = False           # 无 known/new 掩码通道
    new_weight = 1.0               # loss 权重全 1

    def __init__(self, require_per_cell: bool = True):
        super().__init__()
        self.require_per_cell = bool(require_per_cell)

    def install(self, dit) -> None:
        """无参数可建；只记下几何信息供 extra_context 的核对用。"""
        self._dim = int(getattr(dit, "dim", 0))

    def encode(self, action_inputs, grid: dict):
        """不消费 action_inputs（Incantation 没有非文本的动作输入）。返回 None ⇒ 钩子收到 feats=None。"""
        return None

    def extra_context(self, context: torch.Tensor, feats, grid: dict) -> torch.Tensor:
        """恒等；只核对 context 真是逐 cell 的（[(b f), L, dim]），否则这个臂根本没有动作通道。"""
        if self.require_per_cell and grid is not None:
            b, f = int(grid["b"]), int(grid["f"])
            if context.shape[0] != b * f:
                raise ValueError(
                    f"prompt 臂要求逐 cell 文本（context [b, f, 512, 4096] ⇒ 嵌入后 {b * f} 行），"
                    f"得到 {context.shape[0]} 行。请给 dataset text_mode='scene_action' 的 [b, 21, 512, 4096] context；"
                    f"确要整条 context 用 PromptArm(require_per_cell=False)。"
                )
        return context

    # ---- 工具 ----
    def n_new_params(self) -> int:
        return 0

    def zero_init_check(self) -> bool:
        return True  # 没有任何通向残差流的新路径

    def describe(self) -> dict:
        return dict(name=self.name, text_mode=self.text_mode, per_cell_context=self.per_cell_context,
                    uses_offset_px=self.uses_offset_px, mask_channel=self.mask_channel,
                    new_weight=self.new_weight, n_new_params=0, upstream="Incantation (text-only action interface)")


# --------------------------------------------------------------------------
# 推理：动作脚本 → 逐 cell prompt（对照 inference.py::PromptGenerator）
# --------------------------------------------------------------------------

def _steps(duration_s: float) -> int:
    """Incantation 的 `_steps`：max(1, round(秒 / 0.25))。"""
    return max(1, int(round(float(duration_s) / TIME_STEP_S)))


def script_to_timeline(script: list, n_cells: int = CELLS, default_duration_s: float = DEFAULT_DURATION_S) -> list[int]:
    """动作脚本 → 长度 n_cells 的动作标签时间线（scenes.ACTIONS 下标）。
    脚本项：`"right"`（缺省时长）或 `["right", 秒]`（None 秒 ⇒ 缺省），方向名见 DIR_NAMES，也可直接给标签下标。
    展开规则照 Incantation：每项 round(秒/0.25) 个 cell；总长不足 n_cells 用最后一个动作补齐
    （脚本为空 ⇒ idle，对应其 "Standing"）；超出则截断（本项目视频固定 21 个 cell，Incantation 是按脚本定长）。
    """
    tl: list[int] = []
    for item in script:
        if isinstance(item, (list, tuple)):
            name, dur = item[0], (item[1] if len(item) > 1 and item[1] is not None else default_duration_s)
        else:
            name, dur = item, default_duration_s
        if isinstance(name, str):
            key = name.strip().lower()
            if key not in DIR_NAMES:
                raise ValueError(f"未知动作 {name!r}，可选 {sorted(DIR_NAMES)} 或标签下标 0..8")
            label = DIR_NAMES[key]
        else:
            label = int(name)
            if not 0 <= label < len(ACTIONS):
                raise ValueError(f"动作标签下标越界: {label}")
        tl += [label] * _steps(dur)
    if not tl:
        tl = [0]
    tl += [tl[-1]] * (n_cells - len(tl))
    return tl[:n_cells]


def scene_clause(scene: str) -> str:
    """scene key（tidal_flats）或完整从句（"in a ..."）。"""
    if scene in SCENE_NL:
        return SCENE_NL[scene]
    if scene.lower().startswith("in a"):
        return scene
    raise ValueError(f"scene 应为 key {sorted(SCENE_NL)} 或 'in a ...' 从句，得到 {scene!r}")


def timeline_to_cell_prompts(scene: str, timeline: list[int], trigger: str | None = None) -> list[str]:
    """标签时间线 → 21 条数据集模板 prompt（动作词 = 真实屏幕方向，配 text_table_eybx_mirror.pt）。
    trigger "burn:<B>:<start>:<n>" / "cut:<B>:<start>:<n>"：与 infer.py::compose_cell_prompts 同一约定，
    cell [start, start+n) 用转场从句，之后切成纯 B 场景。
    """
    if len(timeline) != CELLS:
        raise ValueError(f"时间线应有 {CELLS} 个 cell，得到 {len(timeline)}")
    clauses = [scene_clause(scene)] * CELLS
    if trigger:
        kind, b, start, n = trigger.split(":")
        start, n = int(start), int(n)
        tmpl = {"burn": "{a} burning away into {b}", "cut": "{a} cutting to {b}"}
        if kind not in tmpl:
            raise ValueError(f"trigger 类型应为 burn/cut，得到 {kind!r}")
        if not (0 <= start < CELLS and n >= 1):
            raise ValueError(f"trigger 的 start/n 越界: {start}/{n}")
        b_nl = scene_clause(b)
        trig = tmpl[kind].format(a=clauses[0], b=b_nl[3:] if b_nl.startswith("in ") else b_nl)
        for k in range(start, CELLS):
            clauses[k] = trig if k < start + n else b_nl
    return [PROMPT_TMPL.format(scene=c, action=ACTIONS[l]) for c, l in zip(clauses, timeline)]


def script_to_cell_prompts(scene: str, script: list, trigger: str | None = None) -> list[str]:
    """Incantation 风格动作脚本 → 21 条逐 cell prompt。例：
    script_to_cell_prompts("tidal_flats", [["right", 2.5], "left", ["idle", 1.0]])
    """
    return timeline_to_cell_prompts(scene, script_to_timeline(script), trigger)


def script_to_action_spec(script: list, mult: float = 1.0) -> str:
    """同一脚本 → infer.py 的 `--actions` 串（"right:1.0:10,left:1.0:11"），让实测位移的参考指令与文本一致。
    时间线里连续相同的标签合并成一段；idle 段速度为 0（"idle:0:n"）。
    """
    tl = script_to_timeline(script)
    segs: list[tuple[int, int]] = []
    for lab in tl:
        if segs and segs[-1][0] == lab:
            segs[-1] = (lab, segs[-1][1] + 1)
        else:
            segs.append((lab, 1))
    return ",".join(f"{LABEL_NAMES[lab]}:{0.0 if lab == 0 else mult}:{n}" for lab, n in segs)

	"""Prompt 臂（Incantation 移植）：动作走文本，逐 latent 帧（逐 cell）的 prompt 控制。零新增参数、零钩子。

	上游机制（vender/Incantation，只读）
	--------------------------------------
	Incantation 的动作接口纯粹是自然语言，模型里没有任何动作专用模块：

	* 模型结构 `modules/causal_model.py`：`CausalWanModel` 与 Wan2.2-TI2V-5B 的 `WanModel` 参数名逐一对应
	（:9 "Weight-compatible with WanModel"、:47-48、:238、:284-286、:386-389），子模块只有
	patch_embedding / text_embedding / time_embedding / time_projection / blocks / head（:434-450），
	没有 action embedding、action cross-attn、adaLN 动作支路。`modules/attention.py` 只有注意力后端、
	RoPE、Norm，同样没有动作模块。它与教师的唯一差异是自注意力改成分块因果 + KV cache（:93-133、:135-224），
	那是流式生成的效率设计，不是动作通路。
	* 动作 → 文本 `inference.py:76-159 PromptGenerator`：动作脚本（`"Move forward"` 或 `["Move forward", 秒]`）
	按 `TIME_STEP = 0.25 s`（:85）展开成时间线，每个 latent 帧一条 prompt：
	`_steps = max(1, round(duration / 0.25))`（:133-134），没给时长的用 `DURATION_MAP`（:87-131，缺省 1.0 s）；
	两个实体的时间线取最长者、短的用最后一个动作补齐（:152-154）；模板
	`f"person_1 performs {p}. boss_1 performs {b}."`（:157）。0.25 s = 16 fps 下 4 帧 = 恰好 1 个 latent 帧，
	与本项目"cell k = 帧 4k−3…4k"的粒度完全一致。
	* 文本编码 `inference.py:247-260 encode_prompts`：每条唯一串过一次 UMT5 得 [512, 4096]（去重，:250-258），
	再按帧展开成 list（:259）。
	* 逐帧对齐 `inference.py:318-353 generate`：第 fi 个 latent 帧用 `prompt_embeds_list[pi]` 做 context（:320-321），
	每帧开始把 `crossattn_cache["is_init"]` 清掉（:323-324）重新算该帧文本的 k/v。也就是说**每帧 token 只
	cross-attend 自己那帧的文本**——没有 attention mask 拼接，也没有跨帧的文本序列；逐帧对齐靠"一次只生成一帧"实现。
	* 逐帧文本掩码 `causal_model.py:340-344`（`cross_out * cross_attn_mask`）+ :673-686 / :760-772
	（`frame_mask = (t > 0)`）：只有噪声帧（t > 0）收文本，干净帧（t = 0，含写 KV cache 的那一遍）不收，
	注释说明是为了"文本信息不泄漏进自注意力 KV cache"（:760-764）。
	* 已发布的训练前向 `causal_model.py:589-712 _forward_train` 收的 context 是每个样本一条（:650-655），
	并不逐帧；逐帧文本只在流式推理里出现。训练配方（Self-Forcing 式蒸馏、逐帧 rollout）未开源
	（README "Release Status"）。

	本实现的对应关系
	----------------
	本仓库 `arope_forward` 原生支持逐 cell 文本（`actionrope/arope.py::_embed_context / _block_forward`：
	context `[b, f, L, 4096]` ⇒ q 重排成 `(b f) (h w) c`，k/v 为 `(b f) L c`，每帧 token 只 attend 自己 cell 的文本；
	唯一串只算一次 text_embedding，`context_ids` 去重），这正是 Incantation "逐 latent 帧一条 prompt、每帧只看自己
	文本" 的双向并行版本。所以这个臂不需要任何钩子（全部继承基类的恒等实现），只声明它要的数据契约：

	* `text_mode = "scene_action"`：数据集不剥动作从句，动作词按 `prompts.mirror_action` 翻回真实屏幕方向
	（`text_table_eybx_mirror.pt`）。这就是它的动作通道。
	* `per_cell_context = True`：context 必须是 `[b, f, 512, 4096]`（dataset 默认如此）。`extra_context` 里做核对，
	拿到整条 context 直接报错，避免"以为逐 cell 其实整条"的静默错误（`require_per_cell=False` 可放行）。
	* `uses_offset_px = False`：`arope_forward(offset_px=None)`（普通 RoPE）、无 mask 通道、loss 权重全 1。
	`action_inputs` 里的 `offset_px / offset_tok / delta_tok / action_idx` 一个都不用——Incantation 也没有任何
	非文本的动作输入；`action_idx` 的信息已经以动作从句的形式在文本里（infer 侧由 `velocity_to_label(mirror=False)`
	得到同一个标签再拼进 prompt）。

	一致处 / 差异处（逐条）
	----------------------
	1. [一致] 动作只走文本，模型零改动、零新增参数（Incantation 的 student 与教师权重同构）。
	2. [一致] 粒度：1 条 prompt / 1 个 latent 帧（0.25 s = 4 视频帧）= 本项目的 1 个 cell。
	3. [一致] 逐帧对齐方式：每帧 token 只 cross-attend 自己那帧的文本 k/v，不是把各帧文本拼成长序列，
	也不是 attention mask。Incantation 靠逐帧流式生成天然实现，这里靠 `(b f)` 分组的 cross-attn 并行实现；
	所有 cell 文本相同时两者都退化成整条条件。
	4. [一致] 文本编码去重：唯一串只编码一次（`encode_prompts` ↔ `_embed_context` + `context_ids`）。
	5. [一致] 时间线展开规则：`round(秒 / 0.25)` 个 cell，缺省时长 1.0 s，脚本短于视频用最后一个动作补齐
	（见下面 `script_to_timeline`）。
	6. [差异] prompt 模板：Incantation 是多实体槽位 `person_1 performs {p}. boss_1 performs {b}.`，不含场景
	（场景由首帧给）；本数据集是单实体 + 场景从句 `In a 2.5D top down view, {scene}, the player is {action}.`
	（`data/code/scenes.py::PROMPT_TMPL`）。不能换成 Incantation 模板：模型在本数据集模板上训练、推理时
	查的是预计算文本表（不挂 T5），且转场语料要求场景从句逐 cell 变化。动作词表也不同
	（Incantation 是 Elden Ring 的 11 个玩家动作 + 31 个 boss 动作；这里是 9 个方向标签）。
	7. [差异，未复刻] `cross_attn_mask = (t > 0)`：Incantation 的干净帧不收文本。本实现里帧 0（TI2V 条件帧，t=0）
	照常 cross-attend cell 0 的文本。原因：(a) 它是因果 student 的 KV-cache 设计（防文本泄漏进 cache），
	不是动作接口本身；本模型双向、无 cache，不存在泄漏问题；(b) 钩子接口里 cross-attn 输出没有单独的
	挂点（`block_mid` 在 cross-attn 之前、`block_post` 在 ffn 之后），且 `cross_attn.o` 带 bias，无法靠构造
	context 让输出恰为 0，精确复刻要改 `actionrope/arope.py`（加一个逐帧 cross-attn 掩码参数）；
	(c) 复刻会让"装上瞬间输出与原模型相同"不成立，并且与其它四臂（帧 0 都 attend 场景文本）不对等。
	若 integration 决定要它，见文末 `CROSS_ATTN_MASK_NOTE`。
	8. [差异] 自注意力：Incantation 是分块因果（帧 f 只看 ≤ f 的帧）+ KV cache 流式；本实现与其它四臂一样是
	Wan 原生的双向全注意力，一次并行出 21 个 cell。这是生成范式的差异，五臂共享，不属于动作通路。
	9. [差异] Incantation 已发布的训练前向是整条 context；本实现训练即逐 cell（与其推理一致，比其发布代码更"逐帧"）。

	信号 / 参数量 / 初始化
	---------------------
	* 信号：文本（动作从句），不读 `action_inputs`。
	* 新增参数：0（Incantation 同样为 0：其 student 的 state_dict 键集与教师一致）。
	* 初始化：无新参数，装上瞬间与原模型逐位相同（`arope_forward(arm=PromptArm())` 只多了一次形状核对）。
	* 旧报告口径："Wan + Prompt" = `train.py --arm plain` 的行为，本臂与之完全相同，只是走 ActionArm 接口。

	推理时"动作脚本 → 逐 cell prompt"
	--------------------------------
	对照 `vender/Incantation/prompts/prompts.txt`（手工模式：一行一条 prompt、逐帧）与
	`examples/actions_example.json`（动作模式：`[["Move forward", 1.5], "Roll forward", ...]`），本臂建议：

	* 动作模式（推荐，与 `infer.py --actions "right:1.0:10,left:1.0:11"` 同一套速度/方向约定）：
	`script_to_cell_prompts(scene, [["right", 2.5], "left", ["idle", 1.0]], trigger=None)` ⇒ 21 条
	`PROMPT_TMPL` 串；`script_to_action_spec(script, mult)` ⇒ `--actions` 字符串，让 `offset_px`
	（只用于实测位移对比、不进模型）与文本说的方向一致。动作词 = 真实屏幕方向（`mirror=False` 口径）。
	* 手工模式：一行一条、共 21 行（cell 0 是条件帧也要给），串必须能在 `text_table_eybx_mirror.pt` 查到，
	否则回退加载 UMT5（`infer.py::TextEncoder`）。
	* CFG 负条件：逐 cell 裸动作串 `In a 2.5D top down view, the player is {action}.`
	（训练 dropout 用的就是它，`infer.py::BARE_TMPL`），不是 `NULL_PROMPT`（不在 eybx 表里）。
	"""
	from __future__ import annotations

	import os
	import sys

	import torch

	from baseline.base import ActionArm

	ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	if os.path.join(ROOT, "data", "code") not in sys.path:
	sys.path.insert(0, os.path.join(ROOT, "data", "code"))
	from scenes import ACTIONS, PROMPT_TMPL, SCENE_NL # noqa: E402

	# ---- Incantation 的时间线常量（inference.py:85, :133-134, :140） ----
	TIME_STEP_S = 0.25 # 1 条 prompt 覆盖 0.25 s = 1 个 latent 帧 = 本项目 1 个 cell
	DEFAULT_DURATION_S = 1.0 # DURATION_MAP 查不到时的缺省时长（inference.py:140）
	CELLS = 21

	# 动作脚本里的方向名 → scenes.ACTIONS 下标（与 actionrope/infer.py::DIR_NAMES 一致；不 import infer 是因为
	# 它 import 时会带起 opencv / diffsynth 等推理依赖）
	LABEL_NAMES = ["idle", "up", "up-left", "left", "down-left", "down", "down-right", "right", "up-right"]
	DIR_NAMES = {
	"idle": 0, "up": 1, "up-left": 2, "left": 3, "down-left": 4,
	"down": 5, "down-right": 6, "right": 7, "up-right": 8,
	"ul": 2, "dl": 4, "dr": 6, "ur": 8, "u": 1, "d": 5, "l": 3, "r": 7,
	}

	CROSS_ATTN_MASK_NOTE = (
	"Incantation 的 (t > 0) 逐帧 cross-attn 掩码未复刻。要复刻需在 actionrope/arope.py::_block_forward 的 "
	"cross_attn 之后乘一个 [b, S, 1] 的逐帧掩码（帧 0 为 0），并在 arope_forward 加参数透传；"
	"这会改变帧 0 token 的表示（进而经自注意力影响全部帧），装上瞬间就不再与原模型逐位相同。"
	)


	class PromptArm(ActionArm):
	"""零钩子：动作全在逐 cell 文本里，模型本身一个参数都不加。"""

	name = "prompt"
	# ---- 数据 / 前向契约（integration 按这几个属性配 dataset 与 arope_forward）----
	text_mode = "scene_action" # 动作从句保留并翻回真实方向（text_table_eybx_mirror.pt）
	per_cell_context = True # context 必须是 [b, f, 512, 4096]
	uses_offset_px = False # arope_forward(offset_px=None)：普通 RoPE
	mask_channel = False # 无 known/new 掩码通道
	new_weight = 1.0 # loss 权重全 1

	def __init__(self, require_per_cell: bool = True):
	super().__init__()
	self.require_per_cell = bool(require_per_cell)

	def install(self, dit) -> None:
	"""无参数可建；只记下几何信息供 extra_context 的核对用。"""
	self._dim = int(getattr(dit, "dim", 0))

	def encode(self, action_inputs, grid: dict):
	"""不消费 action_inputs（Incantation 没有非文本的动作输入）。返回 None ⇒ 钩子收到 feats=None。"""
	return None

	def extra_context(self, context: torch.Tensor, feats, grid: dict) -> torch.Tensor:
	"""恒等；只核对 context 真是逐 cell 的（[(b f), L, dim]），否则这个臂根本没有动作通道。"""
	if self.require_per_cell and grid is not None:
	b, f = int(grid["b"]), int(grid["f"])
	if context.shape[0] != b * f:
	raise ValueError(
	f"prompt 臂要求逐 cell 文本（context [b, f, 512, 4096] ⇒ 嵌入后 {b * f} 行），"
	f"得到 {context.shape[0]} 行。请给 dataset text_mode='scene_action' 的 [b, 21, 512, 4096] context；"
	f"确要整条 context 用 PromptArm(require_per_cell=False)。"
	)
	return context

	# ---- 工具 ----
	def n_new_params(self) -> int:
	return 0

	def zero_init_check(self) -> bool:
	return True # 没有任何通向残差流的新路径

	def describe(self) -> dict:
	return dict(name=self.name, text_mode=self.text_mode, per_cell_context=self.per_cell_context,
	uses_offset_px=self.uses_offset_px, mask_channel=self.mask_channel,
	new_weight=self.new_weight, n_new_params=0, upstream="Incantation (text-only action interface)")


	# --------------------------------------------------------------------------
	# 推理：动作脚本 → 逐 cell prompt（对照 inference.py::PromptGenerator）
	# --------------------------------------------------------------------------

	def _steps(duration_s: float) -> int:
	"""Incantation 的 `_steps`：max(1, round(秒 / 0.25))。"""
	return max(1, int(round(float(duration_s) / TIME_STEP_S)))


	def script_to_timeline(script: list, n_cells: int = CELLS, default_duration_s: float = DEFAULT_DURATION_S) -> list[int]:
	"""动作脚本 → 长度 n_cells 的动作标签时间线（scenes.ACTIONS 下标）。

	脚本项：`"right"`（缺省时长）或 `["right", 秒]`（None 秒 ⇒ 缺省），方向名见 DIR_NAMES，也可直接给标签下标。
	展开规则照 Incantation：每项 round(秒/0.25) 个 cell；总长不足 n_cells 用最后一个动作补齐
	（脚本为空 ⇒ idle，对应其 "Standing"）；超出则截断（本项目视频固定 21 个 cell，Incantation 是按脚本定长）。
	"""
	tl: list[int] = []
	for item in script:
	if isinstance(item, (list, tuple)):
	name, dur = item[0], (item[1] if len(item) > 1 and item[1] is not None else default_duration_s)
	else:
	name, dur = item, default_duration_s
	if isinstance(name, str):
	key = name.strip().lower()
	if key not in DIR_NAMES:
	raise ValueError(f"未知动作 {name!r}，可选 {sorted(DIR_NAMES)} 或标签下标 0..8")
	label = DIR_NAMES[key]
	else:
	label = int(name)
	if not 0 <= label < len(ACTIONS):
	raise ValueError(f"动作标签下标越界: {label}")
	tl += [label] * _steps(dur)
	if not tl:
	tl = [0]
	tl += [tl[-1]] * (n_cells - len(tl))
	return tl[:n_cells]


	def scene_clause(scene: str) -> str:
	"""scene key（tidal_flats）或完整从句（"in a ..."）。"""
	if scene in SCENE_NL:
	return SCENE_NL[scene]
	if scene.lower().startswith("in a"):
	return scene
	raise ValueError(f"scene 应为 key {sorted(SCENE_NL)} 或 'in a ...' 从句，得到 {scene!r}")


	def timeline_to_cell_prompts(scene: str, timeline: list[int], trigger: str \| None = None) -> list[str]:
	"""标签时间线 → 21 条数据集模板 prompt（动作词 = 真实屏幕方向，配 text_table_eybx_mirror.pt）。

	trigger "burn:<B>:<start>:<n>" / "cut:<B>:<start>:<n>"：与 infer.py::compose_cell_prompts 同一约定，
	cell [start, start+n) 用转场从句，之后切成纯 B 场景。
	"""
	if len(timeline) != CELLS:
	raise ValueError(f"时间线应有 {CELLS} 个 cell，得到 {len(timeline)}")
	clauses = [scene_clause(scene)] * CELLS
	if trigger:
	kind, b, start, n = trigger.split(":")
	start, n = int(start), int(n)
	tmpl = {"burn": "{a} burning away into {b}", "cut": "{a} cutting to {b}"}
	if kind not in tmpl:
	raise ValueError(f"trigger 类型应为 burn/cut，得到 {kind!r}")
	if not (0 <= start < CELLS and n >= 1):
	raise ValueError(f"trigger 的 start/n 越界: {start}/{n}")
	b_nl = scene_clause(b)
	trig = tmpl[kind].format(a=clauses[0], b=b_nl[3:] if b_nl.startswith("in ") else b_nl)
	for k in range(start, CELLS):
	clauses[k] = trig if k < start + n else b_nl
	return [PROMPT_TMPL.format(scene=c, action=ACTIONS[l]) for c, l in zip(clauses, timeline)]


	def script_to_cell_prompts(scene: str, script: list, trigger: str \| None = None) -> list[str]:
	"""Incantation 风格动作脚本 → 21 条逐 cell prompt。例：
	script_to_cell_prompts("tidal_flats", [["right", 2.5], "left", ["idle", 1.0]])
	"""
	return timeline_to_cell_prompts(scene, script_to_timeline(script), trigger)


	def script_to_action_spec(script: list, mult: float = 1.0) -> str:
	"""同一脚本 → infer.py 的 `--actions` 串（"right:1.0:10,left:1.0:11"），让实测位移的参考指令与文本一致。

	时间线里连续相同的标签合并成一段；idle 段速度为 0（"idle:0:n"）。
	"""
	tl = script_to_timeline(script)
	segs: list[tuple[int, int]] = []
	for lab in tl:
	if segs and segs[-1][0] == lab:
	segs[-1] = (lab, segs[-1][1] + 1)
	else:
	segs.append((lab, 1))
	return ",".join(f"{LABEL_NAMES[lab]}:{0.0 if lab == 0 else mult}:{n}" for lab, n in segs)