from fastapi import FastAPI, UploadFile, File, Response import cv2 import numpy as np import torch import torchvision.transforms as T from PIL import Image import io app = FastAPI() # Load AI Model MiDaS midas = torch.hub.load("intel-isl/MiDaS", "MiDaS_small") midas.eval() transform = T.Compose([ T.Resize((256, 256)), T.ToTensor(), T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) @app.post("/upload/") async def upload_image(file: UploadFile = File(...)): try: start_time = time.time() image_bytes = await file.read() print(f"📷 Ảnh nhận được ({len(image_bytes)} bytes)") image = Image.open(io.BytesIO(image_bytes)).convert("RGB") print("✅ Ảnh mở thành công!") image = image.transpose(Image.FLIP_TOP_BOTTOM) image = image.transpose(Image.FLIP_LEFT_RIGHT) # Chuyển đổi ảnh sang tensor img_tensor = transform(image).unsqueeze(0) with torch.no_grad(): depth_map = midas(img_tensor).squeeze().cpu().numpy() # Chuẩn hóa depth map depth_map = cv2.normalize(depth_map, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) depth_resized = cv2.resize(depth_map, (160, 120)) # Mã hóa ảnh thành JPEG _, buffer = cv2.imencode(".jpg", depth_resized) print("✅ Depth Map đã được tạo!") end_time = time.time() start_detect_time = time.time() command = detect_path(depth_map) end_detect_time = time.time() print(f"⏳ detect_path() xử lý trong {end_detect_time - start_detect_time:.4f} giây") return {"command": command} except Exception as e: print("❌ Lỗi xử lý ảnh:", str(e)) return {"error": str(e)} def detect_path(depth_map): """Phân tích đường đi từ ảnh Depth Map""" h, w = depth_map.shape center_x = w // 2 scan_y = int(h * 0.8) # Quét dòng 80% từ trên xuống left_region = np.mean(depth_map[scan_y, :center_x]) right_region = np.mean(depth_map[scan_y, center_x:]) center_region = np.mean(depth_map[scan_y, center_x - 40:center_x + 40]) # 🟢 Cải thiện logic xử lý threshold = 100 # Ngưỡng phân biệt vật cản if center_region > threshold: return "forward" elif left_region > right_region: return "left" elif right_region > left_region: return "right" else: return "backward" if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)