Spaces:
Sleeping
Sleeping
| """ | |
| utils/model_loader.py | |
| ====================== | |
| Singleton model loader for the FGH-Dermatology inference pipeline. | |
| The trained `.h5` Keras model is loaded exactly once at process startup and | |
| cached in a module-level variable, avoiding the cost of re-deserialising the | |
| network on every incoming request. | |
| Model sourcing | |
| --------------- | |
| Two sourcing strategies are supported, resolved in this order: | |
| 1. HUGGING FACE HUB — if `Config.HF_MODEL_REPO_ID` is set, the `.h5` file is | |
| downloaded (or read from local cache, on subsequent restarts) from the | |
| given Hugging Face model repository using `huggingface_hub.hf_hub_download`. | |
| This is the recommended path for deployment targets with small repo/slug | |
| size limits (e.g. Render, Heroku) where a multi-hundred-MB model file | |
| cannot be committed to the Git repository that gets deployed. | |
| 2. LOCAL FILE PATH — if `Config.HF_MODEL_REPO_ID` is not set, the loader | |
| falls back to reading a local `.h5` file at `Config.MODEL_PATH`. This | |
| suits local development or any environment without a strict size limit. | |
| Demo mode | |
| --------- | |
| If neither a Hugging Face repo is configured nor a local file is found | |
| (e.g. a fresh checkout before a clinician has wired up the trained model), | |
| the loader transparently builds and persists a small, randomly-initialised | |
| CNN with the correct architecture shape (a `Conv2D` backbone feeding a | |
| softmax head over `Config.CLASS_LABELS`). This keeps the full application — | |
| UI, upload flow, Grad-CAM, NLP reporting — runnable end-to-end for | |
| evaluation, onboarding, and QA, while making it obvious in the UI that | |
| predictions are not yet backed by a clinically validated model. | |
| """ | |
| import logging | |
| import os | |
| import numpy as np | |
| import tensorflow as tf | |
| from config import Config | |
| logger = logging.getLogger("fgh_dermatology.model_loader") | |
| _model = None | |
| _is_demo_model = False | |
| _load_error = None | |
| def _resolve_model_path() -> str: | |
| """ | |
| Determine the local filesystem path to the `.h5` model, downloading it | |
| from Hugging Face Hub first if `Config.HF_MODEL_REPO_ID` is configured. | |
| Returns | |
| ------- | |
| str | |
| Path to a local `.h5` file, or an empty string if no model could be | |
| resolved (triggering demo mode in `get_model`). | |
| """ | |
| if Config.HF_MODEL_REPO_ID: | |
| try: | |
| from huggingface_hub import hf_hub_download | |
| from huggingface_hub.utils import HfHubHTTPError, LocalEntryNotFoundError | |
| except ImportError as exc: | |
| raise RuntimeError( | |
| "HF_MODEL_REPO_ID is set, but the 'huggingface_hub' package " | |
| "is not installed. Add 'huggingface_hub' to requirements.txt " | |
| "and redeploy." | |
| ) from exc | |
| logger.info( | |
| "Resolving model '%s' from Hugging Face Hub repo '%s' (revision: %s)...", | |
| Config.HF_MODEL_FILENAME, | |
| Config.HF_MODEL_REPO_ID, | |
| Config.HF_MODEL_REVISION, | |
| ) | |
| os.makedirs(Config.HF_CACHE_DIR, exist_ok=True) | |
| try: | |
| local_path = hf_hub_download( | |
| repo_id=Config.HF_MODEL_REPO_ID, | |
| filename=Config.HF_MODEL_FILENAME, | |
| revision=Config.HF_MODEL_REVISION, | |
| cache_dir=Config.HF_CACHE_DIR, | |
| token=Config.HF_TOKEN, | |
| ) | |
| logger.info("Model resolved from Hugging Face Hub: %s", local_path) | |
| return local_path | |
| except HfHubHTTPError as exc: | |
| status = getattr(getattr(exc, "response", None), "status_code", None) | |
| if status == 401 or status == 403: | |
| raise RuntimeError( | |
| f"Authentication/permission error while accessing Hugging " | |
| f"Face repo '{Config.HF_MODEL_REPO_ID}'. If this is a " | |
| "private repo, verify HF_TOKEN is set correctly and has " | |
| "at least 'Read' access. If it is public, double-check " | |
| "the repo ID for typos." | |
| ) from exc | |
| if status == 404: | |
| raise RuntimeError( | |
| f"File '{Config.HF_MODEL_FILENAME}' was not found in " | |
| f"Hugging Face repo '{Config.HF_MODEL_REPO_ID}' at " | |
| f"revision '{Config.HF_MODEL_REVISION}'. Verify the " | |
| "repo ID, filename (case-sensitive), and revision are " | |
| "exactly correct." | |
| ) from exc | |
| raise RuntimeError( | |
| f"Failed to download '{Config.HF_MODEL_FILENAME}' from Hugging " | |
| f"Face repo '{Config.HF_MODEL_REPO_ID}' (HTTP status: {status}). " | |
| "Check that the repo ID and filename are correct, that the " | |
| "repo is public (or HF_TOKEN is set correctly for a private " | |
| "repo), and that the deployment environment has outbound " | |
| "internet access." | |
| ) from exc | |
| except LocalEntryNotFoundError as exc: | |
| raise RuntimeError( | |
| f"Could not reach Hugging Face Hub to download " | |
| f"'{Config.HF_MODEL_FILENAME}' from repo " | |
| f"'{Config.HF_MODEL_REPO_ID}', and no cached copy was found " | |
| f"locally under '{Config.HF_CACHE_DIR}'. This usually means " | |
| "the deployment environment has no outbound internet access, " | |
| "or huggingface.co is unreachable/blocked on this network. " | |
| "Verify the service has outbound internet access enabled." | |
| ) from exc | |
| except Exception as exc: # noqa: BLE001 | |
| raise RuntimeError( | |
| f"Unexpected error while downloading the model from Hugging " | |
| f"Face Hub repo '{Config.HF_MODEL_REPO_ID}': {exc}" | |
| ) from exc | |
| # No Hugging Face repo configured — fall back to a local file path. | |
| if os.path.exists(Config.MODEL_PATH): | |
| return Config.MODEL_PATH | |
| return "" | |
| def _build_demo_model(num_classes: int, input_size: int) -> tf.keras.Model: | |
| """ | |
| Construct a lightweight CNN with a clear final Conv2D layer (for | |
| Grad-CAM) and a softmax classification head. Used only when no trained | |
| `.h5` model is available at `Config.MODEL_PATH`. | |
| """ | |
| inputs = tf.keras.Input(shape=(input_size, input_size, 3), name="input_image") | |
| x = tf.keras.layers.Conv2D(32, 3, padding="same", activation="relu", name="conv_block1")(inputs) | |
| x = tf.keras.layers.MaxPooling2D()(x) | |
| x = tf.keras.layers.Conv2D(64, 3, padding="same", activation="relu", name="conv_block2")(x) | |
| x = tf.keras.layers.MaxPooling2D()(x) | |
| x = tf.keras.layers.Conv2D(128, 3, padding="same", activation="relu", name="conv_block3_final")(x) | |
| pooled = tf.keras.layers.GlobalAveragePooling2D()(x) | |
| dense = tf.keras.layers.Dense(64, activation="relu")(pooled) | |
| outputs = tf.keras.layers.Dense(num_classes, activation="softmax", name="predictions")(dense) | |
| model = tf.keras.Model(inputs, outputs, name="fgh_dermatology_demo_model") | |
| return model | |
| def get_model(): | |
| """ | |
| Return the cached singleton model instance. | |
| Resolution order: Hugging Face Hub -> local file path -> demo fallback. | |
| Design note: any failure while resolving or loading a configured model | |
| (network failure, bad repo ID, auth error, corrupted file, etc.) is | |
| caught here and causes a graceful fallback to the untrained demo model, | |
| rather than raising and taking down every route in the app (landing | |
| page, health check, etc.). The failure reason is logged and recorded via | |
| `get_load_error()` so it can be surfaced in the UI/health endpoint | |
| without crashing the platform for clinicians trying to use it. | |
| """ | |
| global _model, _is_demo_model, _load_error | |
| if _model is not None: | |
| return _model | |
| input_size = Config.MODEL_INPUT_SIZE | |
| num_classes = len(Config.CLASS_LABELS) | |
| try: | |
| model_path = _resolve_model_path() | |
| except RuntimeError as exc: | |
| logger.exception("Model resolution failed: %s", exc) | |
| _load_error = str(exc) | |
| model_path = "" | |
| else: | |
| _load_error = None | |
| if model_path: | |
| try: | |
| logger.info("Loading trained model from %s", model_path) | |
| _model = tf.keras.models.load_model(model_path, compile=False) | |
| _is_demo_model = False | |
| _load_error = None | |
| logger.info("Model loaded successfully.") | |
| return _model | |
| except Exception as exc: # noqa: BLE001 | |
| logger.exception("Failed to load model at %s: %s", model_path, exc) | |
| _load_error = ( | |
| f"Unable to load the model file at '{model_path}'. The file " | |
| "may be corrupted, truncated (check the Hugging Face upload " | |
| "completed fully), or incompatible with the installed " | |
| "TensorFlow/Keras version." | |
| ) | |
| source_hint = ( | |
| f"Hugging Face repo '{Config.HF_MODEL_REPO_ID}'" | |
| if Config.HF_MODEL_REPO_ID | |
| else f"local path '{Config.MODEL_PATH}'" | |
| ) | |
| logger.warning( | |
| "Falling back to DEMO MODE with an untrained placeholder network " | |
| "(source attempted: %s). Predictions are NOT clinically valid. " | |
| "Configure HF_MODEL_REPO_ID (recommended) or MODEL_PATH correctly " | |
| "to enable live inference.", | |
| source_hint, | |
| ) | |
| _model = _build_demo_model(num_classes, input_size) | |
| _is_demo_model = True | |
| return _model | |
| def is_demo_model() -> bool: | |
| """Return True if the active model is the untrained demo placeholder.""" | |
| # Ensure model has been initialised so the flag is accurate. | |
| get_model() | |
| return _is_demo_model | |
| def get_load_error(): | |
| """ | |
| Return a human-readable string describing why the configured model | |
| (Hugging Face or local) could not be loaded, or None if the active | |
| model loaded successfully (including the case where demo mode was | |
| entered intentionally because no model was configured at all). | |
| """ | |
| get_model() | |
| return _load_error | |
| def get_input_size() -> int: | |
| """ | |
| Infer the expected square input resolution from the loaded model when | |
| possible, otherwise fall back to Config.MODEL_INPUT_SIZE. | |
| """ | |
| model = get_model() | |
| try: | |
| shape = model.input_shape | |
| if isinstance(shape, list): | |
| shape = shape[0] | |
| height = shape[1] | |
| if isinstance(height, int): | |
| return height | |
| except Exception: # noqa: BLE001 | |
| pass | |
| return Config.MODEL_INPUT_SIZE | |
| def predict(preprocessed_image: np.ndarray): | |
| """ | |
| Run inference on a preprocessed image batch. | |
| Parameters | |
| ---------- | |
| preprocessed_image : np.ndarray | |
| Array of shape (1, H, W, 3), values normalised as expected by the model. | |
| Returns | |
| ------- | |
| dict with: | |
| predicted_class : str — top-1 class label | |
| confidence : float — top-1 softmax probability (0-1) | |
| top_k : list[tuple[str, float]] — full ranked differential, | |
| most likely first. With this platform's 4-class | |
| taxonomy, all 4 classes are returned so clinicians | |
| see the complete probability distribution rather | |
| than an arbitrary subset. | |
| raw_probs : np.ndarray — full softmax vector | |
| """ | |
| model = get_model() | |
| raw_probs = model.predict(preprocessed_image, verbose=0)[0] | |
| labels = Config.CLASS_LABELS | |
| # Guard against a custom/uploaded model with a different class count. | |
| n = min(len(labels), len(raw_probs)) | |
| ranked_idx = np.argsort(raw_probs[:n])[::-1] | |
| top_k = [(labels[i], float(raw_probs[i])) for i in ranked_idx] | |
| predicted_class, confidence = top_k[0] | |
| return { | |
| "predicted_class": predicted_class, | |
| "confidence": confidence, | |
| "top_k": top_k, | |
| "raw_probs": raw_probs, | |
| } | |