Deploy FastAPI image classifier

.dockerignore

@@ -0,0 +1,9 @@
+__pycache__/
+.pytest_cache/
+.git/
+.github/
+tests/
+docs/*.json
+*.pyc
+*.pyo
+*.pyd

.github/workflows/test.yml

@@ -0,0 +1,47 @@
+name: test-and-deploy
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: python -m pip install --upgrade pip && pip install -r requirements.txt
+
+      - name: Run pytest
+        run: pytest
+
+  deploy:
+    needs: test
+    if: github.event_name == 'push'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install deployment dependency
+        run: python -m pip install --upgrade pip && pip install huggingface-hub
+
+      - name: Deploy to Hugging Face Spaces
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+          HF_SPACE_REPO: ${{ secrets.HF_SPACE_REPO }}
+        run: python scripts/deploy_to_hf_spaces.py

.gitignore

@@ -0,0 +1,115 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+pip-wheel-metadata/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.env.*
+.venv
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Docker
+*.log
+docker-compose*.yml
+Dockerfile*
+!Dockerfile
+
+# Model artifacts
+model/*.pth
+model/*.pt
+model/*.h5
+model/*.onnx
+model/*.pb
+model/*.tflite
+model/*.joblib
+model/*.pkl
+model/*.ckpt
+
+# VS Code
+.vscode/
+
+# System files
+Thumbs.db
+ehthumbs.db
+Desktop.ini
+.DS_Store
+__MACOSX/

Dockerfile

@@ -0,0 +1,27 @@
+FROM cgr.dev/chainguard/python:latest-dev AS builder
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /build
+
+COPY requirements.txt ./
+RUN python -m pip install --upgrade pip \
+    && pip install --prefix=/install -r requirements.txt
+
+FROM cgr.dev/chainguard/python:latest
+
+ENV PYTHONPATH=/install/lib/python3.12/site-packages
+
+WORKDIR /app
+
+COPY --from=builder /install /install
+COPY app ./app
+COPY model ./model
+COPY scripts ./scripts
+COPY README.md ./README.md
+
+EXPOSE 8000
+
+CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,10 +1,95 @@
----
-title: SPACE
-emoji: 🐠
-colorFrom: yellow
-colorTo: green
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# FastAPI Image Classifier
+
+A small CPU-friendly image classification service built around a Hugging Face model, exported to ONNX, and quantized for faster inference.
+
+## What is included
+
+- Hugging Face source model: `timm/mobilenetv4_conv_medium.e500_r224_in1k`
+- Optimization scripts for PyTorch, ONNX, and INT8 quantized ONNX
+- Async FastAPI `POST /predict` endpoint
+- ProcessPoolExecutor-based CPU offload
+- Pytest coverage for API behavior and validation errors
+- Dockerfile for container packaging
+- GitHub Actions workflow for CI and Hugging Face Spaces deployment
+
+## Project structure
+
+- `app/` FastAPI application and model helpers
+- `model/` exported model artifacts
+- `scripts/` export, benchmark, and deployment utilities
+- `tests/` pytest suite
+- `.github/workflows/` CI/CD workflow
+
+## Setup
+
+```bash
+python -m pip install -r requirements.txt
+```
+
+## Export model artifacts
+
+```bash
+python scripts/export_models.py
+```
+
+This generates:
+
+- `model/model_original.pt`
+- `model/model.onnx`
+- `model/model_quantized.onnx`
+
+## Run the API locally
+
+```bash
+uvicorn app.main:app --reload
+```
+
+## Predict with curl
+
+```bash
+curl -X POST \
+  -F "file=@cat.jpg" \
+  http://127.0.0.1:8000/predict
+```
+
+Example response:
+
+```json
+{
+  "label": "cat",
+  "confidence": 0.98
+}
+```
+
+## Run tests
+
+```bash
+pytest
+```
+
+## Docker
+
+```bash
+docker build -t image-classifier .
+docker run -p 8000:8000 image-classifier
+```
+
+## Benchmarking
+
+Run the benchmark script after exporting the models:
+
+```bash
+python scripts/benchmark_models.py --image path/to/sample.jpg
+```
+
+The results are written to `docs/benchmark_results.json`.
+
+## Hugging Face Spaces deployment
+
+Set these secrets before the deployment job runs:
+
+- `HF_TOKEN`
+- `HF_SPACE_REPO`
+
+The workflow uploads the repository to a Docker-based Space.
+

app/config.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="APP_", env_file=".env", extra="ignore")
+
+    hf_model_name: str = "timm/mobilenetv4_conv_medium.e500_r224_in1k"
+    input_size: int = 224
+    max_upload_mb: int = 5
+    accepted_extensions: tuple[str, ...] = (".jpg", ".jpeg", ".png", ".webp", ".bmp")
+    accepted_content_types: tuple[str, ...] = ("image/jpeg", "image/png", "image/webp", "image/bmp")
+    worker_processes: int = 2
+    model_dir: Path = Field(default_factory=lambda: Path(__file__).resolve().parents[1] / "model")
+    docs_dir: Path = Field(default_factory=lambda: Path(__file__).resolve().parents[1] / "docs")
+
+    @property
+    def max_upload_bytes(self) -> int:
+        return self.max_upload_mb * 1024 * 1024
+
+    @property
+    def torch_weights_path(self) -> Path:
+        return self.model_dir / "model_original.pt"
+
+    @property
+    def onnx_path(self) -> Path:
+        return self.model_dir / "model.onnx"
+
+    @property
+    def quantized_onnx_path(self) -> Path:
+        return self.model_dir / "model_quantized.onnx"
+
+    @property
+    def benchmark_output_path(self) -> Path:
+        return self.docs_dir / "benchmark_results.json"
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    return Settings()

app/image_validation.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from io import BytesIO
+from pathlib import Path
+
+from fastapi import HTTPException, UploadFile, status
+from PIL import Image, UnidentifiedImageError
+from pydantic import BaseModel, Field, model_validator, ValidationError
+
+from .config import Settings
+
+
+class ImageUploadMetadata(BaseModel):
+    filename: str
+    content_type: str
+    size_bytes: int = Field(gt=0)
+
+    @model_validator(mode="after")
+    def validate_metadata(self) -> "ImageUploadMetadata":
+        allowed_extensions = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
+        allowed_content_types = {"image/jpeg", "image/png", "image/webp", "image/bmp"}
+        suffix = Path(self.filename).suffix.lower()
+        if suffix not in allowed_extensions:
+            raise ValueError("Unsupported file type")
+        if self.content_type not in allowed_content_types:
+            raise ValueError("Unsupported file type")
+        return self
+
+
+async def read_and_validate_image(file: UploadFile, settings: Settings) -> bytes:
+    if not file.filename:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="File name is required")
+
+    data = await file.read(settings.max_upload_bytes + 1)
+    try:
+        metadata = ImageUploadMetadata(
+            filename=file.filename,
+            content_type=file.content_type or "",
+            size_bytes=len(data),
+        )
+    except ValidationError:
+        raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail="Unsupported file type")
+
+    if metadata.size_bytes > settings.max_upload_bytes:
+        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File exceeds 5 MB limit")
+
+    try:
+        with Image.open(BytesIO(data)) as image:
+            image.verify()
+        with Image.open(BytesIO(data)) as image:
+            image.convert("RGB")
+    except (UnidentifiedImageError, OSError, ValueError) as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Unable to decode image") from exc
+
+    return data

app/label_map.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+DOG_HINTS = {
+    "terrier",
+    "retriever",
+    "spaniel",
+    "hound",
+    "poodle",
+    "collie",
+    "shepherd",
+    "pug",
+    "bulldog",
+    "beagle",
+    "dalmatian",
+    "chihuahua",
+    "maltese",
+    "samoyed",
+    "husky",
+    "boxer",
+    "doberman",
+    "shiba",
+    "akita",
+    "rottweiler",
+    "dog",
+    "canine",
+}
+
+CAT_HINTS = {
+    "cat",
+    "tabby",
+    "tiger cat",
+    "persian cat",
+    "siamese",
+    "lynx",
+}
+
+
+def coarse_label(raw_label: str) -> str:
+    normalized = raw_label.lower()
+    if any(keyword in normalized for keyword in DOG_HINTS):
+        return "dog"
+    if any(keyword in normalized for keyword in CAT_HINTS):
+        return "cat"
+    cleaned = raw_label.split(",", maxsplit=1)[0].strip().lower()
+    return cleaned.replace(" ", "_")

app/main.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import asyncio
+from concurrent.futures import ProcessPoolExecutor
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, File, HTTPException, Request, UploadFile, status
+
+from .config import Settings, get_settings
+from .image_validation import read_and_validate_image
+from .model_backend import PredictionResult, predict_image_bytes
+from .schemas import PredictionResponse
+
+
+def _predict_in_worker(image_bytes: bytes, settings_dict: dict[str, str]) -> PredictionResult:
+    settings = Settings(**settings_dict)
+    return predict_image_bytes(image_bytes, settings)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    settings = get_settings()
+    executor = ProcessPoolExecutor(max_workers=settings.worker_processes)
+    app.state.executor = executor
+    app.state.settings = settings
+
+    async def process_pool_predictor(image_bytes: bytes) -> PredictionResult:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(
+            executor,
+            _predict_in_worker,
+            image_bytes,
+            settings.model_dump(mode="json"),
+        )
+
+    app.state.predictor = process_pool_predictor
+    try:
+        yield
+    finally:
+        executor.shutdown(wait=True, cancel_futures=True)
+
+
+def create_app(use_process_pool: bool = True, predictor=None) -> FastAPI:
+    app = FastAPI(title="Image Classifier API", version="1.0.0", lifespan=lifespan if use_process_pool else None)
+
+    if not use_process_pool:
+        app.state.settings = get_settings()
+        app.state.predictor = predictor or (lambda image_bytes: predict_image_bytes(image_bytes, app.state.settings))
+    elif predictor is not None:
+        app.state.predictor = predictor
+
+    @app.post("/predict", response_model=PredictionResponse)
+    async def predict(request: Request, file: UploadFile = File(...)) -> PredictionResponse:
+        settings = get_settings()
+        image_bytes = await read_and_validate_image(file, settings)
+        app_instance = request.app
+        predictor_callable = getattr(app_instance.state, "predictor", None)
+        if predictor_callable is None:
+            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Predictor is not initialized")
+
+        result = predictor_callable(image_bytes)
+        if asyncio.iscoroutine(result):
+            result = await result
+
+        return PredictionResponse(label=result.label, confidence=result.confidence)
+
+    @app.get("/healthz")
+    async def healthz() -> dict[str, str]:
+        return {"status": "ok"}
+
+    return app
+
+
+app = create_app()

app/model_backend.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import io
+import json
+import statistics
+import time
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Callable
+
+import numpy as np
+import onnxruntime as ort
+import torch
+import torch.nn as nn
+from PIL import Image
+from onnxruntime.quantization import QuantType, quantize_dynamic
+from transformers import AutoConfig, AutoImageProcessor, AutoModelForImageClassification
+
+from .config import Settings, get_settings
+from .label_map import coarse_label
+
+
+@dataclass(frozen=True)
+class PredictionResult:
+    label: str
+    confidence: float
+
+
+class _TorchExportWrapper(nn.Module):
+    def __init__(self, model: nn.Module) -> None:
+        super().__init__()
+        self.model = model
+
+    def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
+        return self.model(pixel_values=pixel_values).logits
+
+
+class TorchImageClassifier:
+    def __init__(self, model_name: str) -> None:
+        self.processor = AutoImageProcessor.from_pretrained(model_name)
+        self.model = AutoModelForImageClassification.from_pretrained(model_name)
+        self.model.eval()
+        self.id2label = self.model.config.id2label
+
+    def predict(self, image_bytes: bytes) -> PredictionResult:
+        image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
+        inputs = self.processor(images=image, return_tensors="pt")
+        with torch.inference_mode():
+            outputs = self.model(**inputs)
+            probabilities = torch.softmax(outputs.logits, dim=-1)[0]
+        top_index = int(torch.argmax(probabilities).item())
+        raw_label = self.id2label.get(top_index, str(top_index))
+        return PredictionResult(
+            label=coarse_label(raw_label),
+            confidence=float(probabilities[top_index].item()),
+        )
+
+    def export_checkpoint(self, output_path: Path, model_name: str) -> None:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            "model_name": model_name,
+            "state_dict": self.model.state_dict(),
+            "id2label": self.id2label,
+        }
+        torch.save(payload, output_path)
+
+    def export_onnx(self, output_path: Path, input_size: int) -> None:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        dummy_image = Image.new("RGB", (input_size, input_size), color=(255, 255, 255))
+        dummy_inputs = self.processor(images=dummy_image, return_tensors="pt")["pixel_values"]
+        wrapper = _TorchExportWrapper(self.model)
+        torch.onnx.export(
+            wrapper,
+            dummy_inputs,
+            output_path,
+            input_names=["pixel_values"],
+            output_names=["logits"],
+            dynamic_axes={"pixel_values": {0: "batch_size"}, "logits": {0: "batch_size"}},
+            opset_version=17,
+        )
+
+
+class OnnxImageClassifier:
+    def __init__(self, model_path: Path, model_name: str) -> None:
+        self.session = ort.InferenceSession(str(model_path), providers=["CPUExecutionProvider"])
+        self.processor = AutoImageProcessor.from_pretrained(model_name)
+        self.id2label = AutoConfig.from_pretrained(model_name).id2label
+        self.input_name = self.session.get_inputs()[0].name
+
+    def predict(self, image_bytes: bytes) -> PredictionResult:
+        image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
+        inputs = self.processor(images=image, return_tensors="np")
+        pixel_values = inputs["pixel_values"].astype(np.float32)
+        logits = self.session.run(None, {self.input_name: pixel_values})[0][0]
+        probabilities = np.exp(logits - np.max(logits))
+        probabilities = probabilities / probabilities.sum()
+        top_index = int(np.argmax(probabilities))
+        raw_label = self.id2label.get(top_index, str(top_index))
+        return PredictionResult(label=coarse_label(raw_label), confidence=float(probabilities[top_index]))
+
+
+@lru_cache(maxsize=1)
+def get_torch_classifier(model_name: str | None = None) -> TorchImageClassifier:
+    settings = get_settings()
+    return TorchImageClassifier(model_name or settings.hf_model_name)
+
+
+@lru_cache(maxsize=1)
+def get_onnx_classifier(model_path: str, model_name: str) -> OnnxImageClassifier:
+    return OnnxImageClassifier(Path(model_path), model_name)
+
+
+def get_best_classifier(settings: Settings | None = None) -> Callable[[bytes], PredictionResult]:
+    config = settings or get_settings()
+    if config.quantized_onnx_path.exists():
+        backend = get_onnx_classifier(str(config.quantized_onnx_path), config.hf_model_name)
+        return backend.predict
+    if config.onnx_path.exists():
+        backend = get_onnx_classifier(str(config.onnx_path), config.hf_model_name)
+        return backend.predict
+    backend = get_torch_classifier(config.hf_model_name)
+    return backend.predict
+
+
+def predict_image_bytes(image_bytes: bytes, settings: Settings | None = None) -> PredictionResult:
+    classifier = get_best_classifier(settings)
+    return classifier(image_bytes)
+
+
+def export_model_assets(settings: Settings | None = None) -> None:
+    config = settings or get_settings()
+    torch_classifier = get_torch_classifier(config.hf_model_name)
+    torch_classifier.export_checkpoint(config.torch_weights_path, config.hf_model_name)
+    torch_classifier.export_onnx(config.onnx_path, config.input_size)
+    quantize_dynamic(str(config.onnx_path), str(config.quantized_onnx_path), weight_type=QuantType.QInt8)
+
+
+def benchmark_predictions(
+    predictor: Callable[[bytes], PredictionResult],
+    image_bytes: bytes,
+    runs: int = 20,
+    warmup_runs: int = 5,
+) -> dict[str, float]:
+    for _ in range(warmup_runs):
+        predictor(image_bytes)
+
+    timings: list[float] = []
+    for _ in range(runs):
+        start = time.perf_counter()
+        predictor(image_bytes)
+        timings.append(time.perf_counter() - start)
+
+    average_seconds = statistics.mean(timings)
+    return {
+        "latency_ms": statistics.median(timings) * 1000.0,
+        "throughput_rps": 1.0 / average_seconds if average_seconds else 0.0,
+    }
+
+
+def serialize_benchmark_report(rows: list[dict[str, float | str]]) -> str:
+    return json.dumps(rows, indent=2)

app/schemas.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class PredictionResponse(BaseModel):
+    label: str
+    confidence: float = Field(ge=0.0, le=1.0)
+
+
+class BenchmarkResult(BaseModel):
+    variant: str
+    model_size_mb: float
+    latency_ms: float
+    throughput_rps: float

docs/report.md

@@ -0,0 +1,63 @@
+# Project Report
+
+## Model Selection
+
+- Model: `timm/mobilenetv4_conv_medium.e500_r224_in1k`
+- Task: image classification on ImageNet-1k
+- Input size: `224 x 224`
+- Reason for selection: compact backbone, low CPU cost, and straightforward export to ONNX
+
+## Optimization Phase
+
+Run `python scripts/export_models.py` and `python scripts/benchmark_models.py --image path/to/sample.jpg` to produce the measured values below.
+
+| Model | Size | Latency |
+| --- | ---: | ---: |
+| Original | TBD MB | TBD ms |
+| ONNX | TBD MB | TBD ms |
+| Quantized | TBD MB | TBD ms |
+
+## Error Handling Strategy
+
+- Missing file: `422`
+- Invalid file type: `415`
+- Too large file: `413`
+- Corrupted image: `422`
+- Unexpected server crash: `500`
+
+Validation checks cover file extension, MIME type, upload size, and image decodability before inference starts.
+
+## System Architecture
+
+```mermaid
+flowchart TD
+    Client --> FastAPI
+    FastAPI --> ProcessPool
+    ProcessPool --> ONNXModel
+    ONNXModel --> Response
+```
+
+## CI/CD Pipeline
+
+```mermaid
+flowchart TD
+    Push --> GitHubActions
+    GitHubActions --> Pytest
+    Pytest --> DockerBuild
+    DockerBuild --> HuggingFaceSpaces
+```
+
+## Performance Testing
+
+Load testing is intended for both local Docker and Hugging Face Spaces with JMeter against `POST /predict`. Collect throughput, request latency, and P95 latency, then identify the CPU saturation point where response time rises sharply.
+
+## Deliverables Checklist
+
+- FastAPI application
+- Model export scripts
+- Benchmarking script
+- Quantized ONNX model path
+- Pytest suite
+- Docker packaging
+- GitHub Actions workflow
+- Deployment script for Hugging Face Spaces

model/README.md

@@ -0,0 +1,15 @@
+# Model Artifacts
+
+This folder is expected to contain the exported model files used by the API and benchmark scripts:
+
+- `model_original.pt`
+- `model.onnx`
+- `model_quantized.onnx`
+
+Generate them with:
+
+```bash
+python scripts/export_models.py
+```
+
+The Docker image copies this directory so the API can start without exporting the model at runtime.

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests

requirements.txt

@@ -0,0 +1,16 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+python-multipart==0.0.20
+pydantic==2.10.6
+pydantic-settings==2.8.1
+pillow==11.1.0
+numpy==2.2.2
+torch==2.6.0
+torchvision==0.21.0
+transformers==4.48.3
+huggingface-hub==0.28.1
+onnx==1.17.0
+onnxruntime==1.20.1
+pytest==8.3.4
+pytest-asyncio==0.25.3
+httpx==0.28.1

scripts/benchmark_models.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from PIL import Image
+
+from app.config import get_settings
+from app.model_backend import (
+    benchmark_predictions,
+    get_onnx_classifier,
+    get_torch_classifier,
+    serialize_benchmark_report,
+)
+
+
+def load_sample_image(image_path: Path) -> bytes:
+    with Image.open(image_path) as image:
+        rgb_image = image.convert("RGB")
+        from io import BytesIO
+
+        buffer = BytesIO()
+        rgb_image.save(buffer, format="JPEG")
+        return buffer.getvalue()
+
+
+def model_size_mb(model_path: Path) -> float:
+    return model_path.stat().st_size / (1024 * 1024)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Benchmark original, ONNX, and quantized model variants.")
+    parser.add_argument("--image", type=Path, required=True, help="Sample image used for benchmarking")
+    args = parser.parse_args()
+
+    settings = get_settings()
+    image_bytes = load_sample_image(args.image)
+
+    torch_backend = get_torch_classifier(settings.hf_model_name)
+    onnx_backend = get_onnx_classifier(str(settings.onnx_path), settings.hf_model_name)
+    quantized_backend = get_onnx_classifier(str(settings.quantized_onnx_path), settings.hf_model_name)
+
+    rows = [
+        {
+            "variant": "Original",
+            "model_size_mb": model_size_mb(settings.torch_weights_path),
+            **benchmark_predictions(torch_backend.predict, image_bytes),
+        },
+        {
+            "variant": "ONNX",
+            "model_size_mb": model_size_mb(settings.onnx_path),
+            **benchmark_predictions(onnx_backend.predict, image_bytes),
+        },
+        {
+            "variant": "Quantized",
+            "model_size_mb": model_size_mb(settings.quantized_onnx_path),
+            **benchmark_predictions(quantized_backend.predict, image_bytes),
+        },
+    ]
+
+    settings.docs_dir.mkdir(parents=True, exist_ok=True)
+    settings.benchmark_output_path.write_text(serialize_benchmark_report(rows), encoding="utf-8")
+    print(serialize_benchmark_report(rows))
+
+
+if __name__ == "__main__":
+    main()

scripts/deploy_to_hf_spaces.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+
+def main() -> None:
+    repo_id = os.environ.get("HF_SPACE_REPO")
+    token = os.environ.get("HF_TOKEN")
+    if not repo_id or not token:
+        raise SystemExit("HF_SPACE_REPO and HF_TOKEN must be set")
+
+    root = Path(__file__).resolve().parents[1]
+    api = HfApi(token=token)
+    api.create_repo(repo_id=repo_id, repo_type="space", space_sdk="docker", exist_ok=True)
+    api.upload_folder(
+        repo_id=repo_id,
+        repo_type="space",
+        folder_path=str(root),
+        commit_message="Deploy FastAPI image classifier",
+        ignore_patterns=[
+            ".git/*",
+            ".pytest_cache/*",
+            "__pycache__/*",
+            "tests/*",
+            "docs/benchmark_results.json",
+        ],
+    )
+
+
+if __name__ == "__main__":
+    main()

scripts/export_models.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from app.config import get_settings
+from app.model_backend import export_model_assets
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Export PyTorch, ONNX, and quantized ONNX model artifacts.")
+    parser.add_argument("--model-dir", type=Path, default=None, help="Override the default model directory")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    settings = get_settings()
+    if args.model_dir is not None:
+        settings = settings.model_copy(update={"model_dir": args.model_dir})
+    export_model_assets(settings)
+
+
+if __name__ == "__main__":
+    main()