##// END OF EJS Templates
Update version_rpi.txt
Update version_rpi.txt

File last commit:

r182:e1ac1491c3e7
r317:7eab564739e2 master
Show More
inference_server.py
111 lines | 2.4 KiB | text/x-python | PythonLexer
/ tools / test / inference_server.py
import numpy
import cv2
import time
from datetime import datetime
import numpy
from PIL import Image, ImageOps
import json
import traceback
import requests
link = url_rstp ='rtsp://admin:lahares2024@192.168.1.64:554/streaming/channels/1'
model_path = '/tools/models/3D_lite.tflite'
frames = []
vid = cv2.VideoCapture(link)
while 1:
try:
_, frame = vid.read()
except:
print("Error al capturar frame")
else:
if frame is None:
print("Frame capturado es None")
else:
print("recibido")
print(frame[-40:-20,-40:-20,0])
frames.append(frame)
if len(frames) == 10:
break
time.sleep(0.2)
def format_frames(frame, output_size):
"""
Redimensionar y añadir padding a una imagen obtenida de un video.
Args:
frame: Imagen que necesita ser redimensionada y rellenada.
output_size: Tamaño en píxeles de la imagen de salida.
Return:
Frame formateado con padding del tamaño especificado.
"""
# Convertir el frame a float32
frame = frame.astype(numpy.uint8)
print("process")
print(frame.min(),frame.max())
time.sleep(0.5)
# Convertir el frame de numpy array a PIL Image
frame_pil = Image.fromarray(frame)
# Redimensionar y añadir padding
frame_padded = ImageOps.pad(frame_pil, output_size, method=Image.Resampling.BILINEAR)
# Convertir de vuelta a numpy array y normalizar a [0, 1]
frame_final = numpy.array(frame_padded)
print("final")
print("min",frame_final.min(),frame_final.max())
time.sleep(0.5)
frame_final /=255.0
print("final/255")
print("min",frame_final.min(),frame_final.max())
return frame_final
output_size = (224,224)
for n,frame in enumerate(frames):
frames[n] = format_frames(frame,output_size)
frames = numpy.array(frames)[...,[2,1,0]]
frames = frames.reshape((1,frames.shape[0],
frames.shape[1],
frames.shape[2],
frames.shape[3]))
url_inference = "http://38.10.105.243:8501/v1/models/cnn_video:predict"
frames = json.dumps({'instances': frames.tolist()})
try:
resp = requests.post(url_inference,data=frames,headers={"Content-Type'": "application/json"})
except:
print(traceback.format_exc())
else:
print(resp)
print("respuesta",resp.json()['predictions'][0][0])