quercus/tests/mock_host/frames.py

"""Mock 宿主取帧：stdlib zlib 手写最小 PNG 编码器 + 确定性画面渲染。

- 不引入 Pillow（保持依赖小）；编码器只写 RGB8 非隔行 PNG（合法 PNG）。
- 帧内容由 (footage_id, source_time, 尺寸) 纯函数决定：随时间点变化、
  不同素材不同、绝不整幅全黑——对齐计划文档 §7 的"帧非全黑"验收。
- 时间线上的取帧语义：找到该时间点最上层启用的视频片段，取其素材
  ``media_in + (time - clip.start)`` 处的源画面（编辑改变片段覆盖关系，
  切口帧像素随之变化）。
"""
from __future__ import annotations

import hashlib
import struct
import zlib

from quercus_core.types import PngBytes, Rational

from mock_host.effects import effect_spec

# 帧像素亮度阈值：低于它视为"全黑"（渲染永远高于此值）。
_MIN_BRIGHTNESS = 24


def _seed(footage_id: str) -> int:
    return int(hashlib.md5(footage_id.encode("utf-8")).hexdigest()[:6], 16) % 256


def encode_png(width: int, height: int, rgb: bytes) -> bytes:
    """把 RGB8 平面数据编码为合法 PNG 字节（filter 0，逐行前置 0）。"""
    if len(rgb) != width * height * 3:
        raise ValueError(f"RGB 数据长度不符: {len(rgb)} != {width}x{height}x3")
    stride = width * 3
    raw = bytearray((width * 3 + 1) * height)
    for y in range(height):
        raw[y * (stride + 1)] = 0  # filter type 0（None）
        raw[y * (stride + 1) + 1 : (y + 1) * (stride + 1)] = rgb[y * stride : (y + 1) * stride]
    compressed = zlib.compress(bytes(raw), 9)

    def _chunk(tag: bytes, data: bytes) -> bytes:
        out = struct.pack(">I", len(data)) + tag + data
        out += struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
        return out

    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # 8bit RGB
    return (
        b"\x89PNG\r\n\x1a\n"
        + _chunk(b"IHDR", ihdr)
        + _chunk(b"IDAT", compressed)
        + _chunk(b"IEND", b"")
    )


def render_footage_frame(
    footage_id: str,
    source_time: Rational,
    width: int,
    height: int,
    effects: list | None = None,
) -> PngBytes:
    """按素材 id 与源时间点渲染一帧（可叠加片段效果）。"""
    seed = _seed(footage_id)
    phase = int(round(source_time.to_float() * 30)) % 256
    data = bytearray(width * height * 3)
    idx = 0
    for y in range(height):
        for x in range(width):
            r = (x * 255 // width + phase + seed) % 256
            g = (y * 255 // height + (phase // 2) + (seed * 3) % 256) % 256
            b = ((x * 255 // width + y * 255 // height) + phase * 2 + (seed * 5) % 256) % 256
            data[idx] = r
            data[idx + 1] = g
            data[idx + 2] = b
            idx += 3
    if effects:
        _apply_effects(data, width, height, effects)
    return PngBytes(encode_png(width, height, bytes(data)))


def frame_is_black(png: bytes, width: int, height: int) -> bool:
    """解码 PNG（仅取像素）判断是否整幅全黑。非法数据抛 ValueError。"""
    rgb = decode_png(png)
    if len(rgb) != width * height * 3:
        raise ValueError("解码尺寸与预期不符")
    return max(rgb) <= _MIN_BRIGHTNESS


def decode_png(png: bytes) -> bytes:
    """测试用最小 PNG 解码器：校验签名/IHDR/CRC 并返回 RGB 平面数据。"""
    if not png.startswith(b"\x89PNG\r\n\x1a\n"):
        raise ValueError("非法 PNG 签名")
    pos = 8
    width = height = None
    idat = b""
    while pos < len(png):
        (length,) = struct.unpack(">I", png[pos : pos + 4])
        tag = png[pos + 4 : pos + 8]
        data = png[pos + 8 : pos + 8 + length]
        (crc,) = struct.unpack(">I", png[pos + 8 + length : pos + 12 + length])
        if crc != (zlib.crc32(tag + data) & 0xFFFFFFFF):
            raise ValueError(f"PNG CRC 校验失败（块 {tag!r}）")
        if tag == b"IHDR":
            width, height, bit_depth, color_type, _, _, _ = struct.unpack(">IIBBBBB", data)
            if bit_depth != 8 or color_type != 2:
                raise ValueError(f"仅支持 8bit RGB PNG，得到 bit={bit_depth} color={color_type}")
        elif tag == b"IDAT":
            idat += data
        pos += 12 + length
    if width is None or height is None:
        raise ValueError("PNG 缺少 IHDR")
    raw = zlib.decompress(idat)
    stride = 1 + width * 3
    if len(raw) != height * stride:
        raise ValueError("PNG 原始数据长度不符")
    out = bytearray(width * height * 3)
    for y in range(height):
        if raw[y * stride] != 0:
            raise ValueError("仅支持 filter 0 的 PNG")
        out[y * width * 3 : (y + 1) * width * 3] = raw[y * stride + 1 : (y + 1) * stride]
    return bytes(out)


def _apply_effects(data: bytearray, width: int, height: int, effects: list) -> None:
    """对 RGB 平面数据叠加片段效果（就地修改）。"""
    for eff in effects:
        params = eff.params
        etype = eff.effect_type
        spec = effect_spec(etype)
        if etype == "brightness":
            if params.get("enabled", True):
                factor = 1.0 + float(params.get("level", 0.0))
                _mul_channels(data, factor)
        elif etype == "color_shift":
            if params.get("enabled", True):
                shift = int(float(params.get("hue", 0.0)) / 360.0 * 255)
                _rotate_channels(data, shift)
        elif etype == "crop":
            left = float(params.get("left", 0.0))
            top = float(params.get("top", 0.0))
            right = float(params.get("right", 0.0))
            bottom = float(params.get("bottom", 0.0))
            _crop_edges(data, width, height, left, top, right, bottom)
        elif etype == "blur":
            if params.get("enabled", True) and float(params.get("radius", 0.0)) > 0:
                _box_blur(data, width, height, int(float(params.get("radius", 0.0))))
        elif etype == "overlay":
            mode = params.get("blend_mode", "normal")
            opacity = float(params.get("opacity", 0.5))
            _blend(data, mode, opacity)


def _clamp(v: int) -> int:
    return 0 if v < 0 else (255 if v > 255 else v)


def _mul_channels(data: bytearray, factor: float) -> None:
    for i in range(0, len(data), 3):
        data[i] = _clamp(round(data[i] * factor))
        data[i + 1] = _clamp(round(data[i + 1] * factor))
        data[i + 2] = _clamp(round(data[i + 2] * factor))


def _rotate_channels(data: bytearray, shift: int) -> None:
    for i in range(0, len(data), 3):
        r, g, b = data[i], data[i + 1], data[i + 2]
        data[i] = (r + shift) % 256
        data[i + 1] = (g + shift // 2) % 256
        data[i + 2] = (b + shift) % 256


def _crop_edges(data, width, height, left, top, right, bottom) -> None:
    x0 = int(left * width)
    y0 = int(top * height)
    x1 = int((1.0 - right) * width)
    y1 = int((1.0 - bottom) * height)
    for y in range(height):
        for x in range(width):
            if x < x0 or x >= x1 or y < y0 or y >= y1:
                i = (y * width + x) * 3
                data[i] = data[i + 1] = data[i + 2] = 0


def _box_blur(data: bytearray, width: int, height: int, radius: int) -> None:
    src = bytes(data)
    r = max(1, radius)
    for y in range(height):
        for x in range(width):
            total = [0, 0, 0]
            count = 0
            for dy in range(-r, r + 1):
                for dx in range(-r, r + 1):
                    ny, nx = y + dy, x + dx
                    if 0 <= ny < height and 0 <= nx < width:
                        i = (ny * width + nx) * 3
                        total[0] += src[i]
                        total[1] += src[i + 1]
                        total[2] += src[i + 2]
                        count += 1
            if count:
                i = (y * width + x) * 3
                data[i] = total[0] // count
                data[i + 1] = total[1] // count
                data[i + 2] = total[2] // count


def _blend(data: bytearray, mode: str, opacity: float) -> None:
    if mode == "normal" or opacity <= 0:
        return
    # 叠加色取固定中灰，screen/multiply 各取一个方向
    for i in range(0, len(data), 3):
        base = (data[i], data[i + 1], data[i + 2])
        if mode == "screen":
            out = (255 - (255 - base[0]) * (255 - 128) // 255,
                   255 - (255 - base[1]) * (255 - 128) // 255,
                   255 - (255 - base[2]) * (255 - 128) // 255)
        elif mode == "multiply":
            out = (base[0] * 128 // 255, base[1] * 128 // 255, base[2] * 128 // 255)
        else:
            out = base
        data[i] = _clamp(round(base[0] + (out[0] - base[0]) * opacity))
        data[i + 1] = _clamp(round(base[1] + (out[1] - base[1]) * opacity))
        data[i + 2] = _clamp(round(base[2] + (out[2] - base[2]) * opacity))