""" 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, }