1.修改更通用的类名\n2.实现检测消防区域是否侵占接口

This commit is contained in:
haotian 2025-08-20 09:47:28 +08:00
parent 106eebd089
commit be6f75b970
5 changed files with 145 additions and 69 deletions

View File

@ -1,21 +1,26 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.services.imageServices import ImageService
from app.schemas.image import ImageBase
from app.schemas.ocr import OCRRequest, OCRResponse
from app.util.responseHttp import ResponseUtil
from app.util.baiduOCR import BaiduOCR
import base64
import io
from PIL import Image
import os
import uuid
from app.core.database import get_db
from app.services.imageServices import ImageService
from app.schemas.image import ImageBase
from app.schemas.ocr import ImageBase64Request
from app.util.responseHttp import ResponseUtil
from app.util.baiduOCR import BaiduOCR
from app.util.yolov8Obj import Yolov8Obj
# from app.crud.event import event
# from app.schemas.event import EventList, EventDetail, EventUpdate, EventQuery, TestEvent
baiduOCR = BaiduOCR()
yolov8Obj = Yolov8Obj()
router = APIRouter(prefix="/api/v1", tags=["ocr"])
@ -54,7 +59,7 @@ async def test_ocr(
@router.post("/ocr_from_base64")
async def ocr_from_base64(request: OCRRequest):
async def ocr_from_base64(request: ImageBase64Request):
"""
从base64图片数据进行OCR识别
"""
@ -86,6 +91,36 @@ async def ocr_from_base64(request: OCRRequest):
# 删除临时文件
os.remove(temp_path)
return ResponseUtil.success(msg="OCR识别成功", data=result[0])
return ResponseUtil.success(msg="OCR识别成功", data=result)
except Exception as e:
return ResponseUtil.error(msg=f"OCR识别失败: {str(e)}", data=None)
@router.post("/detect_from_base64_0")
async def ocr_from_base64(request: ImageBase64Request):
""" 从吧色图64图片进行目标检测, 检测是否侵占消防区域"""
try:
image_base64 = request.image_base64
if image_base64.startswith('data:image'):
image_base64 = image_base64.split(',')[1]
image_data = base64.b64decode(image_base64)
image = Image.open(io.BytesIO(image_data))
temp_dir = "./tmp/detect_images"
os.makedirs(temp_dir, exist_ok=True)
temp_filename = f"{uuid.uuid4()}.{request.image_type or 'jpg'}"
temp_path = os.path.join(temp_dir, temp_filename)
image.save(temp_path)
result = yolov8Obj.detect(temp_path)
os.remove(temp_path)
return ResponseUtil.success(msg="OCR识别成功", data=result)
except Exception as e:
return ResponseUtil.error(msg=f"j检测是否侵占消防区域失败: {str(e)}", data=None)

View File

@ -1,11 +1,11 @@
from pydantic import BaseModel
from typing import Optional
class OCRRequest(BaseModel):
class ImageBase64Request(BaseModel):
image_base64: str
image_type: Optional[str] = "jpg" # 图片类型,如 jpg, png 等
class OCRResponse(BaseModel):
success: bool
message: str
data: Optional[list] = None
# class OCRResponse(BaseModel):
# success: bool
# message: str
# data: Optional[list] = None

View File

@ -18,7 +18,7 @@ class BaiduOCR:
def ocr(self, image_path:str):
result = self.model.predict(image_path)
result = self.parse_result(result)
return result
return result[0]
def parse_result(self, result):
text_list = []

View File

@ -0,0 +1,96 @@
import base64
import requests
import json
def image_to_base64(image_path):
"""
将图片文件转换为base64编码
"""
with open(image_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
return encoded_string
def test_ocr_api(image_path, api_url="http://10.0.0.202:12342/api/v1/ocr_from_base64"):
"""
测试OCR API接口
"""
# 将图片转换为base64
image_base64 = image_to_base64(image_path)
with open("base64.txt", "w") as f:
f.write(repr(image_base64))
# 准备请求数据
payload = {
"image_base64": image_base64,
"image_type": "jpg" # 根据实际图片类型修改
}
# 发送POST请求
headers = {
"Content-Type": "application/json"
}
try:
response = requests.post(api_url, data=json.dumps(payload), headers=headers)
if response.status_code == 200:
result = response.json()
print("OCR识别结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
else:
print(f"请求失败,状态码: {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"请求异常: {str(e)}")
return None
def test_detect_0(image_path: str, api_url:str = "http://10.0.0.202:12342/api/v1/detect_from_base64_0"):
"""
测试yolov8消防区域侵占接口
"""
image_base64 = image_to_base64(image_path)
# 准备请求数据
payload = {
"image_base64": image_base64,
"image_type": "jpg" # 根据实际图片类型修改
}
# 发送POST请求
headers = {
"Content-Type": "application/json"
}
try:
response = requests.post(api_url, data=json.dumps(payload), headers=headers)
if response.status_code == 200:
result = response.json()
print("detect识别结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
else:
print(f"请求失败,状态码: {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"请求异常: {str(e)}")
return None
if __name__ == "__main__":
# 测试图片路径,请根据实际情况修改
# test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/data_image/001读表图片/2c7cc83019e7388a7041101da92c9829_frame_000000.jpg"
# #---------------------------------------测试ocr-----------------------------------------
# test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/data_image/001读表图片/632e474452d560edd7004f745319ff00_frame_000730.jpg"
# # 调用测试函数
# test_ocr_api(test_image_path)
# #---------------------------------------测试ocrender-----------------------------------------
#-----------------------------------------测试yolov8-----------------------------------------
# test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/YoloV8Obj/dataset_20250819/train/images/1e4c75b76e531606e2adc491a8f09ae8_frame_000000.jpg"
test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/YoloV8Obj/dataset_20250819/train/images/1e4c75b76e531606e2adc491a8f09ae8_frame_000720.jpg"
test_detect_0(test_image_path)
#-----------------------------------------测试yolov8 end-----------------------------------------

View File

@ -1,55 +0,0 @@
import base64
import requests
import json
def image_to_base64(image_path):
"""
将图片文件转换为base64编码
"""
with open(image_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
return encoded_string
def test_ocr_api(image_path, api_url="http://10.0.0.202:12342/api/v1/ocr_from_base64"):
"""
测试OCR API接口
"""
# 将图片转换为base64
image_base64 = image_to_base64(image_path)
with open("base64.txt", "w") as f:
f.write(repr(image_base64))
# 准备请求数据
payload = {
"image_base64": image_base64,
"image_type": "jpg" # 根据实际图片类型修改
}
# 发送POST请求
headers = {
"Content-Type": "application/json"
}
try:
response = requests.post(api_url, data=json.dumps(payload), headers=headers)
if response.status_code == 200:
result = response.json()
print("OCR识别结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
else:
print(f"请求失败,状态码: {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"请求异常: {str(e)}")
return None
if __name__ == "__main__":
# 测试图片路径,请根据实际情况修改
# test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/data_image/001读表图片/2c7cc83019e7388a7041101da92c9829_frame_000000.jpg"
test_image_path = "/home/admin-root/haotian/康达瑞贝斯机器狗/data_image/001读表图片/632e474452d560edd7004f745319ff00_frame_000730.jpg"
# 调用测试函数
test_ocr_api(test_image_path)