Spaces:
Sleeping
Sleeping
| """Reproducibly evaluate the saved Mission17 model on a held-out dataset.""" | |
| import argparse | |
| import hashlib | |
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| import seaborn as sns | |
| import tensorflow as tf | |
| from PIL import Image | |
| from sklearn.metrics import ( | |
| accuracy_score, | |
| confusion_matrix, | |
| precision_recall_fscore_support, | |
| precision_score, | |
| recall_score, | |
| f1_score, | |
| ) | |
| from tensorflow.keras.applications.efficientnet import preprocess_input | |
| AI_ROOT = Path(__file__).resolve().parents[2] | |
| DEFAULT_DATASET = AI_ROOT.parent / 'dataset' / 'mission_dataset_split' / 'test' | |
| DEFAULT_TRAINING_DATASET = AI_ROOT.parent / 'dataset' / 'mission_dataset_split' / 'train' | |
| SUPPORTED_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp'} | |
| def sha256_file(path): | |
| digest = hashlib.sha256() | |
| with path.open('rb') as source: | |
| for chunk in iter(lambda: source.read(1024 * 1024), b''): | |
| digest.update(chunk) | |
| return digest.hexdigest() | |
| def dataset_manifest_hash(dataset_path): | |
| digest = hashlib.sha256() | |
| files = sorted(path for path in dataset_path.rglob('*') if path.is_file()) | |
| for path in files: | |
| relative = path.relative_to(dataset_path).as_posix() | |
| digest.update(f'{relative}\0{path.stat().st_size}\n'.encode('utf-8')) | |
| return digest.hexdigest(), len(files) | |
| def file_hashes(dataset_path): | |
| hashes = {} | |
| for path in sorted(candidate for candidate in dataset_path.rglob('*') if candidate.is_file()): | |
| hashes[path] = sha256_file(path) | |
| return hashes | |
| def load_labels(path): | |
| labels = [line.strip() for line in path.read_text(encoding='utf-8').splitlines() if line.strip()] | |
| if not labels: | |
| raise ValueError('labels.txt is empty.') | |
| return labels | |
| def discover_samples(dataset_path, labels): | |
| unexpected_directories = sorted( | |
| path.name for path in dataset_path.iterdir() | |
| if path.is_dir() and path.name not in labels | |
| ) | |
| if unexpected_directories: | |
| raise ValueError(f'Unexpected class directories: {unexpected_directories}') | |
| samples = [] | |
| distribution = {} | |
| for class_index, label in enumerate(labels): | |
| class_path = dataset_path / label | |
| if not class_path.is_dir(): | |
| raise FileNotFoundError(f'Missing class directory: {class_path}') | |
| class_files = sorted( | |
| path for path in class_path.rglob('*') | |
| if path.is_file() and path.suffix.lower() in SUPPORTED_EXTENSIONS | |
| ) | |
| distribution[label] = len(class_files) | |
| samples.extend((path, class_index) for path in class_files) | |
| return samples, distribution | |
| def predict_samples(model, samples, batch_size): | |
| predictions = [] | |
| actual = [] | |
| for start in range(0, len(samples), batch_size): | |
| batch_records = samples[start:start + batch_size] | |
| batch_images = [] | |
| for image_path, class_index in batch_records: | |
| try: | |
| with Image.open(image_path) as image: | |
| image = image.convert('RGB').resize((224, 224), Image.Resampling.LANCZOS) | |
| batch_images.append(np.asarray(image, dtype=np.float32)) | |
| except Exception as error: | |
| raise ValueError(f'Could not decode test image: {image_path}') from error | |
| actual.append(class_index) | |
| batch = preprocess_input(np.stack(batch_images, axis=0)) | |
| probabilities = model.predict(batch, verbose=0) | |
| predictions.extend(np.argmax(probabilities, axis=1).tolist()) | |
| return np.asarray(actual), np.asarray(predictions) | |
| def parse_args(): | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument('--dataset', type=Path, default=DEFAULT_DATASET) | |
| parser.add_argument('--training-dataset', type=Path, default=DEFAULT_TRAINING_DATASET) | |
| parser.add_argument('--dataset-version', required=True, help='Human-readable immutable dataset version.') | |
| parser.add_argument('--model', type=Path, default=AI_ROOT / 'mission_model.h5') | |
| parser.add_argument('--labels', type=Path, default=AI_ROOT / 'labels.txt') | |
| parser.add_argument('--output-dir', type=Path, default=AI_ROOT / 'outputs') | |
| parser.add_argument('--batch-size', type=int, default=32) | |
| return parser.parse_args() | |
| def main(): | |
| args = parse_args() | |
| dataset_path = args.dataset.resolve() | |
| if not dataset_path.is_dir(): | |
| raise FileNotFoundError(f'Held-out dataset not found: {dataset_path}') | |
| if args.batch_size < 1: | |
| raise ValueError('Batch size must be positive.') | |
| labels = load_labels(args.labels) | |
| model = tf.keras.models.load_model(args.model) | |
| model_output_count = int(model.output_shape[-1]) | |
| if model_output_count != len(labels): | |
| raise ValueError( | |
| f'Model output count ({model_output_count}) does not match label count ({len(labels)}).' | |
| ) | |
| samples, class_distribution = discover_samples(dataset_path, labels) | |
| if not samples: | |
| raise ValueError('Held-out dataset contains no supported images.') | |
| actual, predicted = predict_samples(model, samples, args.batch_size) | |
| matrix = confusion_matrix(actual, predicted, labels=list(range(len(labels)))) | |
| per_precision, per_recall, per_f1, per_support = precision_recall_fscore_support( | |
| actual, | |
| predicted, | |
| labels=list(range(len(labels))), | |
| zero_division=0, | |
| ) | |
| manifest_hash, manifest_files = dataset_manifest_hash(dataset_path) | |
| training_path = args.training_dataset.resolve() | |
| if not training_path.is_dir(): | |
| raise FileNotFoundError(f'Training dataset not found: {training_path}') | |
| training_hashes = set(file_hashes(training_path).values()) | |
| evaluation_hashes = file_hashes(dataset_path) | |
| exact_overlap_count = sum( | |
| 1 for content_hash in evaluation_hashes.values() if content_hash in training_hashes | |
| ) | |
| per_class = { | |
| label: { | |
| 'precision': float(per_precision[index]), | |
| 'recall': float(per_recall[index]), | |
| 'f1': float(per_f1[index]), | |
| 'support': int(per_support[index]), | |
| } | |
| for index, label in enumerate(labels) | |
| } | |
| report = { | |
| 'evaluatedAtUtc': datetime.now(timezone.utc).isoformat(), | |
| 'dataset': { | |
| 'version': args.dataset_version, | |
| 'path': str(dataset_path), | |
| 'manifestSha256': manifest_hash, | |
| 'manifestFileCount': manifest_files, | |
| 'evaluatedSampleCount': len(samples), | |
| 'excludedFileCount': manifest_files - len(samples), | |
| 'classDistribution': class_distribution, | |
| 'leakageAudit': { | |
| 'trainingPath': str(training_path), | |
| 'exactCrossSplitDuplicateCount': exact_overlap_count, | |
| 'validForPublication': exact_overlap_count == 0, | |
| 'limitation': ( | |
| 'Exact duplicate audit passed; near-duplicate and source-provenance review are still required.' | |
| if exact_overlap_count == 0 | |
| else 'Evaluation is invalid for publication because test files also occur in training data.' | |
| ), | |
| }, | |
| }, | |
| 'model': { | |
| 'file': args.model.name, | |
| 'sha256': sha256_file(args.model), | |
| 'outputCount': model_output_count, | |
| 'labelOrder': labels, | |
| }, | |
| 'metrics': { | |
| 'accuracy': float(accuracy_score(actual, predicted)), | |
| 'weightedPrecision': float(precision_score(actual, predicted, average='weighted', zero_division=0)), | |
| 'weightedRecall': float(recall_score(actual, predicted, average='weighted', zero_division=0)), | |
| 'weightedF1': float(f1_score(actual, predicted, average='weighted', zero_division=0)), | |
| 'perClass': per_class, | |
| 'confusionMatrix': matrix.tolist(), | |
| }, | |
| } | |
| args.output_dir.mkdir(parents=True, exist_ok=True) | |
| metrics_path = args.output_dir / 'evaluation_metrics.json' | |
| metrics_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding='utf-8') | |
| plt.figure(figsize=(12, 10)) | |
| sns.heatmap(matrix, annot=True, fmt='d', cmap='Blues', xticklabels=labels, yticklabels=labels) | |
| plt.title(f'Mission17 AI confusion matrix — {args.dataset_version}') | |
| plt.ylabel('Actual class') | |
| plt.xlabel('Predicted class') | |
| plt.xticks(rotation=45, ha='right') | |
| plt.yticks(rotation=0) | |
| plt.tight_layout() | |
| chart_path = args.output_dir / 'confusion_matrix.png' | |
| plt.savefig(chart_path, dpi=180) | |
| plt.close() | |
| summary = { | |
| 'metricsFile': str(metrics_path), | |
| 'confusionMatrixFile': str(chart_path), | |
| 'samples': report['dataset']['evaluatedSampleCount'], | |
| 'accuracy': report['metrics']['accuracy'], | |
| 'weightedPrecision': report['metrics']['weightedPrecision'], | |
| 'weightedRecall': report['metrics']['weightedRecall'], | |
| 'weightedF1': report['metrics']['weightedF1'], | |
| } | |
| print(json.dumps(summary, indent=2, sort_keys=True)) | |
| return 0 | |
| if __name__ == '__main__': | |
| raise SystemExit(main()) | |