更新到HyperLPR3版本
This commit is contained in:
tunmx
@@ -0,0 +1,16 @@
|
||||
import click
|
||||
|
||||
|
||||
class AliasedGroup(click.Group):
|
||||
def get_command(self, ctx, cmd_name):
|
||||
rv = click.Group.get_command(self, ctx, cmd_name)
|
||||
if rv is not None:
|
||||
return rv
|
||||
matches = [
|
||||
x for x in self.list_commands(ctx) if x.startswith(cmd_name)
|
||||
]
|
||||
if not matches:
|
||||
return None
|
||||
elif len(matches) == 1:
|
||||
return click.Group.get_command(self, ctx, matches[0])
|
||||
ctx.fail('Too many matches: %s' % ', '.join(sorted(matches)))
|
||||
@@ -0,0 +1,20 @@
|
||||
import click
|
||||
from hyperlpr3.command.aliased_group import AliasedGroup
|
||||
from hyperlpr3.command.sample import sample
|
||||
from hyperlpr3.command.serve import rest
|
||||
|
||||
__all__ = ['cli']
|
||||
|
||||
CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
|
||||
|
||||
|
||||
@click.command(cls=AliasedGroup, context_settings=CONTEXT_SETTINGS)
|
||||
def cli():
|
||||
pass
|
||||
|
||||
|
||||
cli.add_command(sample)
|
||||
cli.add_command(rest)
|
||||
|
||||
if __name__ == '__main__':
|
||||
cli()
|
||||
@@ -0,0 +1,77 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import hyperlpr3 as lpr3
|
||||
import cv2
|
||||
import urllib
|
||||
import numpy as np
|
||||
import re
|
||||
import click
|
||||
from loguru import logger
|
||||
|
||||
type_list = ["蓝牌", "黄牌单层", "白牌单层", "绿牌新能源", "黑牌港澳", "香港单层", "香港双层", "澳门单层", "澳门双层", "黄牌双层"]
|
||||
|
||||
|
||||
def is_http_url(s):
|
||||
regex = re.compile(
|
||||
r'^(?:http|ftp)s?://' # http:// or https://
|
||||
r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|'
|
||||
r'localhost|' # localhost...
|
||||
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
|
||||
r'(?::\d+)?' # optional port
|
||||
r'(?:/?|[/?]\S+)$', re.IGNORECASE)
|
||||
|
||||
if regex.match(s):
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def url_to_image(url):
|
||||
try:
|
||||
resp = urllib.request.urlopen(url)
|
||||
image = np.asarray(bytearray(resp.read()), dtype="uint8")
|
||||
image = cv2.imdecode(image, cv2.IMREAD_COLOR)
|
||||
except Exception as err:
|
||||
return None
|
||||
|
||||
return image
|
||||
|
||||
|
||||
def get_image(path: str):
|
||||
image = None
|
||||
if is_http_url(path):
|
||||
# url
|
||||
image = url_to_image(path)
|
||||
else:
|
||||
# local path
|
||||
if path.split('.')[-1].lower() in ('jpg', 'png', 'jpeg', 'bmp',):
|
||||
image = cv2.imread(path)
|
||||
try:
|
||||
h, w, c = image.shape
|
||||
except Exception as err:
|
||||
logger.error("Failed to read image from path or url.")
|
||||
return False, None
|
||||
|
||||
return True, image
|
||||
|
||||
|
||||
@click.command(help="Exec HyperLPR3 Test Sample.")
|
||||
@click.option("-src", "--src", type=str, )
|
||||
@click.option("-det", "--det", default='low', type=click.Choice(['low', 'high']), )
|
||||
def sample(src, det):
|
||||
ret, image = get_image(src)
|
||||
if ret:
|
||||
if det == 'low':
|
||||
level = lpr3.DETECT_LEVEL_LOW
|
||||
else:
|
||||
level = lpr3.DETECT_LEVEL_HIGH
|
||||
catcher = lpr3.LicensePlateCatcher(detect_level=level)
|
||||
print("--" * 20)
|
||||
result = catcher(image)
|
||||
logger.info(f"共检测到车牌: {len(result)}")
|
||||
for res in result:
|
||||
code, conf, plate_type, box = res
|
||||
logger.success(f'[{type_list[plate_type]}]{code} {conf} {box}')
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sample()
|
||||
@@ -0,0 +1,129 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from fastapi import FastAPI, APIRouter, UploadFile, File
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from typing import List
|
||||
from fastapi.responses import JSONResponse
|
||||
import numpy as np
|
||||
import cv2
|
||||
import hyperlpr3 as lpr3
|
||||
import uvicorn
|
||||
import click
|
||||
|
||||
type_list = ["蓝牌", "黄牌单层", "白牌单层", "绿牌新能源", "黑牌港澳", "香港单层", "香港双层", "澳门单层", "澳门双层", "黄牌双层"]
|
||||
|
||||
catcher = lpr3.LicensePlateCatcher(detect_level=lpr3.DETECT_LEVEL_HIGH)
|
||||
|
||||
class BaseResponse():
|
||||
def __init__(self, *args, **kwags) -> None:
|
||||
self.response = {
|
||||
'result': None,
|
||||
}
|
||||
# self.response.update(response)
|
||||
|
||||
def http_ok_response(self, response_data):
|
||||
self.response['result'] = response_data
|
||||
self.response['code'] = 5000
|
||||
self.response['msg'] = '请求成功'
|
||||
return JSONResponse(self.response)
|
||||
|
||||
def http_prermission_denied_response(self, response_data=None, error_msg=None):
|
||||
"""
|
||||
权限错误
|
||||
"""
|
||||
self.response['result'] = response_data
|
||||
self.response['code'] = 5005
|
||||
if error_msg:
|
||||
self.response['msg'] = error_msg
|
||||
else:
|
||||
self.response['msg'] = '权限校验失败，请重新检查权限'
|
||||
|
||||
return JSONResponse(self.response)
|
||||
|
||||
def http_request_parameter_error(self, response_data=None, error_msg=None):
|
||||
"""
|
||||
请求参数错误
|
||||
"""
|
||||
self.response['result'] = response_data
|
||||
self.response['code'] = 5007
|
||||
if error_msg:
|
||||
self.response['msg'] = error_msg
|
||||
else:
|
||||
self.response['msg'] = '提交参数异常，请重新检查接口参数'
|
||||
|
||||
return JSONResponse(self.response)
|
||||
|
||||
def http_server_error(self, response_data=None, error_msg=None):
|
||||
self.response['result'] = response_data
|
||||
self.response['code'] = 5009
|
||||
if error_msg:
|
||||
self.response['msg'] = error_msg
|
||||
else:
|
||||
self.response['msg'] = '服务异常'
|
||||
|
||||
return JSONResponse(self.response)
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
title="HyperLPR3-Api",
|
||||
version='0.0.9',
|
||||
docs_url='/api/v1/docs',
|
||||
description='HyperLPR3 Api Serving'
|
||||
)
|
||||
|
||||
origins = [
|
||||
'http://localhost.slsmart.com',
|
||||
'https://localhost.slsmart.com',
|
||||
'http://localhost',
|
||||
'http://localhost:8715'
|
||||
]
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=origins,
|
||||
allow_credentials=True,
|
||||
allow_methods=['*'],
|
||||
allow_headers=['*'],
|
||||
)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def running():
|
||||
'''当前api服务程序有在正常运行'''
|
||||
return """HyperLpr3 WebApi Server Running..."""
|
||||
|
||||
|
||||
@app.post("/api/v1/rec", tags=['车牌识别'])
|
||||
async def vehicle_license_plate_recognition(file: List[UploadFile] = File(...)):
|
||||
"""上传图片进行车牌识别，上传必须为图片类型png/jpg/jpge/wabp"""
|
||||
if len(file[0].filename) == 0:
|
||||
return BaseResponse().http_request_parameter_error(error_msg='单次上传图片不能为空')
|
||||
if len(file) > 1:
|
||||
return BaseResponse().http_request_parameter_error(error_msg='该接口仅支持单张图片上传')
|
||||
if len(file) == 1:
|
||||
if file[0].filename.rsplit('.', 1)[1].lower() not in ['png', 'jpeg', 'jpg', 'wabp']:
|
||||
return BaseResponse().http_request_parameter_error(error_msg='上传必须为图片类型png/jpg/jpge')
|
||||
content = await file[0].read()
|
||||
nparr = np.fromstring(content, np.uint8)
|
||||
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR).astype(np.uint8)
|
||||
plates = catcher(img)
|
||||
results = list()
|
||||
for code, conf, plate_type, box in plates:
|
||||
plate = dict(code=code, conf=float(conf), plate_type=type_list[plate_type], box=box)
|
||||
results.append(plate)
|
||||
return BaseResponse().http_ok_response({'plate_list': results})
|
||||
|
||||
|
||||
def get_application():
|
||||
return app
|
||||
|
||||
|
||||
@click.command(help="Exec HyperLPR3 WebApi Server.")
|
||||
@click.option("-host", "--host", default="0.0.0.0", type=str, )
|
||||
@click.option("-port", "--port", default=8715, type=int, )
|
||||
@click.option("-workers", "--workers", default=1, type=int, )
|
||||
def rest(host, port, workers):
|
||||
uvicorn.run(app="hyperlpr3.command.serve:app", host=host, port=port, workers=workers)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
rest()
|
||||
Reference in New Issue
Block a user