import pickle from minisom import MiniSom import numpy as np from PIL import Image import math import requests from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import os import cv2 from scipy.ndimage import median_filter from scipy.signal import convolve2d class InputData(BaseModel): data: str # Lista de características numéricas (flotantes) app = FastAPI() # Función para construir el modelo manualmente def build_model(): with open('somlucuma.pkl', 'rb') as fid: som_cargado = pickle.load(fid) MM = np.loadtxt('matrizMM.txt', delimiter=" ") return som_cargado,MM som,MM = build_model() # Construir el modelo al iniciar la aplicación def sobel(patron): gx = np.array([[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]], dtype=np.float32) gy = np.array([[1, 2, 1], [0, 0, 0], [-1, -2, -1]], dtype=np.float32) Gx = convolve2d(patron, gx, mode='valid') Gy = convolve2d(patron, gy, mode='valid') return Gx, Gy def medfilt2(G, d=3): return median_filter(G, size=d) def orientacion(patron, w): Gx, Gy = sobel(patron) Gx = medfilt2(Gx) Gy = medfilt2(Gy) m, n = Gx.shape mOrientaciones = np.zeros((m // w, n // w), dtype=np.float32) for i in range(m // w): for j in range(n // w): Gx_patch = Gx[i*w:(i+1)*w, j*w:(j+1)*w] Gy_patch = Gy[i*w:(i+1)*w, j*w:(j+1)*w] YY = np.sum(2 * Gx_patch * Gy_patch) XX = np.sum(Gx_patch**2 - Gy_patch**2) mOrientaciones[i, j] = (0.5 * np.arctan2(YY, XX) + np.pi / 2.0) * (18.0 / np.pi) return mOrientaciones def redimensionar(img, h, v): return cv2.resize(img, (h, v), interpolation=cv2.INTER_AREA) def testeo(som,archivo): Xtest = redimensionar(cv2.imread(archivo),256,256) Xtest = np.array(Xtest) Xtest = cv2.cvtColor(Xtest, cv2.COLOR_BGR2GRAY) #print(Xtest) #np.savetxt('matrizTT.txt', Xtest, delimiter=",") orientaciones = orientacion(Xtest, w=14) Xtest = Xtest.astype('float32') / 255.0 orientaciones = orientaciones.reshape(-1) #np.savetxt('matrizXX.txt', orientaciones, delimiter=",") #orientaciones = np.concatenate([orientaciones.ravel()]) return som.winner(orientaciones) @app.post("/predict/") async def predict(data: InputData): print(f"Data: {data}") global som global MM try: # Descargar la imagen desde la URL respuesta = requests.get(data.data) imagen_temporal = "imagen_temporal.jpg" with open(imagen_temporal, 'wb') as f: f.write(respuesta.content) if respuesta.status_code != 200: raise HTTPException(status_code=400, detail="No se pudo descargar la imagen") w = testeo(som,imagen_temporal) #img = cv2.imread(imagen_temporal, cv2.IMREAD_GRAYSCALE) #resized = cv2.resize(img, (256, 256), interpolation=cv2.INTER_AREA) #imagen_normalizada = resized.astype(np.float32) / 255.0 #cv2.imwrite(imagen_temporal, imagen_normalizada) # Convertir la imagen descargada a un array de NumPy #B = representativo(imagen_temporal) #w = som.winner(B) os.remove(imagen_temporal) #imagen = Image.open(BytesIO(respuesta.content)) #imagen_array = np.array(imagen) # Aquí puedes hacer cualquier cosa que necesites con la imagen # Por ejemplo, procesarla o realizar la predicción # Supongamos que aquí se realiza la predicción con la imagen # Retornar el resultado de la predicción (puedes adaptar esto según tus necesidades) return {"prediction": MM[w]} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ''' # Ruta de predicción @app.post("/predict/") async def predict(data: InputData): print(f"Data: {data}") global som global MM try: # Convertir la lista de entrada a un array de NumPy para la predicción input_data = np.array(data.data).reshape( 1, -1 ) # Asumiendo que la entrada debe ser de forma (1, num_features) #input_data = [float(f) for f in input_data] w = som.winner(input_data) prediction = MM[w] return {"prediction": prediction.tolist()} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) '''