更新居中画图

Test/model_test01.py

@@ -55,7 +55,7 @@ def predict_single_image(config_params: dict,
 
 
 if __name__ == '__main__':
-    big_img_path = r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\250805_pokemon_0001.jpg"
+    big_img_path = r"C:\Code\ML\Project\untitled10\WebNetwork\Gradio\upload_downloda\temp\front_0_1.jpg"
 
     config = settings.CARD_MODELS_CONFIG
     # predict_single_image(config['pokemon_front_inner_box'],
@@ -67,9 +67,9 @@ if __name__ == '__main__':
     #                      output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\back_inner")
     #
     #
-    # predict_single_image(config['outer_box'],
-    #                      big_img_path,
-    #                      output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\outer")
+    predict_single_image(config['outer_box'],
+                         big_img_path,
+                         output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\outer")
 
     # predict_single_image(config['pokemon_back_corner_defect'],
     #                      img_path=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\img_2.png",
@@ -79,9 +79,9 @@ if __name__ == '__main__':
     #                      img_path=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\250805_pokemon_0001_grid_r3_c4.jpg",
     #                      output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\face_no_reflect")
 
-    result = predict_single_image(config['pokemon_front_corner_no_reflect_defect'],
-                                  img_path=r"C:\Users\wow38\Downloads\_250829_1656_最新模型汇总\_250829_1817_宝可梦非闪光卡正面边角模型\pth_and_images\images\250730_pokemon_0031_bottom_grid_r0_c5.jpg",
-                                  output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\corner_no_reflect")
+    # result = predict_single_image(config['pokemon_front_corner_no_reflect_defect'],
+    #                               img_path=r"C:\Users\wow38\Downloads\_250829_1656_最新模型汇总\_250829_1817_宝可梦非闪光卡正面边角模型\pth_and_images\images\250730_pokemon_0031_bottom_grid_r0_c5.jpg",
+    #                               output_dir=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\corner_no_reflect")
 
     # result = predict_single_image(config['pokemon_front_face_no_reflect_defect'],
     #                      img_path=r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\temp\250805_pokemon_0001_grid_r3_c4.jpg",

Test/test01.py

@@ -1,35 +1,77 @@
-from multiprocessing import Process, Queue, Lock
+import asyncio
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
 import time
+import logging
 
-lock = Lock()
+# 配置日志
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
+# 创建一个 asyncio.Event 来作为关闭信号
+shutdown_event = asyncio.Event()
 
-def f(q, a):
-    lock.acquire()
-    if a == 3:
-        time.sleep(2)
 
-        q.put(666)
-        print('input')
-    if a == 4:
-        data = q.get()
-        print(data)
-    print(a)
+async def my_background_worker(name: str):
+    """
+    一个模拟的后台工作任务，它会持续运行直到收到关闭信号。
+    """
+    logging.info(f"Worker '{name}' starting.")
+    count = 0
+    while not shutdown_event.is_set():
+        try:
+            # 你的业务逻辑
+            logging.info(f"Worker '{name}' is running, count: {count}")
+            count += 1
 
-    lock.release()
+            # 使用 asyncio.sleep 而不是 time.sleep，避免阻塞事件循环
+            await asyncio.sleep(2)
+        except asyncio.CancelledError:
+            # 当任务被取消时，正常退出循环
+            logging.warning(f"Worker '{name}' received cancellation request. Stopping.")
+            break
+    logging.info(f"Worker '{name}' has stopped gracefully.")
 
 
-if __name__ == '__main__':
-    q = Queue()
-    p_list = []
-    for i in range(8):
-        p = Process(target=f, args=(q, i,))
-        p_list.append(p)
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logging.info("Application startup...")
+
+    # 在应用启动时，创建并启动后台任务
+    # 将 task 存储在 app.state 中，以便在关闭时可以访问它
+    app.state.worker_task = None
+    yield  # 应用在此处运行
+
+    logging.info("Application shutdown...")
+
+    # 1. 发送关闭信号
+    logging.info("Signaling worker to shut down.")
+    shutdown_event.set()
+
+    # 2. 等待后台任务完成
+    try:
+        # 给一个超时时间，防止无限期等待
+        await asyncio.wait_for(app.state.worker_task, timeout=5.0)
+        logging.info("Worker task has been awaited.")
+    except asyncio.TimeoutError:
+        logging.error("Worker task did not finish in time, cancelling.")
+        app.state.worker_task.cancel()  # 如果超时，强制取消
+        # 等待取消操作完成
+        await asyncio.gather(app.state.worker_task, return_exceptions=True)
+    except Exception as e:
+        logging.error(f"An error occurred during worker shutdown: {e}")
+
 
-    for p in p_list:
-        p.start()
+app = FastAPI(lifespan=lifespan)
 
-    print('-=------')
 
-    for p in p_list:
-        p.join()
+@app.get("/")
+async def root():
+    app.state.worker_task = asyncio.create_task(my_background_worker("Worker-1"))
+
+    return {"message": "Hello World"}
+
+
+if __name__ == '__main__':
+    import uvicorn
+
+    uvicorn.run(app)

Test/test02.py

@@ -0,0 +1,16 @@
+import cv2
+import numpy as np
+from pathlib import Path
+import os
+
+
+img: np.ndarray = cv2.imread(r"")
+print(img.shape)
+
+path = "./a/b/33.jpg"
+path_obj = Path(path)
+path_obj.parent.mkdir(parents=True, exist_ok=True)
+
+success = cv2.imwrite(path, img)
+
+print(success)

app/api/card_inference.py

@@ -5,9 +5,11 @@ from enum import Enum
 from ..core.config import settings
 from app.services.card_service import CardInferenceService, card_service
 from app.services.defect_service import DefectInferenceService
-from app.core.logger import logger
+from app.core.logger import get_logger
 import json
 
+logger = get_logger(__name__)
+
 router = APIRouter()
 
 model_names = list(settings.CARD_MODELS_CONFIG.keys())
@@ -69,13 +71,4 @@ async def card_model_inference(
         raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
         logger.error(e)
-        raise HTTPException(status_code=500, detail=f"服务器内部错误: {e}")
-
-
-@router.post("/mock_query")
-async def mock_query(img_id: int):
-    # json_data = {"img_id": img_id}
-    with open("_temp_work/mock_result.json", "r") as f:
-        json_data = json.load(f)
-
-    return json_data
+        raise HTTPException(status_code=500, detail=f"服务器内部错误: {e}")

app/api/score_inference.py

@@ -4,8 +4,10 @@ from fastapi.concurrency import run_in_threadpool
 from enum import Enum
 from ..core.config import settings
 from app.services.score_service import ScoreService
-from app.core.logger import logger
 import json
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
 
 router = APIRouter()
 

app/core/config.py

@@ -146,7 +146,6 @@ class Settings:
 
 settings = Settings()
 print(f"项目根目录: {settings.BASE_PATH}")
-print(f"数据存储目录: {settings.DATA_DIR}")
 
 # DefectType = Enum("InferenceType", {name: name for name in settings.DEFECT_TYPE})
 # print()

app/core/logger.py

@@ -1,15 +1,53 @@
 import logging
+import sys
 
-logging.basicConfig(
-    level=logging.INFO,  # 生产环境通常设置为 INFO 或 WARNING
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-    filename='app.log',  # 指定日志输出到文件 app.log
-    filemode='w'
-)
-logger = logging.getLogger(__name__)
+# 定义一个全局的日志格式
+LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 
 
-def get_logger(name):
-    # 获取一个日志器实例，通常以模块名命名
-    logger = logging.getLogger(name)
-    return logger
+def setup_logging():
+    """
+    配置日志系统，使其同时输出到文件和控制台。
+    这个函数应该在应用程序启动时只调用一次。
+    """
+    # 获取根日志记录器
+    root_logger = logging.getLogger()
+    root_logger.setLevel(logging.INFO)  # 设置根日志记录器的级别
+
+    # 如果已经有处理器，先清空，防止重复添加
+    if root_logger.hasHandlers():
+        root_logger.handlers.clear()
+
+    # 创建一个通用的格式化器
+    formatter = logging.Formatter(LOG_FORMAT)
+
+    # 1. 创建控制台处理器 (StreamHandler)
+    #    - 将日志输出到标准输出（您的终端）
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(formatter)
+    root_logger.addHandler(console_handler)
+
+    # 2. 创建文件处理器 (FileHandler)
+    #    - 将日志写入文件 app.log
+    #    - mode='w' 表示每次启动都覆盖旧日志
+    #    - encoding='utf-8' 确保正确处理中文字符
+    file_handler = logging.FileHandler('app.log', mode='w', encoding='utf-8')
+    file_handler.setFormatter(formatter)
+    root_logger.addHandler(file_handler)
+
+    # 配置完成后，可以记录一条消息来确认
+    logging.info("日志系统已成功配置，将同时输出到控制台和 app.log 文件。")
+
+
+def get_logger(name: str) -> logging.Logger:
+    """
+    获取一个指定名称的日志记录器实例。
+    假设 setup_logging() 已经在此之前被调用。
+    """
+    return logging.getLogger(name)
+
+if __name__ == '__main__':
+    # setup_logging()
+    logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
+    logger = get_logger(__name__)
+    logger.info("1234")

app/main.py

@@ -6,15 +6,19 @@ from app.api.card_inference import router as card_inference_router
 from app.api.score_inference import router as score_inference_router
 import os
 
+from .core.logger import setup_logging, get_logger
 from .core.config import settings
 
+setup_logging()
+# 获取一个用于 main 模块的日志记录器
+logger = get_logger(__name__)
+
 
 @asynccontextmanager
 async def lifespan(main_app: FastAPI):
     print("--- 应用启动 ---")
     # --- 文件和目录准备 ---
     os.makedirs(settings.TEMP_WORK_DIR, exist_ok=True)
-    os.makedirs(settings.DATA_DIR, exist_ok=True)
 
     # --- 模型加载 ---
     load_models()
@@ -27,4 +31,4 @@ async def lifespan(main_app: FastAPI):
 app = FastAPI(title="卡片框和缺陷检测服务", lifespan=lifespan)
 
 app.include_router(card_inference_router, prefix=settings.API_Inference_prefix)
-app.include_router(score_inference_router, prefix=settings.API_Score_prefix)
+app.include_router(score_inference_router, prefix=settings.API_Score_prefix)

app/services/card_service.py

@@ -2,7 +2,9 @@
 import cv2
 import numpy as np
 from ..core.model_loader import get_predictor
+from app.core.logger import get_logger
 
+logger = get_logger(__name__)
 
 class CardInferenceService:
     def predict(self, inference_type: str, image_bytes: bytes) -> dict:

app/services/defect_service.py

@@ -4,6 +4,7 @@ from ..core.model_loader import get_predictor
 from app.utils.defect_inference.CardDefectAggregator import CardDefectAggregator
 from app.utils.defect_inference.arean_anylize_draw import DefectProcessor, DrawingParams
 from app.utils.defect_inference.AnalyzeCenter import analyze_centering_rotated
+from app.utils.defect_inference.DrawCenterInfo import draw_boxes_and_center_info
 from app.utils.defect_inference.ClassifyEdgeCorner import ClassifyEdgeCorner
 from app.utils.json_data_formate import formate_center_data, formate_face_data
 from app.core.config import settings
@@ -149,6 +150,10 @@ class DefectInferenceService:
             logger.info("格式化居中数据")
             center_result = formate_center_data(center_result, inner_result, outer_result)
 
+            draw_img = draw_boxes_and_center_info(img_bgr, center_result)
+            temp_center_img_path = settings.TEMP_WORK_DIR / f'{inference_type}-center_result.jpg'
+            cv2.imwrite(temp_center_img_path, draw_img)
+
             temp_center_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-center_result.json'
             with open(temp_center_json_path, 'w', encoding='utf-8') as f:
                 json.dump(center_result, f, ensure_ascii=False, indent=2)

app/utils/defect_inference/AnalyzeCenter.py

@@ -192,6 +192,7 @@ def analyze_centering_rotated(inner_points: Union[str, List], outer_points: Unio
 #     inner_pts = get_points_from_file(inner_file_path)
 #     print(inner_pts)
 #     print(type(inner_pts))
-# result = analyze_centering_rotated(inner_file_path, outer_file_path)
-# print(result)
-# draw_rotated_bounding_boxes(img_path, inner_file_path, outer_file_path)
+#
+#     result = analyze_centering_rotated(inner_file_path, outer_file_path)
+#     print(result)
+#     draw_rotated_bounding_boxes(img_path, inner_file_path, outer_file_path)

app/utils/defect_inference/DrawCenterInfo.py

@@ -0,0 +1,68 @@
+import cv2
+import json
+import numpy as np
+from typing import Union
+
+def draw_boxes_and_center_info(image:Union[str, np.ndarray], json_data:dict):
+    """
+    将JSON数据中的内外框和居中信息绘制到图片上。
+
+    Args:
+        image (str): 原始图片文件的路径。
+        json_data (dict): 包含内外框检测结果和居中计算结果的字典。
+    """
+    try:
+        if isinstance(image, str):
+            img = cv2.imread(image)
+        else:
+            img = image
+        if img is None:
+            raise FileNotFoundError(f"无法读取图片: {image}")
+
+        h, w, _ = img.shape  # 获取图像的宽度和高度
+
+        # --------------------- 绘制内框 (inner_box) ---------------------
+        if "inner_box" in json_data["box_result"] and json_data["box_result"]["inner_box"]["num"] > 0:
+            for shape_info in json_data["box_result"]["inner_box"]["shapes"]:
+                points = np.array(shape_info["points"], dtype=np.int32)
+                # 绘制多边形
+                cv2.polylines(img, [points], isClosed=True, color=(0, 230, 130), thickness=2)  # 绿色
+                # 标记内框的标签
+                cv2.putText(img, "Inner Box", tuple(points[0]), cv2.FONT_HERSHEY_SIMPLEX, 3, (0, 255, 0), 5,
+                            cv2.LINE_AA)
+
+        # --------------------- 绘制外框 (outer_box) ---------------------
+        if "outer_box" in json_data["box_result"] and json_data["box_result"]["outer_box"]["num"] > 0:
+            for shape_info in json_data["box_result"]["outer_box"]["shapes"]:
+                points = np.array(shape_info["points"], dtype=np.int32)
+                # 绘制多边形
+                cv2.polylines(img, [points], isClosed=True, color=(120, 130, 250), thickness=2)
+                # 标记外框的标签
+                cv2.putText(img, "Outer Box", tuple(points[0] + [0, -50]), cv2.FONT_HERSHEY_SIMPLEX, 3, (0, 0, 255), 5,
+                            cv2.LINE_AA)
+        # --------------------- 显示结果 ---------------------
+
+        # 如果需要显示图片
+        # cv2.imshow("Detection Result", img)
+        # cv2.waitKey(0)
+        # cv2.destroyAllWindows()
+        return img
+
+
+    except FileNotFoundError as e:
+        print(e)
+    except Exception as e:
+        print(f"处理图片时发生错误: {e}")
+
+
+# if __name__ == '__main__':
+#     image_file = r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\_temp_work\pokemon_front_corner_no_reflect_defect-corner_result.jpg"  # 请替换为你的实际图片文件名
+#
+#     # 从JSON文件中加载数据
+#     json_file_path = r"C:\Code\ML\Project\CheckCardBoxAndDefectServer\_temp_work\pokemon_front_card_center-center_result.json"
+#     with open(json_file_path, 'r') as f:
+#         data = json.load(f)
+#
+#     # 调用函数绘制
+#     img = draw_boxes_and_center_info(image_file, data)
+#     cv2.imwrite("test01.jpg", img)