tools/audit_exercise_frames.py - develop - DemienMedich/BodyweightBase

tools/audit_exercise_frames.py
384 строки15 KB
DemienMedich
Harden package verification and clean startup
18 июл 2026, 21:15
18 июл 2026, 21:1535f16e9
from __future__ import annotations

import argparse
import json
import re
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any

from PIL import Image, UnidentifiedImageError


REPO_ROOT = Path(__file__).resolve().parents[1]
FRAME_NAME = re.compile(r"^(?P<exercise_id>.+)_(?P<index>[1-9]\d*)$")
ALLOWED_PLAYBACK = {"loop", "pingPong", "steps", "static"}


def runtime_id_key(exercise_id: str) -> str:
    return exercise_id.lower()


def image_files(
    directory: Path, suffix: str, kind: str, errors: list[str]
) -> list[Path]:
    if not directory.is_dir():
        return []
    paths: list[Path] = []
    for path in directory.iterdir():
        if not path.is_file():
            continue
        if path.suffix == suffix:
            paths.append(path)
        elif path.suffix.lower() == suffix:
            errors.append(
                f"invalid {kind} extension case: {path.name} (expected lowercase {suffix})"
            )
    return sorted(paths, key=lambda path: path.name.lower())


def load_json(path: Path, label: str, errors: list[str]) -> Any | None:
    try:
        with path.open("r", encoding="utf-8-sig") as stream:
            return json.load(stream)
    except FileNotFoundError:
        errors.append(f"{label} not found: {path}")
    except (OSError, json.JSONDecodeError) as error:
        errors.append(f"cannot read {label} {path}: {error}")
    return None


def load_catalog(path: Path, errors: list[str]) -> tuple[list[dict[str, Any]], set[str]]:
    document = load_json(path, "catalog", errors)
    if document is None:
        return [], set()
    if not isinstance(document, dict) or not isinstance(document.get("exercises"), list):
        errors.append("catalog root must contain an 'exercises' array")
        return [], set()

    exercises: list[dict[str, Any]] = []
    exercise_ids: set[str] = set()
    exercise_ids_by_key: dict[str, str] = {}
    for position, value in enumerate(document["exercises"], start=1):
        if not isinstance(value, dict):
            errors.append(f"catalog exercise #{position} is not an object")
            continue
        exercise_id = value.get("id")
        if not isinstance(exercise_id, str) or not exercise_id.strip():
            errors.append(f"catalog exercise #{position} has no valid id")
            continue
        exercise_id_key = runtime_id_key(exercise_id)
        if exercise_id_key in exercise_ids_by_key:
            errors.append(
                f"duplicate catalog exercise id (case-insensitive): {exercise_id} "
                f"conflicts with {exercise_ids_by_key[exercise_id_key]}"
            )
            continue
        exercise_ids_by_key[exercise_id_key] = exercise_id
        exercise_ids.add(exercise_id)
        exercises.append(value)

    if not exercises:
        errors.append("catalog contains no usable exercises")

    return exercises, exercise_ids


def index_frames(
    paths: list[Path], exercise_ids: set[str], kind: str, errors: list[str]
) -> dict[str, list[int]]:
    indexed: dict[str, list[int]] = defaultdict(list)
    for path in paths:
        match = FRAME_NAME.fullmatch(path.stem)
        if match is None:
            errors.append(f"invalid {kind} frame name: {path.name} (expected <id>_<n>{path.suffix.lower()})")
            continue
        exercise_id = match.group("exercise_id")
        if exercise_id not in exercise_ids:
            errors.append(f"unknown exercise id in {kind} frame: {path.name}")
            continue
        indexed[exercise_id].append(int(match.group("index")))
    return indexed


def validate_png_sequences(
    exercise_ids: set[str], frames_by_id: dict[str, list[int]], max_frames: int, errors: list[str]
) -> None:
    for exercise_id in sorted(exercise_ids):
        indices = sorted(frames_by_id.get(exercise_id, []))
        if not indices:
            errors.append(f"exercise has no PNG frames: {exercise_id}")
            continue
        expected = list(range(1, len(indices) + 1))
        if indices != expected:
            errors.append(
                f"non-contiguous PNG frames for {exercise_id}: got {indices}, expected {expected}"
            )
        if len(indices) > max_frames:
            errors.append(
                f"too many PNG frames for {exercise_id}: {len(indices)}, runtime limit is {max_frames}"
            )


def validate_parity(png_paths: list[Path], jpg_paths: list[Path], errors: list[str]) -> None:
    png_stems = {path.stem for path in png_paths}
    jpg_stems = {path.stem for path in jpg_paths}
    for stem in sorted(png_stems - jpg_stems):
        errors.append(f"mobile JPG missing for PNG: {stem}.png")
    for stem in sorted(jpg_stems - png_stems):
        errors.append(f"PNG missing for mobile JPG: {stem}.jpg")


def validate_images(
    paths: list[Path], min_size: int, min_aspect: float, max_aspect: float, errors: list[str]
) -> None:
    for path in paths:
        try:
            with Image.open(path) as image:
                width, height = image.size
                image.verify()
        except (OSError, ValueError, UnidentifiedImageError) as error:
            errors.append(f"cannot open image {path}: {error}")
            continue

        if width < min_size or height < min_size:
            errors.append(f"image too small: {path} is {width}x{height}, minimum is {min_size}x{min_size}")
        aspect = width / height if height else 0.0
        if not min_aspect <= aspect <= max_aspect:
            errors.append(
                f"unreasonable aspect ratio: {path} is {width}x{height} ({aspect:.3f}), "
                f"expected {min_aspect:g}..{max_aspect:g}"
            )


def validate_animation_metadata(
    exercises: list[dict[str, Any]], frames_by_id: dict[str, list[int]], errors: list[str]
) -> None:
    for exercise in exercises:
        exercise_id = exercise["id"]
        if "framePlayback" in exercise:
            playback = exercise["framePlayback"]
            if not isinstance(playback, str):
                errors.append(f"framePlayback for {exercise_id} must be a string")
            elif playback not in ALLOWED_PLAYBACK:
                errors.append(
                    f"invalid framePlayback for {exercise_id}: {playback!r}; "
                    f"allowed: {', '.join(sorted(ALLOWED_PLAYBACK))}"
                )

        if "frameDurationsMs" not in exercise:
            continue
        durations = exercise["frameDurationsMs"]
        if not isinstance(durations, list):
            errors.append(f"frameDurationsMs for {exercise_id} must be an array")
            continue
        for position, duration in enumerate(durations, start=1):
            if not isinstance(duration, int) or isinstance(duration, bool):
                errors.append(
                    f"frameDurationsMs[{position}] for {exercise_id} must be an integer"
                )
            elif not 350 <= duration <= 10000:
                errors.append(
                    f"frameDurationsMs[{position}] for {exercise_id} is {duration}; expected 350..10000"
                )
        frame_count = len(frames_by_id.get(exercise_id, []))
        if len(durations) != frame_count:
            errors.append(
                f"frameDurationsMs length mismatch for {exercise_id}: "
                f"{len(durations)} durations, {frame_count} PNG frames"
            )


def validate_state_plans(
    state_path: Path, exercise_ids: set[str], errors: list[str]
) -> tuple[int, int]:
    document = load_json(state_path, "default state", errors)
    if document is None:
        return 0, 0
    if not isinstance(document, dict) or not isinstance(document.get("plans"), list):
        errors.append("default state root must contain a 'plans' array")
        return 0, 0

    plan_count = 0
    step_count = 0
    seen_plan_ids: set[str] = set()
    for position, value in enumerate(document["plans"], start=1):
        if not isinstance(value, dict):
            errors.append(f"default state plan #{position} is not an object")
            continue
        plan_count += 1
        plan_id = value.get("id")
        if not isinstance(plan_id, str) or not plan_id:
            errors.append(f"default state plan #{position} has no valid id")
            plan_id = f"#{position}"
        elif plan_id in seen_plan_ids:
            errors.append(f"duplicate default state plan id: {plan_id}")
        else:
            seen_plan_ids.add(plan_id)
        steps = value.get("steps")
        if not isinstance(steps, list):
            errors.append(f"default state plan {plan_id} has no steps array")
            continue
        for step_position, step in enumerate(steps, start=1):
            step_count += 1
            if not isinstance(step, dict):
                errors.append(f"default state plan {plan_id} step #{step_position} is not an object")
                continue
            exercise_id = step.get("exerciseId")
            if not isinstance(exercise_id, str) or exercise_id not in exercise_ids:
                errors.append(
                    f"unknown exerciseId in default state plan {plan_id} step #{step_position}: "
                    f"{exercise_id!r}"
                )
    if plan_count == 0:
        errors.append("default state contains no usable plans")
    return plan_count, step_count


def manifest_entries(document: Any, errors: list[str]) -> list[dict[str, Any]]:
    values = document.get("exercises") if isinstance(document, dict) else document
    if not isinstance(values, list):
        errors.append("manifest must be an array or contain an 'exercises' array")
        return []
    entries: list[dict[str, Any]] = []
    for position, value in enumerate(values, start=1):
        if not isinstance(value, dict):
            errors.append(f"manifest entry #{position} is not an object")
            continue
        entries.append(value)
    return entries


def validate_manifest(
    manifest_path: Path,
    sheets_dir: Path,
    exercise_ids: set[str],
    frames_by_id: dict[str, list[int]],
    errors: list[str],
) -> int:
    document = load_json(manifest_path, "manifest", errors)
    if document is None:
        return 0

    entries = manifest_entries(document, errors)
    seen_ids: set[str] = set()
    seen_ids_by_key: dict[str, str] = {}
    for position, entry in enumerate(entries, start=1):
        exercise_id = entry.get("id")
        if not isinstance(exercise_id, str) or not exercise_id:
            errors.append(f"manifest entry #{position} has no valid id")
            continue
        exercise_id_key = runtime_id_key(exercise_id)
        if exercise_id_key in seen_ids_by_key:
            errors.append(
                f"duplicate manifest exercise id (case-insensitive): {exercise_id} "
                f"conflicts with {seen_ids_by_key[exercise_id_key]}"
            )
        else:
            seen_ids_by_key[exercise_id_key] = exercise_id
        seen_ids.add(exercise_id)
        if exercise_id not in exercise_ids:
            errors.append(f"unknown exercise id in manifest: {exercise_id}")

        declared_frames = entry.get("frames")
        actual_frames = len(frames_by_id.get(exercise_id, []))
        if not isinstance(declared_frames, int) or isinstance(declared_frames, bool):
            errors.append(f"manifest frames for {exercise_id} must be an integer")
        elif declared_frames != actual_frames:
            errors.append(
                f"manifest frame count mismatch for {exercise_id}: "
                f"declared {declared_frames}, found {actual_frames} PNG frames"
            )

        source = next(
            (entry.get(key) for key in ("source", "sheet", "sheetFile") if entry.get(key)),
            None,
        )
        if source is not None:
            if not isinstance(source, str):
                errors.append(f"manifest sheet for {exercise_id} must be a path string")
            else:
                sheet_path = Path(source)
                if not sheet_path.is_absolute():
                    sheet_path = sheets_dir / sheet_path
                if not sheet_path.is_file():
                    errors.append(f"manifest sheet missing for {exercise_id}: {sheet_path}")

    for exercise_id in sorted(exercise_ids - seen_ids):
        errors.append(f"manifest entry missing for catalog exercise: {exercise_id}")

    return len(entries)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Read-only audit of exercise animation assets.")
    parser.add_argument("--catalog", type=Path, default=REPO_ROOT / "resources" / "exercises.json")
    parser.add_argument("--state", type=Path, default=REPO_ROOT / "resources" / "default-state.json")
    parser.add_argument("--frames", type=Path, default=REPO_ROOT / "Assets" / "ExerciseFrames")
    parser.add_argument(
        "--mobile-frames", type=Path, default=REPO_ROOT / "Assets" / "ExerciseFramesMobile"
    )
    parser.add_argument(
        "--manifest",
        type=Path,
        default=REPO_ROOT / "Assets" / "GeneratedSheets" / "exercise_generation_manifest.json",
        help="Generation manifest; it must cover the catalog exactly.",
    )
    parser.add_argument(
        "--sheets", type=Path, default=REPO_ROOT / "Assets" / "GeneratedSheets"
    )
    parser.add_argument("--min-size", type=int, default=256)
    parser.add_argument("--min-aspect", type=float, default=0.45)
    parser.add_argument("--max-aspect", type=float, default=2.2)
    parser.add_argument("--max-frames", type=int, default=32)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    errors: list[str] = []

    if args.min_size < 1:
        errors.append("--min-size must be positive")
    if args.min_aspect <= 0 or args.max_aspect < args.min_aspect:
        errors.append("aspect bounds must be positive and ordered")
    if args.max_frames < 1:
        errors.append("--max-frames must be positive")

    exercises, exercise_ids = load_catalog(args.catalog, errors)
    if not args.frames.is_dir():
        errors.append(f"PNG frame directory not found: {args.frames}")
    if not args.mobile_frames.is_dir():
        errors.append(f"mobile JPG frame directory not found: {args.mobile_frames}")

    png_paths = image_files(args.frames, ".png", "PNG", errors)
    jpg_paths = image_files(args.mobile_frames, ".jpg", "mobile JPG", errors)
    png_by_id = index_frames(png_paths, exercise_ids, "PNG", errors)
    index_frames(jpg_paths, exercise_ids, "mobile JPG", errors)

    validate_png_sequences(exercise_ids, png_by_id, args.max_frames, errors)
    validate_parity(png_paths, jpg_paths, errors)
    validate_images(
        png_paths + jpg_paths, args.min_size, args.min_aspect, args.max_aspect, errors
    )
    validate_animation_metadata(exercises, png_by_id, errors)
    manifest_count = validate_manifest(
        args.manifest, args.sheets, exercise_ids, png_by_id, errors
    )
    plan_count, step_count = validate_state_plans(args.state, exercise_ids, errors)

    manifest_status = str(manifest_count) if args.manifest.exists() else "missing"
    print(
        f"exercise-frame audit: exercises={len(exercise_ids)} png={len(png_paths)} "
        f"mobile_jpg={len(jpg_paths)} plans={plan_count} steps={step_count} "
        f"manifest={manifest_status} errors={len(errors)}"
    )
    for error in errors:
        print(f"ERROR: {error}")
    print("OK" if not errors else "FAILED")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())