"""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))