import cv2 import numpy as np import tensorflow as tf import random from typing import Any, Dict, List class EndpointHandler(): def __init__(self, path=""): self.model = tf.saved_model.load(f'{path}/my_model') self.classes_1 = ["RoadAccidents", "Fighting", "NormalVideos"] self.locations = ['Miami', 'Smouha', 'Mandara', 'Sporting', 'Montazah'] def get_top_k(self, probs, k=1, label_map=None): if label_map is None: label_map = self.classes_1 top_predictions = tf.argsort(probs, axis=-1, direction='DESCENDING')[:k] top_labels = tf.gather(label_map, top_predictions, axis=-1) top_labels = [label.decode('utf8') for label in top_labels.numpy()] top_probs = tf.gather(probs, top_predictions, axis=-1).numpy() return top_labels[0] def perform_action_recognition(self, frame, k=1): outputs = self.model.signatures['serving_default'](image=frame) probs = tf.nn.softmax(outputs['classifier_head_1']) return self.get_top_k(probs[0], k=k) def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: if isinstance(data, str): data = json.loads(data) # Parse JSON string to dictionary frame = np.array(data.get("frame")) if frame is None: raise ValueError("'frame' is missing from the request body") if not isinstance(frame, np.ndarray): raise ValueError(f"Expected 'frame' to be a np.ndarray, but found {type(frame)}") prediction = self.perform_action_recognition(frame) return {"prediction": prediction}