FORMAT_CONVERSION_MAXCOMPUTE/botr/yolov8/yolo_interface.py

import base64
import json
import os
import sys
import traceback
import torch
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../../")
from botr.yolov8.model import Predictor
from botr.yolov8.predict import detect
from config.max_compute_config import MAX_COMPUTE
import time
import cv2
from flask import Flask, request
from format_convert.utils import request_post, log, get_md5_from_bytes, get_platform, bytes2np
from format_convert import _global


ROOT = os.path.abspath(os.path.dirname(__file__)) + '/../../'
model_path = ROOT + 'botr/yolov8/weights.pt'

# 接口配置
app = Flask(__name__)


@app.route('/yolo', methods=['POST'])
def _yolo():
    _global._init()
    _global.update({"port": globals().get("port")})
    start_time = time.time()

    log("into yolo_interface _yolo")
    try:
        if not request.form:
            log("yolo no data!")
            return json.dumps({"b_table_list": str([-9])})

        yolo_predictor = globals().get("global_yolo_predictor")
        if yolo_predictor is None:
            image_size = 640
            device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
            # device = 'cpu'
            yolo_predictor = Predictor(image_size, device, model_path)
            globals().update({"global_yolo_predictor": yolo_predictor})

        data = request.form.get("data")
        _md5 = request.form.get("md5")
        _global.update({"md5": _md5})

        b_table_list = yolo(data, yolo_predictor).get('b_table_list')
        return json.dumps({"b_table_list": b_table_list})
    except TimeoutError:
        return json.dumps({"b_table_list": str([-5])})
    except:
        traceback.print_exc()
        return json.dumps({"b_table_list": str([-1])})
    finally:
        log("yolo interface finish time " + str(time.time()-start_time))


def yolo(data, predictor):
    log("into yolo_interface yolo")
    try:
        img_data = base64.b64decode(data)
        img = bytes2np(img_data)
        b_table_list = detect(img, predictor)
        return {"b_table_list": b_table_list}
    except TimeoutError:
        raise TimeoutError


def test_yolo_model(from_remote=True):
    _global._init()
    file_path = "C:/Users/Administrator/Desktop/test_b_table/yolo_error/error6.png"
    # file_path = "C:/Users/Administrator/Desktop/test_b_table/error10.png"
    # file_path = "C:/Users/Administrator/Downloads/1652672734044.jpg"

    from format_convert.convert_image import get_best_predict_size
    from format_convert.utils import np2bytes, pil_resize

    image_np = cv2.imread(file_path)
    # best_h, best_w = get_best_predict_size(image_np)
    # image_np = pil_resize(image_np, best_h, best_w)

    file_bytes = np2bytes(image_np)
    file_base64 = base64.b64encode(file_bytes)
    _md5 = get_md5_from_bytes(file_bytes)[0]

    _global.update({"port": 15010, "md5": _md5})

    file_json = {"data": file_base64, "md5": _md5}
    # _url = "http://192.168.2.104:18080/yolo"
    _url = "http://127.0.0.1:18080/yolo"

    r = json.loads(request_post(_url, file_json))
    print(r)
    r = r.get('b_table_list')[0]
    for bbox in r:
        bbox = [int(x) for x in bbox]
        cv2.rectangle(image_np, bbox[0:2], bbox[2:4], (0, 0, 255), 2)
    cv2.namedWindow('img', cv2.WINDOW_NORMAL)
    cv2.imshow('img', image_np)
    cv2.waitKey(0)


if __name__ == '__main__':
    # port = 18080
    # using_gpu_index = 0
    # app.run(host='0.0.0.0', port=port)

    test_yolo_model(True)