File size: 2,258 Bytes
cc770a8 a7ff454 cc770a8 ee6350b cc770a8 a7ff454 7cae191 a7ff454 ee6350b a7ff454 cc770a8 a7ff454 cc770a8 a7ff454 2188260 a7ff454 cc770a8 a7ff454 cc770a8 a7ff454 cc770a8 a7ff454 |
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 |
import cv2
import numpy as np
import tensorflow as tf
import time
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 preprocess_frame(self, frame):
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = cv2.resize(frame, (224, 224))
frame = frame.astype('float32') / 255.0
frame = np.expand_dims(frame, axis=0)
return frame
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, url, k=1):
cap = cv2.VideoCapture(url)
start_time = time.time()
while True:
ret, frame = cap.read()
if not ret:
break
preprocessed_frame = self.preprocess_frame(frame)
outputs = self.model.signatures['serving_default'](image=preprocessed_frame[tf.newaxis])
probs = tf.nn.softmax(outputs['classifier_head_1'])
current_time = time.time() - start_time
m, s = divmod(current_time, 60)
h, m = divmod(m, 60)
ip_address = url.split("/")[-1]
output = {
"class": self.get_top_k(probs[0], k=k),
"elapsed_time": f"{int(h):02d}:{int(m):02d}:{int(s):02d}",
"location": random.choice(self.locations),
"ip_address": ip_address
}
yield output
cap.release()
def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
url = data.get("url")
if url is None:
raise ValueError("'url' is missing from the request body")
if not isinstance(url, str):
raise ValueError(f"Expected 'url' to be a str, but found {type(url)}")
outputs = []
for output in self.perform_action_recognition(url):
outputs.append(output)
return outputs |