update: api calls and data preprocessing

app.py

@@ -2,9 +2,11 @@
 from fastapi import FastAPI, File, UploadFile, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 import uvicorn
-from model import load_model, predict
+from model import load_model, predict, predict_from_frames
 import time
-import os
+from pydantic import BaseModel
+from typing import List
+import base64
 
 app = FastAPI(
     title="ISL Recognition API",
@@ -12,50 +14,149 @@ app = FastAPI(
     version="1.0.0"
 )
 
-# Allow all origins so your Flutter app can talk to it
+# Allow all origins (for Flutter / frontend apps)
 app.add_middleware(
-    CORSMiddleware,    allow_origins=["*"],
+    CORSMiddleware,
+    allow_origins=["*"],
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-# Global variable for the model
+# Global state
 model = None
+model_loaded = False
+model_error = None
 
+
+#  STARTUP 
 @app.on_event("startup")
 async def startup_event():
-    global model
-    # This calls the function in your model.py to download and load the .pt file
-    model = load_model()
-    print("Model loaded and API is ready!")
+    global model, model_loaded, model_error
+    try:
+        model = load_model()
+        model_loaded = True
+        model_error = None
+        print("Model loaded and API is ready!")
+    except Exception as e:
+        model_loaded = False
+        model_error = str(e)
+        print("Model failed to load:", e)
+
 
+#  ROOT 
 @app.get("/")
 def root():
-    return {"status": "ISL API is running", "message": "Send a POST request to /predict"}
+    return {
+        "status": "ISL API is running",
+        "message": "Send a POST request to /predict (video) or /predict_frames (frames list)"
+    }
+
+
+#  HEALTH 
+@app.get("/health")
+def health():
+    if not model_loaded or model is None:
+        return {
+            "status": "error",
+            "model_loaded": False,
+            "error": model_error
+        }
+
+    return {
+        "status": "ok",
+        "model_loaded": True,
+        "device": str(next(model.parameters()).device)
+    }
+
+
+#  DEEP HEALTH 
+@app.get("/health/deep")
+def health_deep():
+    if not model_loaded or model is None:
+        raise HTTPException(status_code=503, detail="Model not loaded")
+
+    try:
+        import torch
+
+        dummy = torch.zeros(1, 3, 16, 224, 224).to(
+            next(model.parameters()).device
+        )
+
+        with torch.no_grad():
+            _ = model(dummy)
+
+        return {
+            "status": "ok",
+            "inference": "working"
+        }
+
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Inference failed: {str(e)}"
+        )
+    
+class FramesPayload(BaseModel):
+    frames: List[str]  # List of base64 encoded JPEG/PNG images
+    top_k: int = 5
+
+@app.post("/predict_frames")
+async def predict_frames_api(payload: FramesPayload):
+    if not model_loaded or model is None:
+        raise HTTPException(status_code=503, detail="Model is not ready")
+
+    if not payload.frames:
+        raise HTTPException(status_code=400, detail="No frames provided")
+
+    start_time = time.time()
+    
+    try:
+        # Convert base64 strings to bytes
+        frames_bytes = [base64.b64decode(f) for f in payload.frames]
+        result = predict_from_frames(model, frames_bytes, top_k=payload.top_k)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Inference error: {str(e)}"
+        )
+
+    result["inference_time_ms"] = round((time.time() - start_time) * 1000, 2)
+    return result
+
 
-@app.post("/predict")
+#  PREDICT 
+@app.post("/predict")   
 async def predict_sign(file: UploadFile = File(...), top_k: int = 5):
-    # Validate that it's a video
+    # Validate file type
     if not file.filename.lower().endswith(('.mp4', '.mov', '.avi', '.mkv')):
         raise HTTPException(
             status_code=400,
             detail="Invalid file type. Please upload a video (.mp4, .mov, etc.)"
         )
 
-    if model is None:
-        raise HTTPException(status_code=503, detail="Model is still loading...")
+    # Ensure model is ready
+    if not model_loaded or model is None:
+        raise HTTPException(
+            status_code=503,
+            detail="Model is not ready"
+        )
 
     start_time = time.time()
     video_bytes = await file.read()
-    
-    # This calls the prediction logic in your model.py
-    result = predict(model, video_bytes, top_k=top_k)
-    
+
+    try:
+        result = predict(model, video_bytes, top_k=top_k)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Inference error: {str(e)}"
+        )
+
     result["inference_time_ms"] = round((time.time() - start_time) * 1000, 2)
     result["filename"] = file.filename
-    
+
     return result
 
+
 if __name__ == "__main__":
-    # Hugging Face Spaces usually look for port 7860
-    uvicorn.run("app.py:app", host="0.0.0.0", port=7860)
+    uvicorn.run("app:app", host="0.0.0.0", port=7860)

model.py

@@ -8,6 +8,8 @@ from decord.bridge import set_bridge
 import gc
 import tempfile
 import os
+import cv2
+import numpy as np
 
 # Exactly 76 classes from your notebook metadata
 CLASSES = [
@@ -90,12 +92,13 @@ def preprocess_video(video_bytes: bytes, clip_length: int = 16):
             
         # Ensure video is a torch tensor in (Frames, Channels, Height, Width)
         video = vr.get_batch(indices) 
-        video = video.permute(0, 3, 1, 2).float() # Convert to Float for the processor
+        video = video.permute(0, 3, 1, 2).to(torch.uint8) # Convert to Float for the processor
         
         # Pass as a list of Tensors
         processed = image_processor(
-            list(video), 
-            return_tensors='pt'
+            list(video),
+            return_tensors='pt',
+            input_data_format='channels_first'
         )
         
         pixel_values = processed['pixel_values'].squeeze(0) # (T, C, H, W)
@@ -106,6 +109,77 @@ def preprocess_video(video_bytes: bytes, clip_length: int = 16):
         if os.path.exists(tmp_path):
             os.remove(tmp_path)
 
+def preprocess_frames(frames_list_bytes: list[bytes], clip_length: int = 16):
+    """
+    Processes a list of raw frame bytes (JPEG/PNG encoded) into the Swin3D model input format.
+    Eliminates video encoding/decoding and disk I/O.
+    """
+    image_processor = VivitImageProcessor(
+        do_resize=True,
+        size={"shortest_edge": 224},
+        do_center_crop=True,
+        crop_size={"height": 224, "width": 224},
+        do_rescale=True,
+        rescale_factor=1/255,
+        do_normalize=True,
+        image_mean=[0.5, 0.5, 0.5],
+        image_std=[0.5, 0.5, 0.5],
+    )
+    
+    frames = []
+    for frame_bytes in frames_list_bytes:
+        # Decode image from bytes
+        nparr = np.frombuffer(frame_bytes, np.uint8)
+        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+        if img is not None:
+            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+            frames.append(img)
+            
+    if not frames:
+        raise ValueError("No valid frames decoded")
+        
+    # Temporal sampling/padding
+    if len(frames) < clip_length:
+        frames += [frames[-1]] * (clip_length - len(frames))
+    elif len(frames) > clip_length:
+        frames = frames[:clip_length]
+        
+    # Processor expects list of numpy arrays (H, W, C)
+    processed = image_processor(
+        frames,
+        return_tensors='pt',
+        # image_processor handles (T, C, H, W) return with return_tensors='pt' 
+        # but we need to check internal dimension order
+    )
+    
+    pixel_values = processed['pixel_values'].squeeze(0) # (T, C, H, W)
+    pixel_values = pixel_values.permute(1, 0, 2, 3)    # (C, T, H, W) for Swin3D
+    
+    return pixel_values.unsqueeze(0)
+
+def predict_from_frames(model, frames_list_bytes: list[bytes], top_k: int = 5):
+    """Runs inference from raw frame bytes"""
+    pixel_values = preprocess_frames(frames_list_bytes).to(DEVICE)
+    
+    with torch.no_grad():
+        outputs = model(pixel_values)
+        probabilities = torch.nn.functional.softmax(outputs, dim=-1)[0]
+    
+    top_probs, top_indices = torch.topk(probabilities, k=top_k)
+    
+    results = []
+    for i in range(top_k):
+        results.append({
+            "class": CLASSES[top_indices[i].item()],
+            "confidence": float(top_probs[i].item())
+        })
+        
+    return {
+        "prediction": results[0]["class"],
+        "confidence": results[0]["confidence"],
+        "top_k": results
+    }
+
 def predict(model, video_bytes: bytes, top_k: int = 5):
     """Runs inference and returns the top results"""
     pixel_values = preprocess_video(video_bytes).to(DEVICE)