Proact-VL-Washing/mage_demo/mymage_window_test.py

import os, gc, time, subprocess, importlib, warnings
from pathlib import Path

os.environ.setdefault(
    "PYTORCH_CUDA_ALLOC_CONF",
    "expandable_segments:True,max_split_size_mb:128"
)
os.environ.setdefault("ONLINE_CODEC_CACHE_DIR", "/tmp/mage_vl_codec_cache")
os.environ.setdefault("STREAMMIND_CLIP_CACHE", "/tmp/mage_vl_clips")
os.makedirs(os.environ["ONLINE_CODEC_CACHE_DIR"], exist_ok=True)
os.makedirs(os.environ["STREAMMIND_CLIP_CACHE"], exist_ok=True)

import torch
from transformers import AutoModelForCausalLM, AutoProcessor


# ===================== 配置 =====================

MODEL_ID = "microsoft/Mage-VL"
VIDEO_FILE = "/content/soccer-broadcast.mp4"

SEGMENT_SEC = 4.0

# Gate 使用：
# 前4秒 + 当前4秒 = 最多8秒视觉上下文
CONTEXT_SEC = 4.0

# LLM 保留最近几条文字事件
HISTORY_SIZE = 3

MAX_SEGMENTS = None          # 测试可改成 8
GATE_THRESHOLD = 0.50
MAX_NEW_TOKENS = 64

CODEC_MAX_PIXELS = 64_000
CODEC_TARGET_CANVAS = 24

SIMULATE_REALTIME = True


# ===================== Prompt =====================

GATE_PROMPT = """
这是足球比赛视频。
请判断视频末尾的当前时间窗口是否包含值得进一步分析的内容。
重点关注传球、带球、射门、抢断、犯规、角球、任意球、进球、
危险进攻、球员明显动作或比赛状态变化。
""".strip()


def make_prompt(start, end, history):
    history_text = "\n".join(
        f"- {x}" for x in history[-HISTORY_SIZE:]
    )

    if not history_text:
        history_text = "暂无历史事件。"

    return f"""
请用中文分析足球比赛视频。

当前重点时间：
{start:.1f}–{end:.1f} 秒

视频包含：
- 前面的视觉内容：用于理解动作上下文
- 当前时间窗口：重点分析当前发生的事情

最近已经观察到的事件：
{history_text}

请重点描述当前时间窗口发生了什么，并结合前面的视觉内容判断动作是否连续。

重点关注：
传球、带球、射门、抢断、犯规、角球、任意球、进球、
危险进攻、球员位置变化以及比赛状态变化。

如果当前只是普通画面，没有明显事件，也请简短说明。

不要猜测无法从视频确认的信息。
尽量简洁，适合实时视频分析。
""".strip()


# ===================== 模型 =====================

if not torch.cuda.is_available():
    raise RuntimeError("未检测到 CUDA")

DEVICE = "cuda"
DTYPE = torch.float16

print(f"[GPU] {torch.cuda.get_device_name(0)}")
print(f"[CUDA] {torch.version.cuda}")

print("[init] 加载 Processor...")

processor = AutoProcessor.from_pretrained(
    MODEL_ID,
    trust_remote_code=True
)

print("[init] 加载 Mage-VL...")

model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    trust_remote_code=True,
    torch_dtype=DTYPE,
    attn_implementation="sdpa",
).to(DEVICE).eval()

print("[init] 模型加载完成")

_REMOTE_PKG = type(processor).__module__.rsplit(".", 1)[0]

codec_mod = importlib.import_module(
    _REMOTE_PKG + ".codec_video_processing_mage_vl"
)

CodecConfig = codec_mod.CodecConfig

try:
    model.model._load_streammind_gate()
    print("[gate] Cognition Gate 加载成功")
except Exception as e:
    print(f"[gate] 加载失败: {e}")


# ===================== 工具 =====================

def get_duration(path):
    try:
        r = subprocess.run(
            [
                "ffprobe", "-v", "error",
                "-show_entries", "format=duration",
                "-of", "default=nw=1:nk=1",
                str(path)
            ],
            capture_output=True,
            text=True,
            check=True,
            timeout=60
        )
        return float(r.stdout.strip())
    except Exception:
        return 0.0


def codec_config():
    override = {
        "engine": "hevc",
        "target_canvas": CODEC_TARGET_CANVAS,
        "patch": 16,
    }

    kwargs = dict(processor._codec_config_defaults)
    dcvc = dict(kwargs.get("dcvc") or {})

    kwargs.update(override)

    if dcvc:
        kwargs["dcvc"] = dcvc

    kwargs["max_pixels"] = CODEC_MAX_PIXELS

    return CodecConfig(**kwargs), override


def make_chat_prompt(text):
    messages = [{
        "role": "user",
        "content": [
            {"type": "video"},
            {"type": "text", "text": text},
        ],
    }]

    return processor.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True,
    )


def extract_clip(src, start, length, out):
    tmp = out.with_name(".tmp_" + out.name)

    if tmp.exists():
        tmp.unlink()

    subprocess.run(
        [
            "ffmpeg", "-y", "-loglevel", "error",
            "-ss", f"{max(0, start):.3f}",
            "-i", str(src),
            "-t", f"{length:.3f}",
            "-c:v", "libx264",
            "-preset", "ultrafast",
            "-crf", "23",
            "-an",
            "-pix_fmt", "yuv420p",
            str(tmp)
        ],
        check=True,
        capture_output=True,
        timeout=180
    )

    tmp.replace(out)
    return out


def to_cuda(inputs):
    result = {}

    for k, v in inputs.items():
        if not hasattr(v, "to"):
            continue

        v = v.to(DEVICE)

        if k == "pixel_values":
            v = v.to(DTYPE)

        result[k] = v

    return result


def cleanup(*objs):
    for obj in objs:
        try:
            del obj
        except Exception:
            pass

    gc.collect()
    torch.cuda.empty_cache()


def gpu_mem():
    a = torch.cuda.memory_allocated() / 1024**3
    r = torch.cuda.memory_reserved() / 1024**3
    return f"{a:.2f}GB / reserved {r:.2f}GB"


# ===================== Gate =====================

@torch.inference_mode()
def run_gate(inputs):
    feat = {}

    for k in (
        "pixel_values",
        "image_grid_thw",
        "patch_positions",
    ):
        if k not in inputs:
            continue

        v = inputs[k].to(DEVICE)

        if k == "pixel_values":
            v = v.to(DTYPE)

        feat[k] = v

    logits = None

    try:
        logits = model.model.streammind_gate_forward_segments(
            [feat]
        )[0]

        # 这里取视觉序列最后位置。
        #
        # 因为输入是：
        # 前4秒 + 当前4秒
        #
        # 所以最后位置对应当前4秒末尾，
        # Gate 可以利用前面的视觉内容作为上下文。

        length = int(
            inputs["image_grid_thw"][:, 0].sum()
        )

        prob = torch.softmax(
            logits[length - 1].float(),
            dim=-1
        )[1].item()

        return float(prob)

    finally:
        cleanup(feat, logits)


# ===================== Codec =====================

def encode_video(
    clip,
    prompt,
    codec_override
):
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")

        return processor(
            text=[prompt],
            videos=[str(clip)],
            video_backend="codec",
            max_pixels=CODEC_MAX_PIXELS,
            codec_config=codec_override,
            return_tensors="pt",
            padding=False,
        )


# ===================== Mage-VL =====================

@torch.inference_mode()
def generate(inputs, max_new_tokens):
    out = model.generate(
        **inputs,
        max_new_tokens=max_new_tokens,
        do_sample=False,
        use_cache=True,
    )

    n = inputs["input_ids"].shape[1]

    text = processor.tokenizer.decode(
        out[0, n:],
        skip_special_tokens=True
    ).strip()

    del out

    return text


# ===================== 主循环 =====================

def stream_video_analysis(
    video_path,
    segment_sec=SEGMENT_SEC,
    context_sec=CONTEXT_SEC,
    gate_threshold=GATE_THRESHOLD,
    max_segments=MAX_SEGMENTS,
    max_new_tokens=MAX_NEW_TOKENS,
    simulate_realtime=SIMULATE_REALTIME,
):
    path = Path(video_path)

    if not path.exists():
        raise FileNotFoundError(video_path)

    total = get_duration(path)

    if total <= 0:
        raise RuntimeError("无法获取视频时长")

    clip_dir = Path(
        os.environ["STREAMMIND_CLIP_CACHE"]
    )
    clip_dir.mkdir(
        parents=True,
        exist_ok=True
    )

    _, codec_override = codec_config()

    start = 0.0
    index = 0

    # 最近几次文字事件
    history = []

    # 所有输出
    results = []

    yield (
        f"开始实时流式分析\n"
        f"视频={total:.1f}s | "
        f"当前窗口={segment_sec:.1f}s | "
        f"视觉上下文={context_sec:.1f}s | "
        f"历史={HISTORY_SIZE}条"
    )

    while start < total - 0.05:

        if (
            max_segments is not None
            and index >= max_segments
        ):
            break

        # 当前时间窗口
        end = min(
            total,
            start + segment_sec
        )

        current_length = end - start

        # --------------------------------------------------
        # 视觉上下文：
        #
        # 第1段：
        # 0-4
        #
        # 第2段：
        # 0-8
        #
        # 第3段：
        # 4-12
        #
        # 第4段：
        # 8-16
        # --------------------------------------------------

        context_start = max(
            0.0,
            start - context_sec
        )

        context_end = end

        visual_length = (
            context_end - context_start
        )

        index += 1

        t0 = time.time()

        clip = clip_dir / (
            f"{path.stem}_"
            f"{index:06d}_"
            f"{context_start:.2f}_"
            f"{context_end:.2f}.mp4"
        )

        inputs = None
        cuda_inputs = None

        try:

            # ==============================================
            # 1. 当前滑动视觉窗口
            # ==============================================

            extract_clip(
                path,
                context_start,
                visual_length,
                clip
            )

            # ==============================================
            # 2. Gate
            #
            # Gate 看到：
            #
            # 前4秒 + 当前4秒
            # ==============================================

            gate_prompt = make_chat_prompt(
                GATE_PROMPT
            )

            inputs = encode_video(
                clip,
                gate_prompt,
                codec_override
            )

            prob = run_gate(inputs)

            head = (
                f"**{start:.1f}–{end:.1f}s** "
                f"Gate=`{prob:.2f}` "
                f"[context={context_start:.1f}–{end:.1f}s]"
            )

            # ==============================================
            # 3. Gate 不通过
            # ==============================================

            if prob < gate_threshold:

                elapsed = time.time() - t0
                speed = current_length / elapsed

                line = (
                    f"{head} → 🔇 跳过 "
                    f"({elapsed:.2f}s / "
                    f"{speed:.2f}×)"
                )

                results.append(line)

                yield "\n\n".join(results)

            # ==============================================
            # 4. Gate 通过
            # ==============================================

            else:

                yield (
                    "\n\n".join(results)
                    + (
                        "\n\n"
                        if results else ""
                    )
                    + f"{head} → 🧠 生成中..."
                )

                # ------------------------------------------
                # LLM 使用：
                #
                # 视觉：
                # 前4秒 + 当前4秒
                #
                # 文本：
                # 最近3个事件
                # ------------------------------------------

                llm_prompt = make_prompt(
                    start,
                    end,
                    history
                )

                # 重新生成带历史 prompt 的输入
                #
                # 这里重新 processor 是为了让：
                # Gate prompt 和 LLM prompt 分开。
                #
                # 视频 Codec 本身仍然是同一个滑动窗口。
                # ------------------------------------------

                cleanup(inputs)
                inputs = None

                inputs = encode_video(
                    clip,
                    make_chat_prompt(llm_prompt),
                    codec_override
                )

                gt = time.time()

                cuda_inputs = to_cuda(
                    inputs
                )

                said = generate(
                    cuda_inputs,
                    max_new_tokens
                )

                gen_time = time.time() - gt
                elapsed = time.time() - t0
                speed = current_length / elapsed

                # ------------------------------------------
                # 保存历史事件
                # ------------------------------------------

                history.append(
                    f"{start:.1f}–{end:.1f}s：{said}"
                )

                if len(history) > HISTORY_SIZE:
                    history.pop(0)

                line = (
                    f"{head} → 🗣️ {said}\n"
                    f"   Generate={gen_time:.2f}s | "
                    f"Total={elapsed:.2f}s | "
                    f"Realtime={speed:.2f}×"
                )

                results.append(line)

                yield "\n\n".join(results)

            # ==============================================
            # 5. 模拟实时
            # ==============================================

            elapsed = time.time() - t0

            if (
                simulate_realtime
                and elapsed < current_length
            ):
                time.sleep(
                    current_length - elapsed
                )

        except Exception as e:

            results.append(
                f"**{start:.1f}–{end:.1f}s** "
                f"→ ❌ "
                f"{type(e).__name__}: {e}"
            )

            yield "\n\n".join(results)

        finally:

            cleanup(
                cuda_inputs,
                inputs
            )

            if clip.exists():
                try:
                    clip.unlink()
                except Exception:
                    pass

            start = end

            print(
                f"[segment {index}] "
                f"{start:.1f}/{total:.1f}s | "
                f"GPU {gpu_mem()}",
                flush=True
            )

    yield (
        "\n\n".join(results)
        + f"\n\n✅ 完成，共 {index} 个片段"
    )


# ===================== Main =====================

if __name__ == "__main__":

    for output in stream_video_analysis(
        VIDEO_FILE,
        segment_sec=4.0,
        context_sec=4.0,
        gate_threshold=0.50,
        max_segments=MAX_SEGMENTS,
        max_new_tokens=64,
        simulate_realtime=True,
    ):
        print("\033[H\033[J", end="")
        print("=== Mage-VL 第六版：上下文实时分析 ===\n")
        print(output)
        print("\n" + "-" * 70)