tests/test_static_rgb_audit.py

import json
from copy import deepcopy

import numpy as np
import pytest
from PIL import Image

from mcss.mechanism_pilot.data_prep import resize_rgb_depth
from mcss.mechanism_pilot.rgb_audit import (
    audit_rgb_chain,
    classify_black_frame,
    rgb_statistics,
)


def test_raw_statistics_preserved_before_preprocessing(tmp_path):
    raw = tmp_path / "raw/ai_003_001/images/scene_cam_00_final_preview"
    prepared = tmp_path / "prepared/ai_003_001"
    raw.mkdir(parents=True)
    prepared.mkdir(parents=True)
    frames = []
    for fid in (0, 1):
        image = np.zeros((7, 9, 3), dtype=np.uint8)
        if fid:
            image[:, :4] = [255, 130, 70]
        source = raw / f"frame.{fid:04d}.color.jpg"
        Image.fromarray(image).save(source)
        with Image.open(source) as im:
            original = np.array(im.convert("RGB"))
        resized, _ = resize_rgb_depth(original, np.ones((7, 9)), (3, 4))
        target = prepared / f"{fid:04d}.png"
        Image.fromarray(resized).save(target)
        frames.append({"frame_id": fid, "rgb": str(target)})
    manifest = tmp_path / "manifest.json"
    manifest.write_text(
        json.dumps({"image_size": [3, 4], "scenes": [{"scene_id": "ai_003_001", "frames": frames}]})
    )
    output = tmp_path / "audit"
    result = audit_rgb_chain(manifest, output)
    rows = json.loads((output / "frame_audit.json").read_text())
    assert result["ROOT_CAUSE"] == "SOURCE_FRAME_IS_BLACK"
    assert result["black_frames"] == [0]
    assert rows[1]["source_rgb"] == rgb_statistics(original)
    assert rows[1]["source_rgb"]["dimensions_hw"] == [7, 9]
    assert rows[1]["prepared_rgb"]["dimensions_hw"] == [3, 4]
    assert all(r["prepared_matches_recomputed_resize"] for r in rows)
    assert all(r["model_evaluator_tensors_equal"] for r in rows)
    assert len((output / "ai003_frame_audit.csv").read_text().splitlines()) == 11
    with pytest.raises(FileExistsError):
        audit_rgb_chain(manifest, output)


def test_stats_distinguish_black_pixels_from_zero_channels_and_invalid():
    image = np.array([[[0, 0, 0], [0, 255, 0], [1, 1, 1], [2, 2, 2]]], np.uint8)
    stats = rgb_statistics(image)
    assert stats["exact_zero_pixels_percent"] == 25
    assert stats["near_black_pixels_percent"] == 50
    assert stats["unique_rgb_values"] == 4
    invalid = rgb_statistics(np.array([[[np.nan, np.inf, 0]]]), scale=1)
    assert invalid["nan_values"] == invalid["inf_values"] == 1
    assert invalid["exact_zero_pixels_percent"] == 0


@pytest.mark.parametrize(
    "field,value,expected",
    [
        ("frame_mapping_valid", False, "FRAME_MAPPING_ERROR"),
        ("decoder_black_agreement", False, "DECODER_ERROR"),
        ("prepared_matches_recomputed_resize", False, "PREPROCESSING_ERROR"),
    ],
)
def test_quality_root_cause_independent_of_model_score(field, value, expected):
    row = {
        "frame_mapping_valid": True,
        "decoder_black_agreement": True,
        "prepared_matches_recomputed_resize": True,
        "source_rgb": {"max": 0},
        "prepared_rgb": {"max": 0},
        "model_input": {"min": 0, "max": 0, "nan_values": 0, "inf_values": 0},
    }
    assert classify_black_frame(row) == "SOURCE_FRAME_IS_BLACK"
    changed = deepcopy(row)
    changed[field] = value
    assert classify_black_frame(changed) == expected
    changed["model_score"] = 999
    assert classify_black_frame(changed) == expected