| | |
| | | import logging |
| | | import socketserver |
| | | from http.server import BaseHTTPRequestHandler |
| | | import cv2 |
| | | import numpy |
| | | |
| | | import ths_industry_util |
| | | import constant |
| | | from log_module.log import logger_system |
| | | from utils import ths_industry_util |
| | | from ocr import ocr_util |
| | | from ocr.ocr_util import OcrUtil |
| | | from third_data import kpl_util |
| | | from trade import bidding_money_manager |
| | | |
| | |
| | | datas = self.ocr_temp_data[matId] |
| | | if rows * cols == len(datas): |
| | | self.ocr_temp_data.pop(matId) |
| | | mat = cv2.numpy.zeros((rows, cols, 1), cv2.numpy.uint8) |
| | | mat = numpy.zeros((rows, cols, 1), numpy.uint8) |
| | | for r in range(0, rows): |
| | | for c in range(0, cols): |
| | | mat[r][c] = [datas[r * cols + c]] |
| | | # cv2.imwrite("D:/test.png", mat) |
| | | # cv2.imwrite(f"{constant.get_path_prefix()}/test.png", mat) |
| | | ocr_results = ocr_util.OcrUtil.ocr_with_key(mat, key) |
| | | if not ocr_results: |
| | | # 多重识别,防止识别出错 |
| | |
| | | datas = self.ocr_temp_data[matId] |
| | | if rows * cols == len(datas): |
| | | self.ocr_temp_data.pop(matId) |
| | | mat = cv2.numpy.zeros((rows, cols, 1), cv2.numpy.uint8) |
| | | mat = numpy.zeros((rows, cols, 1), numpy.uint8) |
| | | for r in range(0, rows): |
| | | for c in range(0, cols): |
| | | mat[r][c] = [datas[r * cols + c]] |
| | | # cv2.imwrite("D:/test.png", mat) |
| | | # cv2.imwrite(f"{constant.get_path_prefix()}/test.png", mat) |
| | | ocr_results = ocr_util.OcrUtil.ocr_with_key(mat, ".") |
| | | code_name = "" |
| | | for res in ocr_results: |
| | | code_name += res[0] |
| | | # TODO 根据代码名称获取代码 |
| | | # 根据代码名称获取代码 |
| | | code = ths_industry_util.get_code_by_name(code_name) |
| | | # 图像识别 |
| | | return_str = json.dumps({"code": 0, "data": {"code": code}}) |
| | |
| | | elif _type == 201: |
| | | imgdata = base64.b64decode(data["img"]) |
| | | results = ocr_util.OcrUtil.easy_ocr(imgdata) |
| | | print(results) |
| | | # print(results) |
| | | kpl_datas = kpl_util.parse_kpl_datas(results) |
| | | if kpl_datas: |
| | | bidding_money_manager.set_bidding_money(kpl_datas) |
| | | with open("D:/kpl.png", mode="wb") as f: |
| | | with open(f"{constant.get_path_prefix()}/kpl.png", mode="wb") as f: |
| | | f.write(imgdata) |
| | | except Exception as e: |
| | | logging.exception(e) |
| | |
| | | |
| | | def run(addr, port): |
| | | handler = OCRServer |
| | | httpd = socketserver.TCPServer((addr, port), handler) |
| | | print("HTTP server is at: http://%s:%d/" % (addr, port)) |
| | | httpd.serve_forever() |
| | | try: |
| | | httpd = socketserver.TCPServer((addr, port), handler) |
| | | # print("HTTP server is at: http://%s:%d/" % (addr, port)) |
| | | httpd.serve_forever() |
| | | except Exception as e: |
| | | logger_system.exception(e) |
| | | logger_system.error(f"端口服务器:{port} 启动失败") |
| | | |
| | | |
| | | if __name__ == "__main__": |
| | | str_={"id":"123"} |
| | | print(type(str_)==str) |
| | | # print(type(str_)==str) |