app.py

import io
import numpy as np
import tensorflow as tf
from tensorflow import keras
from keras import layers
from fastapi import FastAPI, File, UploadFile, Form, HTTPException
from PIL import Image

app = FastAPI(title="DyslexiaLens Prediction API")

# ─── RE-REGISTER CUSTOM ARCHITECTURE COMPONENTS ───
# CHANGED: Swapped .saving to .utils to match the tf.keras wrapper ecosystem
@keras.utils.register_keras_serializable(package="Custom")
class AdaptiveContrastNorm(layers.Layer):
    def __init__(self, epsilon: float = 1e-6, **kwargs):
        super().__init__(**kwargs)
        self.epsilon = epsilon

    def build(self, input_shape):
        channels = input_shape[-1]
        self.gamma = self.add_weight(name='gamma', shape=(1, 1, 1, channels), initializer='ones', trainable=True)
        self.beta = self.add_weight(name='beta', shape=(1, 1, 1, channels), initializer='zeros', trainable=True)
        super().build(input_shape)

    def call(self, x, training=None):
        axes = [1, 2]
        mu = tf.reduce_mean(x, axis=axes, keepdims=True)
        sigma = tf.math.reduce_std(x, axis=axes, keepdims=True) + self.epsilon
        return self.gamma * ((x - mu) / sigma) + self.beta

    def get_config(self):
        config = super().get_config()
        config.update({'epsilon': self.epsilon})
        return config

# CHANGED: Swapped .saving to .utils here as well
@keras.utils.register_keras_serializable(package="Custom")
class MaskedHuberLoss(keras.losses.Loss):
    def __init__(self, delta: float = 0.5, **kwargs):
        super().__init__(**kwargs)
        self.delta = delta
        self._huber_fn = keras.losses.Huber(delta=delta, reduction='none')

    def call(self, y_true, y_pred):
        y_true = tf.cast(tf.reshape(y_true, [-1, 1]), tf.float32)
        y_pred = tf.cast(tf.reshape(y_pred, [-1, 1]), tf.float32)
        mask = tf.cast(y_true > 0.0, tf.float32)
        per_sample = self._huber_fn(y_true, y_pred)
        masked = per_sample * tf.squeeze(mask, axis=-1)
        return tf.reduce_sum(masked) / (tf.reduce_sum(mask) + 1e-8)

    def get_config(self):
        config = super().get_config()
        config.update({'delta': self.delta})
        return config

# ─── GLOBAL WEIGHT LOADING ───
MODEL_PATH = "dyslexialens_model.keras"
model = None

@app.on_event("startup")
def load_model():
    global model
    try:
        model = keras.models.load_model(
            MODEL_PATH,
            custom_objects={
                'AdaptiveContrastNorm': AdaptiveContrastNorm,
                'MaskedHuberLoss': MaskedHuberLoss
            }
        )
        print("Model successfully loaded onto CPU context.")
    except Exception as e:
        print(f"Error loading Keras model: {str(e)}")

@app.get("/")
def health_check():
    return {"status": "online", "model": "DyslexiaLens Late Fusion Pipeline ready"}

@app.post("/predict")
async def predict(
    stroke_density: float = Form(...),
    center_of_mass_x: float = Form(...),
    center_of_mass_y: float = Form(...),
    bounding_box_ratio: float = Form(...),
    stroke_transitions: float = Form(...),
    horizontal_symmetry: float = Form(...),
    file: UploadFile = File(...)
):
    if model is None:
        raise HTTPException(status_code=503, detail="Model is loading or uninitialized.")

    try:
        contents = await file.read()
        image = Image.open(io.BytesIO(contents)).convert('L')
        image = image.resize((128, 128), Image.BILINEAR)
        img_array = np.array(image, dtype=np.float32) / 255.0
        img_tensor = np.expand_dims(img_array, axis=(0, -1))

        feature_vector = np.array([
            stroke_density, center_of_mass_x, center_of_mass_y, 
            bounding_box_ratio, stroke_transitions, horizontal_symmetry
        ], dtype=np.float32).reshape(1, 6)

        predictions = model.predict({
            'image_input': img_tensor, 
            'feature_input': feature_vector
        })

        clf_probability = float(predictions[0][0][0])
        severity_score = float(predictions[1][0][0])
        is_dyslexia = clf_probability >= 0.40

        return {
            "status": "success",
            "prediction": {
                "has_dyslexia": is_dyslexia,
                "dyslexia_probability": round(clf_probability, 4),
                "severity_score": round(severity_score, 4)
            }
        }

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Inference Failure: {str(e)}")