mirror of
https://gitlvb.teallvbs.xyz/IQ.Lvbs/IQ.Pilot.git
synced 2026-07-24 21:12:05 +08:00
IQ.Lvbs CI [bot]
244 lines
7.3 KiB
Python
244 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from PIL import Image, ImageChops, ImageStat
|
|
except ModuleNotFoundError:
|
|
Image = None
|
|
ImageChops = None
|
|
ImageStat = None
|
|
|
|
|
|
DEFAULT_PLANNER_ROI = (0.22, 0.54, 0.78, 0.97)
|
|
|
|
|
|
def run(cmd: list[str]) -> str:
|
|
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
|
|
return result.stdout
|
|
|
|
|
|
def ffprobe_video(path: Path) -> dict:
|
|
payload = run([
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-print_format", "json",
|
|
"-show_streams",
|
|
"-show_format",
|
|
str(path),
|
|
])
|
|
return json.loads(payload)
|
|
|
|
|
|
def parse_fraction(value: str) -> float:
|
|
if "/" in value:
|
|
num, den = value.split("/", 1)
|
|
return float(num) / float(den)
|
|
return float(value)
|
|
|
|
|
|
def extract_frames(video_path: Path, output_dir: Path) -> list[Path]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
run([
|
|
"ffmpeg",
|
|
"-loglevel", "error",
|
|
"-i", str(video_path),
|
|
"-vsync", "0",
|
|
str(output_dir / "frame_%06d.png"),
|
|
"-y",
|
|
])
|
|
return sorted(output_dir.glob("frame_*.png"))
|
|
|
|
|
|
def file_hash(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def crop_box(width: int, height: int, roi: tuple[float, float, float, float]) -> tuple[int, int, int, int]:
|
|
left = int(width * roi[0])
|
|
top = int(height * roi[1])
|
|
right = int(width * roi[2])
|
|
bottom = int(height * roi[3])
|
|
return left, top, right, bottom
|
|
|
|
|
|
def rms_diff(prev_img, cur_img) -> float:
|
|
diff = ImageChops.difference(prev_img, cur_img)
|
|
return float(ImageStat.Stat(diff).rms[0])
|
|
|
|
|
|
def summarize(values: list[float]) -> dict[str, float]:
|
|
if not values:
|
|
return {"count": 0, "avg": 0.0, "max": 0.0, "min": 0.0}
|
|
return {
|
|
"count": len(values),
|
|
"avg": sum(values) / len(values),
|
|
"max": max(values),
|
|
"min": min(values),
|
|
}
|
|
|
|
|
|
def write_contact_sheet(flagged: list[Path], output_path: Path, *, columns: int = 3) -> None:
|
|
if Image is None or not flagged:
|
|
return
|
|
|
|
images = []
|
|
for path in flagged:
|
|
with Image.open(path) as img:
|
|
images.append(img.convert("RGB").copy())
|
|
|
|
thumb_w = min(img.width for img in images)
|
|
thumb_h = min(img.height for img in images)
|
|
rows = math.ceil(len(images) / columns)
|
|
sheet = Image.new("RGB", (thumb_w * columns, thumb_h * rows), color=(0, 0, 0))
|
|
|
|
for idx, img in enumerate(images):
|
|
thumb = img.resize((thumb_w, thumb_h))
|
|
x = (idx % columns) * thumb_w
|
|
y = (idx // columns) * thumb_h
|
|
sheet.paste(thumb, (x, y))
|
|
|
|
sheet.save(output_path)
|
|
|
|
|
|
def audit_video(video_path: Path, output_dir: Path, roi: tuple[float, float, float, float]) -> dict:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
temp_root = Path(tempfile.mkdtemp(prefix="iqpilot_video_audit_"))
|
|
try:
|
|
frames = extract_frames(video_path, temp_root / "frames")
|
|
probe = ffprobe_video(video_path)
|
|
streams = probe.get("streams", [])
|
|
video_stream = next((stream for stream in streams if stream.get("codec_type") == "video"), {})
|
|
fps = parse_fraction(video_stream.get("avg_frame_rate", "0")) if video_stream.get("avg_frame_rate") else 0.0
|
|
duration = float(probe.get("format", {}).get("duration", 0.0) or 0.0)
|
|
|
|
records: list[dict] = []
|
|
duplicate_runs: list[int] = []
|
|
current_duplicate_run = 0
|
|
exact_duplicate_indices: list[int] = []
|
|
low_motion_indices: list[int] = []
|
|
suspicious_paths: list[Path] = []
|
|
full_rms_values: list[float] = []
|
|
planner_rms_values: list[float] = []
|
|
|
|
prev_hash = None
|
|
prev_img = None
|
|
prev_planner = None
|
|
|
|
for idx, frame_path in enumerate(frames):
|
|
frame_hash = file_hash(frame_path)
|
|
exact_duplicate = prev_hash == frame_hash
|
|
|
|
full_rms = 0.0
|
|
planner_rms = 0.0
|
|
if Image is not None:
|
|
with Image.open(frame_path) as img:
|
|
current = img.convert("L")
|
|
planner = current.crop(crop_box(current.width, current.height, roi))
|
|
if prev_img is not None:
|
|
full_rms = rms_diff(prev_img, current)
|
|
planner_rms = rms_diff(prev_planner, planner)
|
|
full_rms_values.append(full_rms)
|
|
planner_rms_values.append(planner_rms)
|
|
prev_img = current.copy()
|
|
prev_planner = planner.copy()
|
|
|
|
if exact_duplicate:
|
|
current_duplicate_run += 1
|
|
exact_duplicate_indices.append(idx)
|
|
elif current_duplicate_run:
|
|
duplicate_runs.append(current_duplicate_run)
|
|
current_duplicate_run = 0
|
|
|
|
if idx > 0 and (exact_duplicate or planner_rms < 1.2 or full_rms < 1.0):
|
|
low_motion_indices.append(idx)
|
|
suspicious_paths.append(frame_path)
|
|
|
|
records.append({
|
|
"frame": idx,
|
|
"exact_duplicate": exact_duplicate,
|
|
"full_rms": round(full_rms, 4),
|
|
"planner_rms": round(planner_rms, 4),
|
|
})
|
|
prev_hash = frame_hash
|
|
|
|
if current_duplicate_run:
|
|
duplicate_runs.append(current_duplicate_run)
|
|
|
|
csv_path = output_dir / f"{video_path.stem}_frame_metrics.csv"
|
|
with csv_path.open("w", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=["frame", "exact_duplicate", "full_rms", "planner_rms"])
|
|
writer.writeheader()
|
|
writer.writerows(records)
|
|
|
|
flagged_dir = output_dir / f"{video_path.stem}_flagged_frames"
|
|
flagged_dir.mkdir(parents=True, exist_ok=True)
|
|
for path in suspicious_paths[:24]:
|
|
shutil.copy2(path, flagged_dir / path.name)
|
|
|
|
write_contact_sheet(sorted(flagged_dir.glob("*.png"))[:12], output_dir / f"{video_path.stem}_contact_sheet.png")
|
|
|
|
summary = {
|
|
"video": str(video_path),
|
|
"frame_count": len(frames),
|
|
"fps": fps,
|
|
"duration_s": duration,
|
|
"exact_duplicate_frames": len(exact_duplicate_indices),
|
|
"max_duplicate_run": max(duplicate_runs) if duplicate_runs else 0,
|
|
"duplicate_runs": duplicate_runs,
|
|
"low_motion_frames": len(low_motion_indices),
|
|
"full_rms": summarize(full_rms_values),
|
|
"planner_rms": summarize(planner_rms_values),
|
|
"planner_roi": {
|
|
"left": roi[0],
|
|
"top": roi[1],
|
|
"right": roi[2],
|
|
"bottom": roi[3],
|
|
},
|
|
"artifacts": {
|
|
"metrics_csv": str(csv_path),
|
|
"flagged_frames_dir": str(flagged_dir),
|
|
"contact_sheet": str(output_dir / f"{video_path.stem}_contact_sheet.png"),
|
|
},
|
|
}
|
|
|
|
summary_path = output_dir / f"{video_path.stem}_summary.json"
|
|
summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
|
|
return summary
|
|
finally:
|
|
shutil.rmtree(temp_root, ignore_errors=True)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Audit an MP4 for duplicate/low-motion frame runs.")
|
|
parser.add_argument("video", type=Path, help="Input MP4")
|
|
parser.add_argument("--output-dir", type=Path, default=Path("video_audit"), help="Artifact output directory")
|
|
parser.add_argument(
|
|
"--planner-roi",
|
|
default="0.22,0.54,0.78,0.97",
|
|
help="ROI fractions left,top,right,bottom for planner-focused RMS stats",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
roi = tuple(float(part.strip()) for part in args.planner_roi.split(","))
|
|
if len(roi) != 4:
|
|
raise ValueError("planner-roi must have four comma-separated floats")
|
|
|
|
summary = audit_video(args.video, args.output_dir, roi)
|
|
print(json.dumps(summary, indent=2, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|