add: 初回の判定失敗時用のエンドポイントを作成

KakinokiKanta · KakinokiKanta · commit 60e854f4fda2 · 2025-09-05T19:12:37.000+09:00
diff --git a/.gitignore b/.gitignore
@@ -2,4 +2,5 @@
 .coverage
 coverage*
 .venv
-image_data/
+image_data/
+models/
diff --git a/pyproject.toml b/pyproject.toml
@@ -14,6 +14,9 @@ dependencies = [
   "requests>=2.32.4",
   "pillow>=11.3.0",
   "fastapi[standard]>=0.116.1",
+  "onnxruntime>=1.22.1",
+  "numpy>=2.3.2",
+  "opencv-python>=4.11.0.86",
 ]
 
 [dependency-groups]
diff --git a/src/minifig_detector.py b/src/minifig_detector.py
@@ -0,0 +1,193 @@
+"""minifig向き検出クラス.
+
+@file minifig_detector.py
+@author Hara1274
+"""
+import cv2
+import numpy as np
+import onnxruntime as ort
+import os
+import json
+
+
+class MinifigDetector:
+    """minifigの向きを検出するクラス."""
+
+    def __init__(self):
+        """検出器を初期化."""
+        self.session = None
+        self.input_name = None
+        self.labels = ["front", "back", "right", "left"]
+        self.conf_threshold = 0.25
+        self.nms_threshold = 0.45
+        self.input_size = 640
+
+        # YOLO11 ONNXモデルのパス
+        project_root = os.path.dirname(os.path.dirname(__file__))
+        model_path = os.path.join(
+            project_root, "models", "11s_100epoch_&_650imgsz_fig.onnx")
+
+        # モデルファイルが存在する場合のみonnxモデルの設定
+        if os.path.exists(model_path):
+            # 推論セッションを作成
+            self.session = ort.InferenceSession(model_path)
+            #  ONNXモデルの入力層の名前を取得
+            self.input_name = self.session.get_inputs()[0].name
+
+    def preprocess_image(self, img):
+        """推論用に画像を前処理.
+
+        Args:
+            img: 入力画像
+
+        Returns:
+            tuple: (処理後画像, スケール比, パディング情報)
+        """
+        # アスペクト比を保持するスケール計算
+        shape = img.shape[:2]  # 元画像サイズ (H, W)
+        # アスペクト比維持のスケール
+        r = min(self.input_size / shape[0], self.input_size / shape[1])
+        # スケール後サイズ (W, H)
+        new_unpad = (int(round(shape[1] * r)), int(round(shape[0] * r)))
+        img_resized = cv2.resize(
+            img, new_unpad, interpolation=cv2.INTER_LINEAR)
+
+        # 640x640にするための余白計算
+        dw = self.input_size - new_unpad[0]  # 水平余白
+        dh = self.input_size - new_unpad[1]  # 垂直余白
+        top, bottom = dh // 2, dh - dh // 2  # 上下分割
+        left, right = dw // 2, dw - dw // 2  # 左右分割
+
+        # 灰色（114）でパディングして640x640に調整
+        img_padded = cv2.copyMakeBorder(
+            img_resized, top, bottom, left, right, cv2.BORDER_CONSTANT, value=(114, 114, 114))
+
+        # YOLO入力形式に変換: HWC→CHW, [0,1]正規化, バッチ次元追加
+        img_input = img_padded.transpose(2, 0, 1).astype(
+            np.float32) / 255.0  # (3,640,640)
+        img_input = np.expand_dims(img_input, axis=0)  # (1,3,640,640)
+
+        return img_input, r, (left, top)
+
+    def postprocess(self, pred, scale, pad):
+        """推論結果を後処理.
+
+        Args:
+            pred: 推論出力 [1, 8, N] (YOLO11形式)
+            scale: レターボックスのスケール比
+            pad: パディング情報 (left, top)
+
+        Returns:
+            str: 検出された向き
+        """
+        if pred.size == 0:
+            return "front"
+
+        # YOLO11出力解析: [1,8,N] -> [8,N]（バッチ次元除去）
+        data = pred[0]
+
+        attributes = 8  # [cx,cy,w,h,class0,class1,class2,class3]
+        num_classes = attributes - 4  # 座標4つ以外はクラススコア
+        num_boxes = data.shape[1]  # 検出候補数を取得
+
+        boxes = []
+        confidences = []
+        class_ids = []
+
+        # 各検出候補を処理
+        for i in range(num_boxes):
+            max_score = -1.0
+            best_class = -1
+
+            # クラススコアの最大値とクラスIDを取得
+            for j in range(num_classes):
+                score = data[4 + j, i]
+                if score > max_score:
+                    max_score = score
+                    best_class = j
+
+            # 信頼度閾値による候補フィルタリング
+            if max_score < self.conf_threshold:
+                continue
+
+            # バウンディングボックス座標抽出（YOLO形式：中心座標＋幅高さ）
+            cx = data[0, i]  # 中心X座標（640x640空間）
+            cy = data[1, i]  # 中心Y座標（640x640空間）
+            w = data[2, i]   # 幅（640x640空間）
+            h = data[3, i]   # 高さ（640x640空間）
+
+            # レターボックス座標系から元画像座標系に逆変換
+            center_x = int((cx - pad[0]) / scale)  # パディング補正＋スケール逆変換
+            center_y = int((cy - pad[1]) / scale)
+            width = int(w / scale)
+            height = int(h / scale)
+
+            # 中心座標形式から左上座標形式（OpenCV形式）に変換
+            left = center_x - width // 2
+            top = center_y - height // 2
+
+            # NMS用データに追加
+            boxes.append([left, top, width, height])  # [x,y,w,h]形式
+            confidences.append(max_score)  # 最高クラススコア
+            class_ids.append(best_class)  # 最高スコアのクラスID
+
+        # 信頼度閾値を満たす検出がない場合
+        if not boxes:
+            return {"wasDetected": False, "direction": "NONE", "confidence": 0.0}
+
+        # Non-Maximum Suppression で重複検出を除去
+        indices = cv2.dnn.NMSBoxes(
+            boxes, confidences, self.conf_threshold, self.nms_threshold)
+
+        # NMS後に有効な検出が残っている場合
+        if len(indices) > 0:
+            best_idx = indices.flatten()[0]  # 最も信頼度の高い検出を選択
+            best_class_id = class_ids[best_idx]  # 対応するクラスID
+            best_confidence = confidences[best_idx]  # 対応する信頼度
+            direction = self.labels[best_class_id]  # ミニフィグの向き
+
+            return {
+                "wasDetected": True,
+                "direction": direction,  # "front", "back", "right", "left"
+                "confidence": float(best_confidence)  # 0.0-1.0の信頼度
+            }
+
+        return {"wasDetected": False, "direction": "NONE", "confidence": 0.0}
+
+    def detect(self, image_path: str) -> dict:
+        """minifigの向きを判定.
+
+        Args:
+            image_path (str): 画像ファイルのパス
+
+        Returns:
+            dict: 検出結果 {"wasDetected": bool, "direction": str, "confidence": float}
+        """
+        # 共通エラーレスポンス
+        error_result = {"wasDetected": False,
+                        "direction": "NONE", "confidence": 0.0}
+
+        # 入力ファイル存在チェック
+        if not os.path.exists(image_path):
+            return error_result
+
+        # モデル読み込み状態チェック
+        if self.session is None:
+            return error_result
+
+        # 画像読み込み（BGR形式）
+        img = cv2.imread(image_path)
+        if img is None:  # 読み込み失敗（不正ファイル等）
+            return error_result
+
+        # 前処理：レターボックス＋正規化
+        img_input, scale, pad = self.preprocess_image(img)
+
+        # YOLO11推論実行
+        outputs = self.session.run(None, {self.input_name: img_input})
+        pred = outputs[0]  # メイン出力 [1,8,N]
+
+        # 後処理：NMS＋結果構築
+        result = self.postprocess(pred, scale, pad)
+
+        return result
diff --git a/src/server/fastapi_server.py b/src/server/fastapi_server.py
@@ -9,10 +9,12 @@
 import socket
 import os
 import uvicorn
+import random
 
 from fastapi import FastAPI, UploadFile, File, status
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
+from ..minifig_detector import MinifigDetector
 from ..official_interface import OfficialInterface
 
 
@@ -25,6 +27,20 @@
     allow_headers=["*"],      # すべてのヘッダーを許可
 )
 
+# MinifigDetectorのインスタンスを生成
+minifig_detector = MinifigDetector()
+
+# ミニフィグ検出結果の最良結果を保持
+best_minifig_result = {
+    "image_count": 0,
+    "best_image_path": None,
+    "best_confidence": 0.0,
+    "best_direction": None
+}
+
+# アップロードされた画像パスを保存するリスト
+uploaded_image_paths = []
+
 
 @app.get("/", response_class=JSONResponse)
 def health_check() -> JSONResponse:
@@ -43,7 +59,7 @@ def health_check() -> JSONResponse:
 @app.post("/images", response_class=JSONResponse)
 def get_image(file: UploadFile = File(...)) -> JSONResponse:
     """
-    走行体から、画像ファイルを取得するための関数.
+    走行体から、画像ファイルを取得し、競技システムにアップロードする関数.
 
     Args:
         file (UploadFile): アップロードされた画像ファイル、FastAPIのFileで受け取る
@@ -60,10 +76,8 @@ def get_image(file: UploadFile = File(...)) -> JSONResponse:
     # 画像のファイル名の取得
     file_name = file.filename
 
-    # プロジェクトルートディレクトリのパスを取得（3階層上に移動）
-    project_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
     # 画像保存用ディレクトリのパスを設定
-    image_data_dir = os.path.join(project_root, 'image_data')
+    image_data_dir = os.path.join('image_data')
 
     # image_dataディレクトリが存在しない場合は作成
     os.makedirs(image_data_dir, exist_ok=True)
@@ -103,6 +117,111 @@ def get_image(file: UploadFile = File(...)) -> JSONResponse:
         )
 
 
+@app.post("/minifig/detect", response_class=JSONResponse)
+def upload_minifig_image(file: UploadFile = File(...)) -> JSONResponse:
+    """
+    走行体から、受け取った４枚のミニフィグの画像から一番正面らしいものを競技システムにアップロードする関数.
+
+    Args:
+        file (UploadFile): アップロードされた画像ファイル、FastAPIのFileで受け取る
+
+    Returns:
+        JSONResponse: 結果メッセージとステータスコード
+    """
+
+    # 画像のファイル名の取得
+    file_name = file.filename
+
+    # 画像保存用ディレクトリのパスを設定
+    image_data_dir = os.path.join("image_data")
+
+    # image_dataディレクトリが存在しない場合は作成
+    os.makedirs(image_data_dir, exist_ok=True)
+
+    # etrobocon2025-comm-device-system\image_dataに画像を保存
+    file_path = os.path.join(image_data_dir, file_name)
+    try:
+        with open(file_path, "wb") as buffer:
+            buffer.write(file.file.read())
+    except Exception as error:
+        return JSONResponse(
+            content={"error": f"Failed to save file: {str(error)}"},
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+    # MinifigDetectorで推論を実行
+    detection_result = minifig_detector.detect(file_path)
+
+    # 画像カウントを増加
+    best_minifig_result["image_count"] += 1
+
+    # アップロードされた画像パスをリストに追加
+    uploaded_image_paths.append(file_path)
+
+    # 最良画像を更新（front優先、次に信頼度優先）
+    if detection_result["wasDetected"]:
+        # 検出結果がfrontの場合
+        if detection_result["direction"] == "front":
+            # frontは既存がfront以外なら即更新、frontなら高信頼度で更新
+            if (best_minifig_result["best_direction"] != "front" or
+                    detection_result["confidence"] > best_minifig_result["best_confidence"]):
+                # それぞれの値を更新
+                best_minifig_result["best_image_path"] = file_path
+                best_minifig_result["best_confidence"] = detection_result["confidence"]
+                best_minifig_result["best_direction"] = detection_result["direction"]
+
+        # 既存の最良画像がfrontでなく、検出結果の信頼度が高い場合
+        elif (best_minifig_result["best_direction"] != "front" and
+              detection_result["confidence"] > best_minifig_result["best_confidence"]):
+            # それぞれの値を更新
+            best_minifig_result["best_image_path"] = file_path
+            best_minifig_result["best_confidence"] = detection_result["confidence"]
+            best_minifig_result["best_direction"] = detection_result["direction"]
+
+    # 4枚未満の場合
+    if best_minifig_result["image_count"] < 4:
+        return JSONResponse(
+            content={
+                "message": f"Image {best_minifig_result['image_count']} processed successfully",
+                "detection_result": detection_result,
+                "images_received": best_minifig_result["image_count"],
+                "remaining": 4 - best_minifig_result["image_count"]
+            },
+            status_code=status.HTTP_200_OK
+        )
+
+    # アップロード対象画像を決定
+    if best_minifig_result["best_image_path"]:
+        upload_image_path = best_minifig_result["best_image_path"]
+    else:
+        # ４枚すべて検出失敗時はランダムで選択
+        upload_image_path = random.choice(uploaded_image_paths)
+
+    # アップロード実行
+    upload_success = OfficialInterface.upload_snap(upload_image_path)
+
+    # リセット
+    best_minifig_result["image_count"] = 0
+    best_minifig_result["best_image_path"] = None
+    best_minifig_result["best_confidence"] = 0.0
+    best_minifig_result["best_direction"] = None
+    uploaded_image_paths.clear()
+
+    if upload_success:
+        return JSONResponse(
+            content={
+                "message": "Image uploaded successfully",
+                "imagePath": upload_image_path
+            },
+            status_code=status.HTTP_200_OK
+        )
+    else:
+        return JSONResponse(
+            content={"error": "Failed to upload image to official system"},
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+
 # ポート番号の設定
 if __name__ == "__main__":
     ip = "127.0.0.1"
@@ -120,5 +239,10 @@ def get_image(file: UploadFile = File(...)) -> JSONResponse:
         ip = connect_interface.getsockname()[0]
         connect_interface.close()
 
-    uvicorn.run("src.server.fastapi_server:app",
-                host=ip, port=8000, reload=True)
+    uvicorn.run(
+        "src.server.fastapi_server:app",
+        host=ip,
+        port=8000,
+        reload=True,
+        reload_excludes=[".venv/*"],  # ← これを追加
+    )
diff --git a/uv.lock b/uv.lock

Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,9 @@ dependencies = [`
`14`	`14`	`"requests>=2.32.4",`
`15`	`15`	`"pillow>=11.3.0",`
`16`	`16`	`"fastapi[standard]>=0.116.1",`
	`17`	`+ "onnxruntime>=1.22.1",`
	`18`	`+ "numpy>=2.3.2",`
	`19`	`+ "opencv-python>=4.11.0.86",`
`17`	`20`	`]`
`18`	`21`
`19`	`22`	`[dependency-groups]`