src/mcss/mechanism_pilot/rgb_audit.py
"""Read-only RGB provenance audit; no carrier predictions or frame exclusions."""
from __future__ import annotations
import csv
import hashlib
import json
from pathlib import Path
import numpy as np
import torch
from PIL import Image
from mcss.mechanism_pilot.data_prep import resize_rgb_depth
from mcss.mechanism_pilot.small_training import TrainScene
def rgb_statistics(array: np.ndarray, *, scale: float = 255.0) -> dict:
"""Record RGB pixel statistics before any preprocessing; near-black <= 1/255."""
x = np.asarray(array)
if x.ndim != 3 or x.shape[-1] != 3:
raise ValueError("Expected HWC RGB")
finite = np.isfinite(x)
values = x[finite].astype(np.float64)
pixel_finite = finite.all(-1)
return {
"dimensions_hw": list(x.shape[:2]),
"shape": list(x.shape),
"dtype": str(x.dtype),
"min": float(values.min()) if values.size else None,
"max": float(values.max()) if values.size else None,
"mean": float(values.mean()) if values.size else None,
"std": float(values.std()) if values.size else None,
"exact_zero_pixels_percent": float(((x == 0).all(-1) & pixel_finite).mean() * 100),
"near_black_pixels_percent": float(
((x >= 0).all(-1) & (x <= scale / 255).all(-1) & pixel_finite).mean() * 100
),
"near_black_definition": "all RGB channels in [0,1/255] of declared full range",
"unique_rgb_values": int(np.unique(x.reshape(-1, 3), axis=0).shape[0]),
"nan_values": int(np.isnan(x).sum()),
"inf_values": int(np.isinf(x).sum()),
"declared_full_range": scale,
}
def file_record(path: Path) -> dict:
return {
"path": str(path.resolve()),
"bytes": path.stat().st_size,
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
}
def classify_black_frame(row: dict) -> str:
"""Localize a black prepared frame without guessing the dataset's render cause."""
if not row["frame_mapping_valid"]:
return "FRAME_MAPPING_ERROR"
if row["decoder_black_agreement"] is False:
return "DECODER_ERROR"
tensor = row["model_input"]
if (
tensor["min"] is None
or tensor["min"] < 0
or tensor["max"] > 1
or tensor["nan_values"]
or tensor["inf_values"]
):
return "RANGE_ERROR"
if not row["prepared_matches_recomputed_resize"]:
return "PREPROCESSING_ERROR"
if row["source_rgb"]["max"] == 0 and row["prepared_rgb"]["max"] == 0:
return "SOURCE_FRAME_IS_BLACK"
return "UNKNOWN"
def _official_rows(directory: Path | None, scene: str, frame: int) -> dict:
if directory is None:
return {"status": "NOT_CHECKED"}
result = {}
for name in (
"metadata_images.csv",
"metadata_images_split_scene_v1.csv",
"metadata_camera_trajectories.csv",
):
path = directory / name
if not path.is_file():
result[name] = {"status": "MISSING"}
continue
with path.open() as handle:
records = list(csv.DictReader(handle))
if name == "metadata_camera_trajectories.csv":
matches = [
r
for r in records
if r.get("Animation") == f"{scene}_cam_00"
or (r.get("scene_name") == scene and r.get("camera_name") == "cam_00")
]
else:
matches = [
r
for r in records
if r.get("scene_name") == scene
and r.get("camera_name") == "cam_00"
and str(r.get("frame_id", "")).lstrip("0") == str(frame).lstrip("0")
]
result[name] = {"matching_rows": matches}
return result
def audit_rgb_chain(manifest_path, output_dir, official_metadata_dir=None) -> dict:
"""Audit existing source JPEGs and exact current preparation/loader outputs."""
manifest_path, output = Path(manifest_path).resolve(), Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
if (output / "frame_audit.json").exists():
raise FileExistsError("Keep previous audit; use a fresh output directory")
metadata = Path(official_metadata_dir) if official_metadata_dir else None
manifest = json.loads(manifest_path.read_text())
rows = []
for record in manifest["scenes"]:
scene = TrainScene(record, manifest["image_size"], manifest_path.parent, "cpu", [])
for frame in record["frames"]:
fid = frame["frame_id"]
source = (
manifest_path.parent
/ "raw"
/ record["scene_id"]
/ f"images/scene_cam_00_final_preview/frame.{fid:04d}.color.jpg"
)
prepared = scene.path(frame, "rgb")
with Image.open(source) as image:
image.load()
native = np.array(image)
rgb = np.array(image.convert("RGB"))
source_mode, source_format = image.mode, image.format
with Image.open(prepared) as image:
saved = np.array(image.convert("RGB"))
resized, _ = resize_rgb_depth(
rgb, np.ones(rgb.shape[:2], np.float32), tuple(manifest["image_size"])
)
tensor = scene.rgb(frame)
evaluator = scene.rgb(frame)
decoder = {"status": "UNAVAILABLE"}
try:
import cv2
other = cv2.imread(str(source), cv2.IMREAD_COLOR)
if other is None:
raise ValueError("Independent OpenCV decoder failed")
other = other[..., ::-1]
decoder = {
"status": "DECODED",
"rgb": rgb_statistics(other),
"exact_pixel_match": bool(np.array_equal(rgb, other)),
}
except ImportError:
pass
row = {
"scene_id": record["scene_id"],
"frame_id": fid,
"source_file": file_record(source),
"prepared_file": file_record(prepared),
"source_format": source_format,
"source_mode": source_mode,
"native_array_shape": list(native.shape),
"native_array_dtype": str(native.dtype),
"source_rgb": rgb_statistics(rgb),
"recomputed_resize_rgb": rgb_statistics(resized),
"prepared_rgb": rgb_statistics(saved),
"model_input": rgb_statistics(tensor.permute(1, 2, 0).numpy(), scale=1),
"evaluator_input": rgb_statistics(evaluator.permute(1, 2, 0).numpy(), scale=1),
"model_evaluator_tensors_equal": bool(torch.equal(tensor, evaluator)),
"prepared_matches_recomputed_resize": bool(np.array_equal(saved, resized)),
"independent_decoder": decoder,
"decoder_black_agreement": None
if decoder["status"] != "DECODED"
else bool((decoder["rgb"]["max"] == 0) == (rgb.max() == 0)),
"frame_mapping_valid": prepared.stem == f"{fid:04d}"
and prepared.parent.name == record["scene_id"],
"official_metadata": _official_rows(metadata, record["scene_id"], fid),
}
row["ROOT_CAUSE"] = classify_black_frame(row)
rows.append(row)
ai003 = [r for r in rows if r["scene_id"] == "ai_003_001"]
black = [r for r in ai003 if r["prepared_rgb"]["max"] == 0]
verdict = (
"SOURCE_FRAME_IS_BLACK"
if black and all(r["ROOT_CAUSE"] == "SOURCE_FRAME_IS_BLACK" for r in black)
else "UNKNOWN"
)
root_cause = {
"ROOT_CAUSE": verdict,
"black_frames": [r["frame_id"] for r in black],
"frames_audited": len(ai003),
"render_failure_status": "UNKNOWN_NOT_INFERRED_FROM_BLACK_PIXELS",
"frame_exclusion_applied": False,
"RGB_PREPROCESSING_CHANGED": False,
"scope": "Existing TRAIN media only, all frames retained; no model outputs used",
"source_is": "Published final_preview JPEG; not the linear HDR color buffer",
"official_metadata_directory": str(metadata) if metadata else None,
"near_black_rule": "descriptive only, all channels <= 1/255; not an exclusion rule",
"future_unseen_quality_rule": (
"official included; camera not BAD; finite positive depth >= .95; "
"reject all-black or >99.9% pixels all RGB channels <=2/255; no model scores"
),
"future_rule_applied_to_existing_train": False,
}
for name, data in (("frame_audit.json", rows), ("ai003_black_root_cause.json", root_cause)):
(output / name).write_text(json.dumps(data, indent=2, allow_nan=False) + "\n")
fields = [
"scene_id",
"frame_id",
"stage",
"path",
"file_size",
"sha256",
"dimensions_hw",
"dtype",
"min",
"max",
"mean",
"std",
"exact_zero_pixels_percent",
"near_black_pixels_percent",
"unique_rgb_values",
"nan_values",
"inf_values",
]
with (output / "ai003_frame_audit.csv").open("w", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=fields)
writer.writeheader()
for r in ai003:
for stage in (
"source_rgb",
"recomputed_resize_rgb",
"prepared_rgb",
"model_input",
"evaluator_input",
):
file = r["source_file" if stage == "source_rgb" else "prepared_file"]
writer.writerow(
{
"scene_id": r["scene_id"],
"frame_id": r["frame_id"],
"stage": stage,
"path": file["path"],
"file_size": file["bytes"],
"sha256": file["sha256"],
**{k: r[stage][k] for k in fields[6:]},
}
)
return root_cause