main.py

# main.py
import io
import os
from ctypes import c_int, pointer, string_at
from datetime import datetime
from typing import List

import cv2
import numpy as np
from fastapi import FastAPI, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from paddleocr import PaddleOCR
from PIL import Image, ImageEnhance

import dds
from app import (
    DEFAULT_THRESHOLDS,
    arrange_data,
    format_dds_data,
    get_player_regions,
    validate_deal,
)
from identify_cards import (
    SUIT_TEMPLATE_PATH,
    determine_and_correct_orientation,
    find_rank_candidates,
    get_not_white_mask,
    get_suit_from_image_rules,
    load_suit_templates,
    save_img_with_rect,
)
from utils import convert2pbn, convert2pbn_txt, is_text_valid

# from app import arrange_data, run_dds_analysis # Gradioのapp.pyからロジックを移植

# FastAPIインスタンスを作成
app = FastAPI()
origins = [
    "http://localhost",
    "http://localhost:5173",  # Default URL for Vite React dev server
    "https://board-recognizer-30ib6veo9-wai572s-projects.vercel.app",  # Your deployed frontend
    "https://board-recognizer.vercel.app",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,  # Specifies the allowed origins
    allow_credentials=True,  # Allows cookies to be included in requests
    allow_methods=["*"],  # Allows all methods (GET, POST, etc.)
    allow_headers=["*"],  # Allows all headers
)

# AIモデルとテンプレートを起動時に読み込む
# trocr_pipeline = None  # load_model()のロジックをここに
reader = None
suit_templates = None


@app.on_event("startup")
def load_ocr_model():
    """

    アプリケーション起動時に一度だけEasyOCRのモデルを読み込み、

    グローバル変数readerに格納する。

    """
    global reader

    # 使用する言語と、モデルの保存先ディレクトリを指定してReaderを初期化
    reader = PaddleOCR(
        lang="en",
        use_doc_orientation_classify=False,
        use_doc_unwarping=False,
        use_textline_orientation=False,
    )
    print(f"PaddleOCR model loaded successfully.")


# @app.on_event("startup")
# def load_dependencies():
#     global trocr_pipeline, suit_templates
#     # TrOCRモデルをロード (Gradioのload_model関数を参考)
#     from transformers import pipeline

#     try:
#         print("Loading TrOCR model...")
#         trocr_pipeline = pipeline(
#             "image-to-text", model="microsoft/trocr-small-printed"
#         )
#         print("TrOCR model loaded.")
#     except Exception as e:
#         print(f"Failed to load TrOCR model: {e}")
#         trocr_pipeline = None

#     # スートテンプレートをロード
#     suit_templates = load_suit_templates("templates/suits/")


@app.post("/analyze/")
async def analyze_image(image_paths: list[UploadFile]):
    # print(request)
    # print(list(request.keys()))
    # image_paths = request["image_paths"]
    print(image_paths)
    progress = lambda x, desc: print(x, desc)
    global reader
    # global trocr_pipeline
    # # モデルが読み込まれているか確認
    if reader is None:
        print(
            "AIモデルがまだ読み込まれていません。しばらく待ってから再度お試しください。"
        )
        # 空の更新を返すことで、UIの状態を変えずに処理を終了
        return

    all_results = []
    num_total_files = len(image_paths)

    progress(0, desc="テンプレート画像読み込み中...")
    suit_templates = load_suit_templates(SUIT_TEMPLATE_PATH)
    if not suit_templates:
        raise (
            f"エラー: {SUIT_TEMPLATE_PATH} フォルダにスートのテンプレート画像が見つかりません。"
        )

    try:
        all_candidates_global = []
        processed_files_info = []
        # image_objects = {}

        for i, image_path in enumerate(image_paths):
            progress(
                (i + 1) / num_total_files * 0.15,
                desc="ステージ1/3: 文字候補を検出中...",
            )
            filename = os.path.basename(image_path.filename)
            progress(
                (i + 1) / num_total_files * 0.3,
                f"分析中 ({i+1}/{num_total_files}): {filename}",
            )

            try:
                # ファイルをバイナリモードで安全に読み込む
                # file_bytes = np.asarray(bytearray(image_path))
                print(image_path.file)
                # file_bytes = np.asarray(bytearray(image_path.file))
                # with open(image_path.filename, "rb") as f:
                #     # バイトデータをNumPy配列に変換
                #     file_bytes = np.asarray(
                #         bytearray(f.read()), dtype=np.uint8
                #     )
                # NumPy配列（メモリ上のデータ）から画像をデコード
                file_bytes = await image_path.read()
                # print("file", file_bytes)
                file_array = np.asarray(bytearray(file_bytes), dtype=np.uint8)
                image = cv2.imdecode(file_array, cv2.IMREAD_COLOR)
                # image = image_path.file

                if image is None:
                    raise (
                        "OpenCVが画像をデコードできませんでした。ファイルが破損しているか、非対応の形式の可能性があります。"
                    )
                # image_objects[filename] = image
            except Exception as e:
                # ファイル読み込み自体のエラーをキャッチ
                print(e)
                all_results.append(
                    {"filename": filename, "error": f"画像読み込みエラー: {e}"}
                )
                # image_objects[filename] = None
                continue

            # box = find_center_box(image)
            print("detect board")
            rotated_image, box, scale = determine_and_correct_orientation(
                image, lambda msg: print(msg)
            )
            if box is None:
                all_results.append(
                    {"filename": filename, "error": "中央ボードの検出に失敗"}
                )
                continue
            print(box)
            save_img_with_rect("debug_rotated.jpg", rotated_image, [box])

            MARGIN = 200
            player_regions = get_player_regions(rotated_image, box, MARGIN)

            for player, region in player_regions.items():
                candidates = find_rank_candidates(
                    region, suit_templates, player, scale
                )
                for cand in candidates:
                    cand["filename"] = filename
                    cand["player"] = player
                    all_candidates_global.append(cand)

            processed_files_info.append({"filename": filename, "error": None})
        progress(
            0.4, desc="ステージ2/3: 文字認識を実行中... (時間がかかります)"
        )

        if not all_candidates_global:
            progress(1, desc="認識する文字候補がありませんでした。")
            print("認識する文字候補がありませんでした。")
            return JSONResponse(
                content=all_results
            )  # エラーがあった画像の結果だけを返す

        try:
            ocr_results = []
            for candidate in all_candidates_global:
                # img = Image.fromarray(
                #     cv2.cvtColor(candidate["img"], cv2.COLOR_BGR2RGB)
                # )
                img = candidate["img"]
                pil_img = Image.fromarray(img)
                enhancer = ImageEnhance.Contrast(pil_img)
                im_con = enhancer.enhance(2.0)
                np_img = np.asarray(im_con)
                # text_mask = get_not_white_mask(img)
                # masked_img = cv2.bitwise_and(img, img, mask=text_mask)
                # result = reader.readtext(candidate["img"])
                result = reader.predict(np_img)
                print(result)
                if len(result) > 0:
                    ocr_results.append(result)
            # candidates_pil_images = [
            #     Image.fromarray(cv2.cvtColor(c["img"], cv2.COLOR_BGR2RGB))
            #     for c in all_candidates_global
            # ]
            # ocr_results = trocr_pipeline(candidates_pil_images)
        except Exception as e:
            print(f"OCR処理中にエラーが発生しました: {e}")

        # --- ステージ3: 結果の仕分けと最終的なカードの特定 ---
        progress(0.9, desc="ステージ3/3: 認識結果を仕分け中...")

        # print([result[0]["generated_text"] for result in ocr_results])

        raw_data = []
        # blacks = []
        # reds = []
        for i, result in enumerate(ocr_results):
            # text = result[0]["generated_text"].upper().strip()
            text = result[0]["rec_texts"]
            if len(text) > 0:
                text = text[0]
            else:
                text = ""
            print(text, is_text_valid(text))

            text = is_text_valid(text)
            if text is not None:
                candidate_info = all_candidates_global[i]
                print(
                    f"--- 診断中: ランク '{text}' of {candidate_info['player']} at {candidate_info['pos']} with thick:{candidate_info['thickness']} ---"
                )
                color_name, avg_lab = get_suit_from_image_rules(
                    candidate_info["no_pad"], DEFAULT_THRESHOLDS
                )
                print(color_name)
                if color_name == "mark":
                    continue
                candidate_info["avg_lab"] = avg_lab
                candidate_info["color"] = color_name
                candidate_info["name"] = text
                raw_data.append(candidate_info)

        # print("\r\n".join(blacks))
        # print("\r\n".join(reds))

        all_results = arrange_data(raw_data)
        # pbn_content = convert2pbn(all_results)
        # pbn_filename = f"analysis_{datetime.now().strftime('%Y%m%d')}.pbn"
        # if processed_files_info:
        #     last_result = {"filename": processed_files_info[0]["filename"], 1ands": all_results[0][1ands"]}

        # if all_results:
        #     # ダウンロード用にPBNコンテンツを値として設定し、表示状態にする
        #     export_update = gr.update(interactive=True)
        # else:
        #     export_update = gr.update(interactive=False)
        # final_result = all_results[0]["hands"]
        # filenames = [os.path.basename(p) for p in image_paths]
        # dropdown_update = gr.update(
        #     choices=filenames, value=filenames[0], interactive=True, open=True
        # )

        dataframes = run_dds_analysis(all_results)
        for result in all_results:
            if result["filename"] in dataframes.keys():
                result["dds"] = dataframes[result["filename"]]

        return JSONResponse(content=all_results)

    except Exception as e:
        raise (f"致命的なエラー: {e}")


def run_dds_analysis(all_results_state):
    """ダブルダミー分析を実行する"""
    valid_deals = []
    for result in all_results_state:
        if "hands" in result:
            print(result["hands"])
            is_valid, _ = validate_deal(result["hands"])
            if is_valid:
                valid_deals.append(result)

    if len(valid_deals) == 0:
        raise ("分析不可", "分析対象となる正常なディールがありません。")

    # self.status_var.set(f"{len(valid_deals)}件のディールを分析中...")

    try:
        deals = dds.ddTableDealsPBN()
        deals.noOfTables = len(valid_deals)
        for i, result in enumerate(valid_deals):
            pbn_deal_string = convert2pbn_txt(result["hands"], "N")
            print(pbn_deal_string)

            # table_deal_pbn = dds.ddTableDealPBN()
            # table_deal_pbn.cards = pbn_deal_string.encode("utf-8")

            deals.deals[i].cards = pbn_deal_string.encode("utf-8")

        dds.SetMaxThreads(0)
        table_res = dds.ddTablesRes()
        per_res = dds.allParResults()
        # table_res_pointer = pointer(table_res)
        res = dds.CalcAllTablesPBN(
            pointer(deals),
            0,
            (c_int * 5)(0, 0, 0, 0, 0),
            pointer(table_res),
            pointer(per_res),
        )
        print("dds")

        if res != dds.RETURN_NO_FAULT:
            err_char_p = dds.ErrorMessage(res)
            err_string = (
                string_at(err_char_p).decode("utf-8")
                if err_char_p
                else "Unknown error"
            )
            raise RuntimeError(
                f"DDS Solver failed with code: {res} ({err_string})"
            )
        print("dds")

        filenames = [d["filename"] for d in valid_deals]
        dataframes = {}
        for i, filename in enumerate(filenames):
            headers, rows = format_dds_data(table_res.results[i].resTable)
            print(rows)
            dataframes[filename] = rows

        return dataframes

        # 3. 結果を新しいウィンドウで表示

    except Exception as e:
        raise (f"DDS分析エラー: 分析中にエラーが発生しました:\n{e}")
        # self.status_var.set("DDS分析中にエラーが発生しました。")