scripts/validate_dataset.py - master - alinaluc/Demetra

scripts/validate_dataset.py
676 строк19 KB
alinaluc
feat: add dataset validation foundation
26 июл 2026, 16:24
26 июл 2026, 16:24737af56
from __future__ import annotations

import argparse
import csv
import json
import logging
from collections import Counter
from collections.abc import Sequence
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import yaml
from PIL import Image


LOGGER = logging.getLogger(__name__)

SUPPORTED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
CSV_REPORT_NAME = "validation_files.csv"
JSON_REPORT_NAME = "validation_summary.json"

EXIT_SUCCESS = 0
EXIT_VALIDATION_FAILED = 1
EXIT_CONFIGURATION_ERROR = 2


class ConfigurationError(ValueError):
    """Raised when a configuration file has an invalid structure."""


@dataclass(frozen=True, slots=True)
class DatasetClass:
    index: int
    name: str
    name_ru: str


@dataclass(frozen=True, slots=True)
class FileValidationResult:
    path: str
    class_name: str
    width: int | None
    height: int | None
    format: str
    file_size: int | None
    status: str
    error: str


@dataclass(frozen=True, slots=True)
class ValidationOutcome:
    results: list[FileValidationResult]
    summary: dict[str, Any]
    csv_report_path: Path
    json_report_path: Path
    exit_code: int


def positive_integer(value: str) -> int:
    """Parse a strictly positive integer for argparse."""
    try:
        parsed_value = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"Expected an integer, received: {value}"
        ) from exc

    if parsed_value <= 0:
        raise argparse.ArgumentTypeError("Value must be greater than zero")

    return parsed_value


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Validate a grape-leaf image dataset without modifying source files."
        )
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        required=True,
        help="Dataset root containing one directory for each configured class.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("data/metadata/validation"),
        help=(
            "Directory for CSV and JSON reports "
            "(default: data/metadata/validation)."
        ),
    )
    parser.add_argument(
        "--min-width",
        type=positive_integer,
        default=224,
        help="Minimum acceptable image width in pixels (default: 224).",
    )
    parser.add_argument(
        "--min-height",
        type=positive_integer,
        default=224,
        help="Minimum acceptable image height in pixels (default: 224).",
    )
    parser.add_argument(
        "--classes-config",
        type=Path,
        default=Path("configs/classes.yaml"),
        help="Path to the class configuration (default: configs/classes.yaml).",
    )
    return parser.parse_args(argv)


def configure_logging() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(levelname)s: %(message)s",
    )


def load_classes(config_path: Path) -> list[DatasetClass]:
    """Load and validate the ordered class definitions from YAML."""
    try:
        raw_config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise ConfigurationError(
            f"Class configuration was not found: {config_path}"
        ) from exc
    except OSError as exc:
        raise ConfigurationError(
            f"Cannot read class configuration {config_path}: {exc}"
        ) from exc
    except yaml.YAMLError as exc:
        raise ConfigurationError(
            f"Invalid YAML in class configuration {config_path}: {exc}"
        ) from exc

    if not isinstance(raw_config, dict):
        raise ConfigurationError(
            "Class configuration root must be a YAML mapping"
        )

    raw_classes = raw_config.get("classes")
    if not isinstance(raw_classes, list) or not raw_classes:
        raise ConfigurationError(
            "Class configuration must contain a non-empty 'classes' list"
        )

    classes: list[DatasetClass] = []

    for position, raw_class in enumerate(raw_classes):
        if not isinstance(raw_class, dict):
            raise ConfigurationError(
                f"Class entry at position {position} must be a mapping"
            )

        index = raw_class.get("index")
        name = raw_class.get("name")
        name_ru = raw_class.get("name_ru")

        if type(index) is not int:
            raise ConfigurationError(
                f"Class index at position {position} must be an integer"
            )

        if not isinstance(name, str) or not name.strip():
            raise ConfigurationError(
                f"Class name at position {position} must be a non-empty string"
            )

        if not isinstance(name_ru, str) or not name_ru.strip():
            raise ConfigurationError(
                f"Russian class name at position {position} "
                "must be a non-empty string"
            )

        classes.append(
            DatasetClass(
                index=index,
                name=name.strip(),
                name_ru=name_ru.strip(),
            )
        )

    classes.sort(key=lambda item: item.index)

    expected_indices = list(range(len(classes)))
    actual_indices = [item.index for item in classes]

    if actual_indices != expected_indices:
        raise ConfigurationError(
            "Class indices must be unique and consecutive, "
            f"starting at zero. Received: {actual_indices}"
        )

    class_names = [item.name for item in classes]
    if len(set(class_names)) != len(class_names):
        raise ConfigurationError("Class names must be unique")

    return classes


def get_file_size(path: Path) -> tuple[int | None, str]:
    """Return a file size or an error message if stat fails."""
    try:
        return path.stat().st_size, ""
    except OSError as exc:
        return None, f"{type(exc).__name__}: {exc}"


def build_basic_result(
    *,
    path: Path,
    data_dir: Path,
    class_name: str,
    status: str,
    error: str,
) -> FileValidationResult:
    file_size, size_error = get_file_size(path)
    combined_error = error

    if size_error:
        combined_error = (
            f"{error}; {size_error}" if error else size_error
        )

    return FileValidationResult(
        path=path.relative_to(data_dir).as_posix(),
        class_name=class_name,
        width=None,
        height=None,
        format="",
        file_size=file_size,
        status=status,
        error=combined_error,
    )


def validate_image(
    *,
    path: Path,
    data_dir: Path,
    class_name: str,
    min_width: int,
    min_height: int,
) -> FileValidationResult:
    """Open and fully decode one image without changing it."""
    file_size, size_error = get_file_size(path)

    if size_error:
        return FileValidationResult(
            path=path.relative_to(data_dir).as_posix(),
            class_name=class_name,
            width=None,
            height=None,
            format="",
            file_size=None,
            status="unreadable_file",
            error=size_error,
        )

    try:
        with Image.open(path) as image:
            width, height = image.size
            image_format = image.format or ""
            image.verify()

        # Reopen the file because verify() invalidates the first image object.
        with Image.open(path) as image:
            image.load()
    except Exception as exc:
        return FileValidationResult(
            path=path.relative_to(data_dir).as_posix(),
            class_name=class_name,
            width=None,
            height=None,
            format="",
            file_size=file_size,
            status="corrupt_image",
            error=f"{type(exc).__name__}: {exc}",
        )

    if width < min_width or height < min_height:
        return FileValidationResult(
            path=path.relative_to(data_dir).as_posix(),
            class_name=class_name,
            width=width,
            height=height,
            format=image_format,
            file_size=file_size,
            status="too_small",
            error=(
                f"Image size {width}x{height} is below "
                f"the minimum {min_width}x{min_height}"
            ),
        )

    return FileValidationResult(
        path=path.relative_to(data_dir).as_posix(),
        class_name=class_name,
        width=width,
        height=height,
        format=image_format,
        file_size=file_size,
        status="valid",
        error="",
    )


def collect_dataset_files(data_dir: Path) -> list[Path]:
    """Collect regular files recursively in deterministic order."""
    try:
        files = [path for path in data_dir.rglob("*") if path.is_file()]
    except OSError as exc:
        raise OSError(f"Cannot scan dataset directory {data_dir}: {exc}") from exc

    return sorted(files, key=lambda path: path.as_posix().lower())


def detect_class_name(
    path: Path,
    data_dir: Path,
    class_names: set[str],
) -> str:
    """Infer the class from the first directory below the dataset root."""
    relative_path = path.relative_to(data_dir)

    if len(relative_path.parts) < 2:
        return ""

    first_directory = relative_path.parts[0]
    return first_directory if first_directory in class_names else ""


def scan_dataset(
    *,
    data_dir: Path,
    classes: list[DatasetClass],
    min_width: int,
    min_height: int,
) -> tuple[
    list[FileValidationResult],
    dict[str, dict[str, Any]],
    list[str],
]:
    """Validate the dataset files and return results and critical errors."""
    results: list[FileValidationResult] = []
    critical_errors: list[str] = []
    class_names = {item.name for item in classes}

    missing_class_directories = [
        item.name
        for item in classes
        if not (data_dir / item.name).is_dir()
    ]

    for class_name in missing_class_directories:
        critical_errors.append(
            f"Required class directory is missing: {class_name}"
        )

    files = collect_dataset_files(data_dir)

    for path in files:
        class_name = detect_class_name(path, data_dir, class_names)
        extension = path.suffix.lower()

        if extension not in SUPPORTED_IMAGE_EXTENSIONS:
            results.append(
                build_basic_result(
                    path=path,
                    data_dir=data_dir,
                    class_name=class_name,
                    status="unsupported_extension",
                    error=(
                        f"Unsupported extension: "
                        f"{extension or '[no extension]'}"
                    ),
                )
            )
            continue

        if not class_name:
            results.append(
                build_basic_result(
                    path=path,
                    data_dir=data_dir,
                    class_name="",
                    status="unexpected_location",
                    error=(
                        "Supported image is outside a configured "
                        "class directory"
                    ),
                )
            )
            continue

        results.append(
            validate_image(
                path=path,
                data_dir=data_dir,
                class_name=class_name,
                min_width=min_width,
                min_height=min_height,
            )
        )

    class_statistics: dict[str, dict[str, Any]] = {}

    for dataset_class in classes:
        class_results = [
            result
            for result in results
            if result.class_name == dataset_class.name
        ]
        status_counts = Counter(result.status for result in class_results)
        directory_exists = (data_dir / dataset_class.name).is_dir()
        valid_images = status_counts.get("valid", 0)

        class_statistics[dataset_class.name] = {
            "index": dataset_class.index,
            "name_ru": dataset_class.name_ru,
            "directory_exists": directory_exists,
            "total_files": len(class_results),
            "supported_images": (
                len(class_results)
                - status_counts.get("unsupported_extension", 0)
            ),
            "valid_images": valid_images,
            "too_small": status_counts.get("too_small", 0),
            "corrupt_images": status_counts.get("corrupt_image", 0),
            "unreadable_files": status_counts.get("unreadable_file", 0),
            "unsupported_files": status_counts.get(
                "unsupported_extension", 0
            ),
        }

        if directory_exists and valid_images == 0:
            critical_errors.append(
                f"Class '{dataset_class.name}' contains no valid images"
            )

    return results, class_statistics, critical_errors


def build_empty_class_statistics(
    classes: list[DatasetClass],
) -> dict[str, dict[str, Any]]:
    """Build statistics used when the dataset root is unavailable."""
    return {
        dataset_class.name: {
            "index": dataset_class.index,
            "name_ru": dataset_class.name_ru,
            "directory_exists": False,
            "total_files": 0,
            "supported_images": 0,
            "valid_images": 0,
            "too_small": 0,
            "corrupt_images": 0,
            "unreadable_files": 0,
            "unsupported_files": 0,
        }
        for dataset_class in classes
    }


def build_summary(
    *,
    data_dir: Path,
    classes_config: Path,
    output_dir: Path,
    min_width: int,
    min_height: int,
    results: list[FileValidationResult],
    class_statistics: dict[str, dict[str, Any]],
    critical_errors: list[str],
) -> dict[str, Any]:
    status_counts = Counter(result.status for result in results)
    warning_count = sum(
        count
        for status, count in status_counts.items()
        if status != "valid"
    )

    if critical_errors:
        validation_status = "failed"
    elif warning_count:
        validation_status = "passed_with_warnings"
    else:
        validation_status = "passed"

    return {
        "schema_version": 1,
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "status": validation_status,
        "paths": {
            "data_dir": str(data_dir),
            "classes_config": str(classes_config),
            "output_dir": str(output_dir),
            "csv_report": str(output_dir / CSV_REPORT_NAME),
            "json_report": str(output_dir / JSON_REPORT_NAME),
        },
        "minimum_image_size": {
            "width": min_width,
            "height": min_height,
        },
        "supported_extensions": sorted(SUPPORTED_IMAGE_EXTENSIONS),
        "totals": {
            "files_found": len(results),
            "valid_images": status_counts.get("valid", 0),
            "too_small": status_counts.get("too_small", 0),
            "corrupt_images": status_counts.get("corrupt_image", 0),
            "unreadable_files": status_counts.get("unreadable_file", 0),
            "unsupported_files": status_counts.get(
                "unsupported_extension", 0
            ),
            "unexpected_location": status_counts.get(
                "unexpected_location", 0
            ),
            "issues": warning_count,
        },
        "classes": class_statistics,
        "critical_errors": critical_errors,
    }


def write_csv_report(
    results: list[FileValidationResult],
    output_path: Path,
) -> None:
    fieldnames = [
        "path",
        "class_name",
        "width",
        "height",
        "format",
        "file_size",
        "status",
        "error",
    ]

    with output_path.open("w", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(asdict(result) for result in results)


def write_json_report(
    summary: dict[str, Any],
    output_path: Path,
) -> None:
    with output_path.open("w", encoding="utf-8") as file:
        json.dump(
            summary,
            file,
            ensure_ascii=False,
            indent=2,
        )
        file.write("\n")


def run_validation(
    *,
    data_dir: Path,
    output_dir: Path,
    classes_config: Path,
    min_width: int,
    min_height: int,
) -> ValidationOutcome:
    """Run validation and write both reports."""
    classes = load_classes(classes_config)
    results: list[FileValidationResult]
    class_statistics: dict[str, dict[str, Any]]
    critical_errors: list[str]

    if not data_dir.exists():
        results = []
        class_statistics = build_empty_class_statistics(classes)
        critical_errors = [
            f"Dataset directory does not exist: {data_dir}"
        ]
    elif not data_dir.is_dir():
        results = []
        class_statistics = build_empty_class_statistics(classes)
        critical_errors = [
            f"Dataset path is not a directory: {data_dir}"
        ]
    else:
        results, class_statistics, critical_errors = scan_dataset(
            data_dir=data_dir,
            classes=classes,
            min_width=min_width,
            min_height=min_height,
        )

    output_dir.mkdir(parents=True, exist_ok=True)

    csv_report_path = output_dir / CSV_REPORT_NAME
    json_report_path = output_dir / JSON_REPORT_NAME

    summary = build_summary(
        data_dir=data_dir,
        classes_config=classes_config,
        output_dir=output_dir,
        min_width=min_width,
        min_height=min_height,
        results=results,
        class_statistics=class_statistics,
        critical_errors=critical_errors,
    )

    write_csv_report(results, csv_report_path)
    write_json_report(summary, json_report_path)

    exit_code = (
        EXIT_VALIDATION_FAILED
        if critical_errors
        else EXIT_SUCCESS
    )

    return ValidationOutcome(
        results=results,
        summary=summary,
        csv_report_path=csv_report_path,
        json_report_path=json_report_path,
        exit_code=exit_code,
    )


def main(argv: Sequence[str] | None = None) -> int:
    configure_logging()
    args = parse_args(argv)

    LOGGER.info("Dataset directory: %s", args.data_dir)
    LOGGER.info("Classes configuration: %s", args.classes_config)
    LOGGER.info(
        "Minimum image size: %sx%s",
        args.min_width,
        args.min_height,
    )

    try:
        outcome = run_validation(
            data_dir=args.data_dir,
            output_dir=args.output_dir,
            classes_config=args.classes_config,
            min_width=args.min_width,
            min_height=args.min_height,
        )
    except ConfigurationError as exc:
        LOGGER.error("Configuration error: %s", exc)
        return EXIT_CONFIGURATION_ERROR
    except OSError as exc:
        LOGGER.error("File system error: %s", exc)
        return EXIT_CONFIGURATION_ERROR

    totals = outcome.summary["totals"]

    LOGGER.info("Files found: %s", totals["files_found"])
    LOGGER.info("Valid images: %s", totals["valid_images"])
    LOGGER.info("Issues found: %s", totals["issues"])
    LOGGER.info("CSV report: %s", outcome.csv_report_path)
    LOGGER.info("JSON report: %s", outcome.json_report_path)

    for error in outcome.summary["critical_errors"]:
        LOGGER.error("Critical error: %s", error)

    return outcome.exit_code


if __name__ == "__main__":
    raise SystemExit(main())