File size: 4,488 Bytes
23b40a9
 
 
 
 
 
 
 
 
 
 
5d7a2e7
 
23b40a9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5d7a2e7
 
23b40a9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5d7a2e7
 
 
 
23b40a9
 
 
 
5d7a2e7
23b40a9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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)}")