path: root/scripts/analyze-p47.py

#!/usr/bin/env python3
"""P47 scene-detail analysis (leader rework).

Reads the `LV_P47_SHOT=1` day/night/closeup frames + <record-dir>/log.txt and writes
docs/evidence/p47-analysis.txt with:

  1. per-frame vertical reflection streaks: max run of consecutive tiles in one
     column (leader ①: length >= 3 tiles) and reflection-vs-background-water
     luma + warmth deltas (leader ① / ②: warm vertical column, not blue ripple);
  2. per-frame foreground roof lit-face vs shadow-face luma delta (leader ③:
     two light tiers from one unified light direction);
  3. element counts (bank reeds / floating props / reflection streaks);
  4. no big black block / no white block / opaque frames.

Usage:
    python3 scripts/analyze-p47.py <record-dir> <out-txt>
"""

import re
import sys
from pathlib import Path

import numpy as np
from PIL import Image

TILE = 32
MIN_BLOCK = 64
MIN_VERTICAL_RUN = 3
REFLECTION_DL_MIN = 8.0
REFLECTION_WARM_MIN = 20.0
ROOF_DELTA_MIN = 16.0
REFLECTION_RE = re.compile(
    r"p47-reflection (\S+) src=(\d+),(\d+) dst=(\d+),(\d+) ctrl=(-?\d+),(-?\d+) alpha=(\d+)")
ROOF_RE = re.compile(r"p47-roof (\S+) x=(\d+) y=(\d+) w=(\d+)")
ELEMENTS_RE = re.compile(r"p47-elements reeds=(\d+) floats=(\d+) reflections=(\d+) sources=(\d+) maxstreak=(\d+)")


def rgb(path):
    return np.asarray(Image.open(path).convert("RGB"), dtype=np.float64)


def luma(a):
    return 0.299 * a[:, :, 0] + 0.587 * a[:, :, 1] + 0.114 * a[:, :, 2]


def has_solid_block(mask, min_side):
    rows, cols = mask.shape
    if rows < min_side or cols < min_side:
        return None
    integral = np.zeros((rows + 1, cols + 1), dtype=np.int64)
    integral[1:, 1:] = mask.astype(np.int64).cumsum(0).cumsum(1)
    for y in range(0, rows - min_side + 1):
        for x in range(0, cols - min_side + 1):
            total = (integral[y + min_side, x + min_side] - integral[y, x + min_side]
                     - integral[y + min_side, x] + integral[y, x])
            if total == min_side * min_side:
                return (x, y)
    return None


def parse_log(log_text, frame):
    reflections, roofs = [], []
    for line in log_text.splitlines():
        m = REFLECTION_RE.search(line)
        if m and m.group(1) == frame:
            _, sx, sy, dx, dy, cx, cy, alpha = m.groups()
            reflections.append((int(sx), int(sy), int(dx), int(dy), int(cx), int(cy), int(alpha)))
            continue
        m = ROOF_RE.search(line)
        if m and m.group(1) == frame:
            _, x, y, w = m.groups()
            roofs.append((int(x), int(y), int(w)))
    return reflections, roofs


def max_vertical_run(reflections):
    cols = {}
    for (_, _, dx, dy, _, _, _) in reflections:
        cols.setdefault(dx, set()).add(dy)
    best = 0
    for ys in cols.values():
        ys = sorted(ys)
        cur = 1
        for i in range(1, len(ys)):
            cur = cur + 1 if ys[i] == ys[i - 1] + TILE else 1
            best = max(best, cur)
    return best


def reflection_metrics(img, reflections):
    """Reflection tiles vs a background-water baseline taken from the same rows,
    masking out +/-24px around every reflection column."""
    h, w, _ = img.shape
    if not reflections:
        return []
    rows = sorted({dy for (_, _, _, dy, _, _, _) in reflections})
    cols = sorted({dx for (_, _, dx, _, _, _, _) in reflections})
    mask = np.zeros((h, w), dtype=bool)
    for dx in cols:
        mask[:, max(0, dx - 24):min(w, dx + TILE + 24)] = True
    background = []
    for dy in rows:
        band = img[dy:dy + TILE]
        m = ~mask[dy:dy + TILE]
        if m.any():
            background.append(band[m].reshape(-1, 3))
    if not background:
        return []
    bg = np.concatenate(background, axis=0)
    bg_luma = (0.299 * bg[:, 0] + 0.587 * bg[:, 1] + 0.114 * bg[:, 2]).mean()
    bg_warm = (bg[:, 0] - bg[:, 2]).mean()
    out = []
    for (_, _, dx, dy, _, _, _) in reflections:
        if dx < 0 or dy < 0 or dx + TILE > w or dy + TILE > h:
            continue
        tile = img[dy:dy + TILE, dx:dx + TILE].reshape(-1, 3)
        tile_luma = (0.299 * tile[:, 0] + 0.587 * tile[:, 1] + 0.114 * tile[:, 2]).mean()
        out.append((tile_luma - bg_luma, (tile[:, 0] - tile[:, 2]).mean() - bg_warm))
    return out


def roof_metrics(img, roofs):
    h, w, _ = img.shape
    lit, shadow = [], []
    for (x, y, ww) in roofs:
        if x < 0 or x + ww > w or y - TILE < 0 or y + TILE > h or ww < 2:
            continue
        half = ww // 2
        lit.append(luma(img[y - TILE:y + TILE, x:x + half]).mean())
        shadow.append(luma(img[y - TILE:y + TILE, x + half:x + ww]).mean())
    return lit, shadow


def main(argv):
    if len(argv) != 3:
        sys.stderr.write(__doc__)
        return 2
    record_dir = Path(argv[1])
    out_txt = Path(argv[2])
    log_text = (record_dir / "log.txt").read_text(encoding="utf-8")
    frames = [n for n in ["p47-day", "p47-night", "p47-close"] if (record_dir / f"{n}.png").exists()]
    if len(frames) < 3:
        sys.stderr.write(f"expected p47-day/night/close, found {frames}\n")
        return 1

    failures = []
    lines = ["P47 scene-detail analysis (LV_P47_SHOT=1, leader rework)", ""]

    # 1) reflection streaks: vertical length + warm vertical column vs background water.
    lines.append("1) 倒影：垂直条纹长度 + 相对背景水面的 luma/暖色差")
    for name in frames:
        img = rgb(record_dir / f"{name}.png")
        reflections, _ = parse_log(log_text, name)
        run = max_vertical_run(reflections)
        metrics = reflection_metrics(img, reflections)
        if not metrics:
            failures.append(f"{name}: no visible reflection tile to measure")
            lines.append(f"   {name}: no visible reflection tile")
            continue
        dl = float(np.mean([m[0] for m in metrics]))
        warm = float(np.mean([m[1] for m in metrics]))
        lines.append(
            f"   {name}: tiles={len(reflections)} max_vertical_run={run} (required >= {MIN_VERTICAL_RUN}) "
            f"luma_delta={dl:+.2f} (required >= {REFLECTION_DL_MIN}) warm_rb_delta={warm:+.2f} (required >= {REFLECTION_WARM_MIN})")
        if run < MIN_VERTICAL_RUN:
            failures.append(f"{name} vertical reflection run {run} < {MIN_VERTICAL_RUN}")
        if dl < REFLECTION_DL_MIN:
            failures.append(f"{name} reflection luma delta {dl:.2f} < {REFLECTION_DL_MIN}")
        if warm < REFLECTION_WARM_MIN:
            failures.append(f"{name} reflection warm delta {warm:.2f} < {REFLECTION_WARM_MIN}")
    lines.append("")

    # 2) foreground roofs: one lit face + one shadow face (>= two luma tiers).
    lines.append("2) 近景屋顶受光面 / 背光面 luma 差")
    for name in frames:
        img = rgb(record_dir / f"{name}.png")
        _, roofs = parse_log(log_text, name)
        lit, shadow = roof_metrics(img, roofs)
        if not lit:
            failures.append(f"{name}: no fully-visible foreground roof to measure")
            lines.append(f"   {name}: no fully-visible roof")
            continue
        delta = float(np.mean(lit) - np.mean(shadow))
        lines.append(
            f"   {name}: roofs={len(lit)} lit={np.mean(lit):.2f} shadow={np.mean(shadow):.2f} "
            f"delta={delta:+.2f} (required >= {ROOF_DELTA_MIN})")
        if delta < ROOF_DELTA_MIN:
            failures.append(f"{name} roof lit/shadow delta {delta:.2f} < {ROOF_DELTA_MIN}")
    lines.append("")

    # 3) element-count metrics.
    lines.append("3) 元素数量指标")
    seen_counts = set()
    for line in log_text.splitlines():
        m = ELEMENTS_RE.search(line)
        if m:
            counts = tuple(int(m.group(i)) for i in range(1, 6))
            if counts in seen_counts:
                continue
            seen_counts.add(counts)
            lines.append(f"   reeds={counts[0]} floats={counts[1]} reflections={counts[2]} sources={counts[3]} maxstreak={counts[4]}")
    if not seen_counts:
        failures.append("no p47-elements line in log")
    else:
        reeds, floats, refl, sources, maxstreak = min(seen_counts)
        if min(reeds, floats, refl, sources) <= 0:
            failures.append("element counts must all be positive")
        if maxstreak < MIN_VERTICAL_RUN:
            failures.append(f"logged maxstreak {maxstreak} < {MIN_VERTICAL_RUN}")
    lines.append("")

    # 4) no big black block / no white block / opaque.
    lines.append("4) 无大黑块 / 无白块 / 不透明校验")
    for name in frames:
        a = np.asarray(Image.open(record_dir / f"{name}.png").convert("RGBA"), dtype=np.uint8)
        rgbv = a[:, :, :3]
        alpha_min = int(a[:, :, 3].min())
        black = (rgbv.max(axis=2) < 6)
        white = (rgbv.min(axis=2) >= 250)
        bb = has_solid_block(black, MIN_BLOCK)
        wb = has_solid_block(white, MIN_BLOCK)
        lines.append(f"   {name}: alpha_min={alpha_min} black_px={int(black.sum())} black_block={bb} white_block={wb}")
        if alpha_min < 250:
            failures.append(f"{name} has transparent pixels (alpha_min {alpha_min})")
        if bb is not None:
            failures.append(f"{name} has a {MIN_BLOCK}x{MIN_BLOCK} black block at {bb}")
        if wb is not None:
            failures.append(f"{name} has a {MIN_BLOCK}x{MIN_BLOCK} white block at {wb}")
    lines.append("")

    lines.append("verdict = " + ("PASS" if not failures else "FAIL"))
    out_txt.parent.mkdir(parents=True, exist_ok=True)
    out_txt.write_text("\n".join(lines) + "\n", encoding="utf-8")
    sys.stdout.write("\n".join(lines) + "\n")
    for f in failures:
        sys.stderr.write(f"FAIL {f}\n")
    return 0 if not failures else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))