You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1031 lines
37 KiB

# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import os
import sys
import cv2
import numpy as np
import paddle
import PIL
from PIL import Image, ImageDraw, ImageFont
import math
from paddle import inference
import random
import yaml
from ppocr.utils.logging import get_logger
def str2bool(v):
return v.lower() in ("true", "yes", "t", "y", "1")
def str2int_tuple(v):
return tuple([int(i.strip()) for i in v.split(",")])
def init_args():
parser = argparse.ArgumentParser()
# params for prediction engine
parser.add_argument("--use_gpu", type=str2bool, default=True)
parser.add_argument("--use_xpu", type=str2bool, default=False)
parser.add_argument("--use_npu", type=str2bool, default=False)
parser.add_argument("--use_mlu", type=str2bool, default=False)
parser.add_argument(
"--use_gcu",
type=str2bool,
default=False,
help="Use Enflame GCU(General Compute Unit)",
)
parser.add_argument("--ir_optim", type=str2bool, default=True)
parser.add_argument("--use_tensorrt", type=str2bool, default=False)
parser.add_argument("--min_subgraph_size", type=int, default=15)
parser.add_argument("--precision", type=str, default="fp32")
parser.add_argument("--gpu_mem", type=int, default=500)
parser.add_argument("--gpu_id", type=int, default=0)
# params for text detector
parser.add_argument("--image_dir", type=str)
parser.add_argument("--page_num", type=int, default=0)
parser.add_argument("--det_algorithm", type=str, default="DB")
parser.add_argument("--det_model_dir", type=str)
parser.add_argument("--det_limit_side_len", type=float, default=960)
parser.add_argument("--det_limit_type", type=str, default="max")
parser.add_argument("--det_box_type", type=str, default="quad")
# DB params
parser.add_argument("--det_db_thresh", type=float, default=0.3)
parser.add_argument("--det_db_box_thresh", type=float, default=0.6)
parser.add_argument("--det_db_unclip_ratio", type=float, default=1.5)
parser.add_argument("--max_batch_size", type=int, default=10)
parser.add_argument("--use_dilation", type=str2bool, default=False)
parser.add_argument("--det_db_score_mode", type=str, default="fast")
# EAST params
parser.add_argument("--det_east_score_thresh", type=float, default=0.8)
parser.add_argument("--det_east_cover_thresh", type=float, default=0.1)
parser.add_argument("--det_east_nms_thresh", type=float, default=0.2)
# SAST params
parser.add_argument("--det_sast_score_thresh", type=float, default=0.5)
parser.add_argument("--det_sast_nms_thresh", type=float, default=0.2)
# PSE params
parser.add_argument("--det_pse_thresh", type=float, default=0)
parser.add_argument("--det_pse_box_thresh", type=float, default=0.85)
parser.add_argument("--det_pse_min_area", type=float, default=16)
parser.add_argument("--det_pse_scale", type=int, default=1)
# FCE params
parser.add_argument("--scales", type=list, default=[8, 16, 32])
parser.add_argument("--alpha", type=float, default=1.0)
parser.add_argument("--beta", type=float, default=1.0)
parser.add_argument("--fourier_degree", type=int, default=5)
# params for text recognizer
parser.add_argument("--rec_algorithm", type=str, default="SVTR_LCNet")
parser.add_argument("--rec_model_dir", type=str)
parser.add_argument("--rec_image_inverse", type=str2bool, default=True)
parser.add_argument("--rec_image_shape", type=str, default="3, 48, 320")
parser.add_argument("--rec_batch_num", type=int, default=6)
parser.add_argument("--max_text_length", type=int, default=25)
parser.add_argument(
"--rec_char_dict_path", type=str, default="./ppocr/utils/ppocr_keys_v1.txt"
)
parser.add_argument("--use_space_char", type=str2bool, default=True)
parser.add_argument("--vis_font_path", type=str, default="./doc/fonts/simfang.ttf")
parser.add_argument("--drop_score", type=float, default=0.5)
# params for e2e
parser.add_argument("--e2e_algorithm", type=str, default="PGNet")
parser.add_argument("--e2e_model_dir", type=str)
parser.add_argument("--e2e_limit_side_len", type=float, default=768)
parser.add_argument("--e2e_limit_type", type=str, default="max")
# PGNet params
parser.add_argument("--e2e_pgnet_score_thresh", type=float, default=0.5)
parser.add_argument(
"--e2e_char_dict_path", type=str, default="./ppocr/utils/ic15_dict.txt"
)
parser.add_argument("--e2e_pgnet_valid_set", type=str, default="totaltext")
parser.add_argument("--e2e_pgnet_mode", type=str, default="fast")
# params for text classifier
parser.add_argument("--use_angle_cls", type=str2bool, default=False)
parser.add_argument("--cls_model_dir", type=str)
parser.add_argument("--cls_image_shape", type=str, default="3, 48, 192")
parser.add_argument("--label_list", type=list, default=["0", "180"])
parser.add_argument("--cls_batch_num", type=int, default=6)
parser.add_argument("--cls_thresh", type=float, default=0.9)
parser.add_argument("--enable_mkldnn", type=str2bool, default=None)
parser.add_argument("--cpu_threads", type=int, default=10)
parser.add_argument("--use_pdserving", type=str2bool, default=False)
parser.add_argument("--warmup", type=str2bool, default=False)
# SR params
parser.add_argument("--sr_model_dir", type=str)
parser.add_argument("--sr_image_shape", type=str, default="3, 32, 128")
parser.add_argument("--sr_batch_num", type=int, default=1)
#
parser.add_argument("--draw_img_save_dir", type=str, default="./inference_results")
parser.add_argument("--save_crop_res", type=str2bool, default=False)
parser.add_argument("--crop_res_save_dir", type=str, default="./output")
# multi-process
parser.add_argument("--use_mp", type=str2bool, default=False)
parser.add_argument("--total_process_num", type=int, default=1)
parser.add_argument("--process_id", type=int, default=0)
parser.add_argument("--benchmark", type=str2bool, default=False)
parser.add_argument("--save_log_path", type=str, default="./log_output/")
parser.add_argument("--show_log", type=str2bool, default=True)
parser.add_argument("--use_onnx", type=str2bool, default=False)
parser.add_argument("--onnx_providers", nargs="+", type=str, default=False)
parser.add_argument("--onnx_sess_options", type=list, default=False)
# extended function
parser.add_argument(
"--return_word_box",
type=str2bool,
default=False,
help="Whether return the bbox of each word (split by space) or chinese character. Only used in ppstructure for layout recovery",
)
return parser
def parse_args():
parser = init_args()
return parser.parse_args()
def create_predictor(args, mode, logger):
if mode == "det":
model_dir = args.det_model_dir
elif mode == "cls":
model_dir = args.cls_model_dir
elif mode == "rec":
model_dir = args.rec_model_dir
elif mode == "table":
model_dir = args.table_model_dir
elif mode == "ser":
model_dir = args.ser_model_dir
elif mode == "re":
model_dir = args.re_model_dir
elif mode == "sr":
model_dir = args.sr_model_dir
elif mode == "layout":
model_dir = args.layout_model_dir
else:
model_dir = args.e2e_model_dir
if model_dir is None:
logger.info("not find {} model file path {}".format(mode, model_dir))
sys.exit(0)
if args.use_onnx:
import onnxruntime as ort
model_file_path = model_dir
if not os.path.exists(model_file_path):
raise ValueError("not find model file path {}".format(model_file_path))
sess_options = args.onnx_sess_options or None
if args.onnx_providers and len(args.onnx_providers) > 0:
sess = ort.InferenceSession(
model_file_path,
providers=args.onnx_providers,
sess_options=sess_options,
)
elif args.use_gpu:
sess = ort.InferenceSession(
model_file_path,
providers=[
(
"CUDAExecutionProvider",
{"device_id": args.gpu_id, "cudnn_conv_algo_search": "DEFAULT"},
)
],
sess_options=sess_options,
)
else:
sess = ort.InferenceSession(
model_file_path,
providers=["CPUExecutionProvider"],
sess_options=sess_options,
)
inputs = sess.get_inputs()
return (
sess,
inputs[0] if len(inputs) == 1 else [vo.name for vo in inputs],
None,
None,
)
else:
file_names = ["model", "inference"]
for file_name in file_names:
params_file_path = f"{model_dir}/{file_name}.pdiparams"
if os.path.exists(params_file_path):
break
if not os.path.exists(params_file_path):
raise ValueError(f"not find {file_name}.pdiparams in {model_dir}")
if not (
os.path.exists(f"{model_dir}/{file_name}.pdmodel")
or os.path.exists(f"{model_dir}/{file_name}.json")
):
raise ValueError(
f"neither {file_name}.json nor {file_name}.pdmodel was found in {model_dir}."
)
if os.path.exists(f"{model_dir}/{file_name}.json"):
model_file_path = f"{model_dir}/{file_name}.json"
else:
model_file_path = f"{model_dir}/{file_name}.pdmodel"
config = inference.Config(model_file_path, params_file_path)
if hasattr(args, "precision"):
if args.precision == "fp16" and args.use_tensorrt:
precision = inference.PrecisionType.Half
elif args.precision == "int8":
precision = inference.PrecisionType.Int8
else:
precision = inference.PrecisionType.Float32
else:
precision = inference.PrecisionType.Float32
if args.use_gpu:
gpu_id = get_infer_gpuid()
if gpu_id is None:
logger.warning(
"GPU is not found in current device by nvidia-smi. Please check your device or ignore it if run on jetson."
)
config.enable_use_gpu(args.gpu_mem, args.gpu_id)
if args.use_tensorrt:
if ".json" in model_file_path:
trt_dynamic_shapes = {}
trt_dynamic_shape_input_data = {}
if os.path.exists(f"{model_dir}/inference.yml"):
model_config = load_config(f"{model_dir}/inference.yml")
trt_dynamic_shapes = (
model_config.get("Hpi", {})
.get("backend_configs", {})
.get("paddle_infer", {})
.get("trt_dynamic_shapes", {})
)
trt_dynamic_shape_input_data = (
model_config.get("Hpi", {})
.get("backend_configs", {})
.get("paddle_infer", {})
.get("trt_dynamic_shapes_input_data", {})
)
if not trt_dynamic_shapes:
raise RuntimeError(
"Configuration Error: 'trt_dynamic_shapes' must be defined in 'inference.yml' for Paddle Inference TensorRT."
)
trt_save_path = f"{model_dir}/.cache/trt/{file_name}"
trt_model_file_path = trt_save_path + ".json"
trt_params_file_path = trt_save_path + ".pdiparams"
if not os.path.exists(trt_model_file_path) or not os.path.exists(
trt_params_file_path
):
_convert_trt(
{},
model_file_path,
params_file_path,
trt_save_path,
args.gpu_id,
trt_dynamic_shapes,
trt_dynamic_shape_input_data,
)
config = inference.Config(model_file_path, params_file_path)
config.exp_disable_mixed_precision_ops({"feed", "fetch"})
config.enable_use_gpu(args.gpu_mem, args.gpu_id)
else:
config.enable_tensorrt_engine(
workspace_size=1 << 30,
precision_mode=precision,
max_batch_size=args.max_batch_size,
min_subgraph_size=args.min_subgraph_size, # skip the minimum trt subgraph
use_calib_mode=False,
)
# collect shape
trt_shape_f = os.path.join(
model_dir, f"{mode}_trt_dynamic_shape.txt"
)
if not os.path.exists(trt_shape_f):
config.collect_shape_range_info(trt_shape_f)
logger.info(f"collect dynamic shape info into : {trt_shape_f}")
try:
config.enable_tuned_tensorrt_dynamic_shape(trt_shape_f, True)
except Exception as E:
logger.info(E)
logger.info("Please keep your paddlepaddle-gpu >= 2.3.0!")
elif args.use_npu:
config.enable_custom_device("npu")
elif args.use_mlu:
config.enable_custom_device("mlu")
elif args.use_xpu:
config.enable_xpu(10 * 1024 * 1024)
elif args.use_gcu: # for Enflame GCU(General Compute Unit)
assert paddle.device.is_compiled_with_custom_device("gcu"), (
"Args use_gcu cannot be set as True while your paddle "
"is not compiled with gcu! \nPlease try: \n"
"\t1. Install paddle-custom-gcu to run model on GCU. \n"
"\t2. Set use_gcu as False in args to run model on CPU."
)
import paddle_custom_device.gcu.passes as gcu_passes
gcu_passes.setUp()
if args.precision == "fp16":
config.enable_custom_device(
"gcu", 0, paddle.inference.PrecisionType.Half
)
gcu_passes.set_exp_enable_mixed_precision_ops(config)
else:
config.enable_custom_device("gcu")
if paddle.framework.use_pir_api():
config.enable_new_ir(True)
config.enable_new_executor(True)
else:
pass_builder = config.pass_builder()
gcu_passes.append_passes_for_legacy_ir(pass_builder, "PaddleOCR")
else:
config.disable_gpu()
if args.enable_mkldnn is not None:
if args.enable_mkldnn:
# cache 10 different shapes for mkldnn to avoid memory leak
config.set_mkldnn_cache_capacity(10)
config.enable_mkldnn()
if args.precision == "fp16":
config.enable_mkldnn_bfloat16()
else:
if hasattr(config, "disable_mkldnn"):
config.disable_mkldnn()
if hasattr(args, "cpu_threads"):
config.set_cpu_math_library_num_threads(args.cpu_threads)
else:
# default cpu threads as 10
config.set_cpu_math_library_num_threads(10)
if hasattr(config, "enable_new_ir"):
config.enable_new_ir()
if hasattr(config, "enable_new_executor"):
config.enable_new_executor()
# enable memory optim
config.enable_memory_optim()
config.disable_glog_info()
if not args.use_gcu: # for Enflame GCU(General Compute Unit)
config.delete_pass("conv_transpose_eltwiseadd_bn_fuse_pass")
config.delete_pass("matmul_transpose_reshape_fuse_pass")
if mode == "rec" and args.rec_algorithm == "SRN":
config.delete_pass("gpu_cpu_map_matmul_v2_to_matmul_pass")
if mode == "re":
config.delete_pass("simplify_with_basic_ops_pass")
if mode == "table":
config.delete_pass("fc_fuse_pass") # not supported for table
config.switch_use_feed_fetch_ops(False)
config.switch_ir_optim(True)
# create predictor
predictor = inference.create_predictor(config)
input_names = predictor.get_input_names()
if mode in ["ser", "re"]:
input_tensor = []
for name in input_names:
input_tensor.append(predictor.get_input_handle(name))
else:
for name in input_names:
input_tensor = predictor.get_input_handle(name)
output_tensors = get_output_tensors(args, mode, predictor)
return predictor, input_tensor, output_tensors, config
def _convert_trt(
trt_cfg_setting,
pp_model_file,
pp_params_file,
trt_save_path,
device_id,
dynamic_shapes,
dynamic_shape_input_data,
):
from paddle.tensorrt.export import Input, TensorRTConfig, convert
def _set_trt_config():
for attr_name in trt_cfg_setting:
assert hasattr(
trt_config, attr_name
), f"The `{type(trt_config)}` don't have the attribute `{attr_name}`!"
setattr(trt_config, attr_name, trt_cfg_setting[attr_name])
def _get_predictor(model_file, params_file):
# HACK
config = inference.Config(str(model_file), str(params_file))
config.enable_use_gpu(100, device_id)
# NOTE: Disable oneDNN to circumvent a bug in Paddle Inference
config.disable_mkldnn()
config.disable_glog_info()
return inference.create_predictor(config)
dynamic_shape_input_data = dynamic_shape_input_data or {}
predictor = _get_predictor(pp_model_file, pp_params_file)
input_names = predictor.get_input_names()
for name in dynamic_shapes:
if name not in input_names:
raise ValueError(
f"Invalid input name {repr(name)} found in `dynamic_shapes`"
)
for name in input_names:
if name not in dynamic_shapes:
raise ValueError(f"Input name {repr(name)} not found in `dynamic_shapes`")
for name in dynamic_shape_input_data:
if name not in input_names:
raise ValueError(
f"Invalid input name {repr(name)} found in `dynamic_shape_input_data`"
)
trt_inputs = []
for name, candidate_shapes in dynamic_shapes.items():
# XXX: Currently we have no way to get the data type of the tensor
# without creating an input handle.
handle = predictor.get_input_handle(name)
dtype = _pd_dtype_to_np_dtype(handle.type())
min_shape, opt_shape, max_shape = candidate_shapes
if name in dynamic_shape_input_data:
min_arr = np.array(dynamic_shape_input_data[name][0], dtype=dtype).reshape(
min_shape
)
opt_arr = np.array(dynamic_shape_input_data[name][1], dtype=dtype).reshape(
opt_shape
)
max_arr = np.array(dynamic_shape_input_data[name][2], dtype=dtype).reshape(
max_shape
)
else:
min_arr = np.ones(min_shape, dtype=dtype)
opt_arr = np.ones(opt_shape, dtype=dtype)
max_arr = np.ones(max_shape, dtype=dtype)
# refer to: https://github.com/PolaKuma/Paddle/blob/3347f225bc09f2ec09802a2090432dd5cb5b6739/test/tensorrt/test_converter_model_resnet50.py
trt_input = Input((min_arr, opt_arr, max_arr))
trt_inputs.append(trt_input)
# Create TensorRTConfig
trt_config = TensorRTConfig(inputs=trt_inputs)
_set_trt_config()
trt_config.save_model_dir = trt_save_path
pp_model_path = pp_model_file.split(".")[0]
convert(pp_model_path, trt_config)
def _pd_dtype_to_np_dtype(pd_dtype):
if pd_dtype == inference.DataType.FLOAT64:
return np.float64
elif pd_dtype == inference.DataType.FLOAT32:
return np.float32
elif pd_dtype == inference.DataType.INT64:
return np.int64
elif pd_dtype == inference.DataType.INT32:
return np.int32
elif pd_dtype == inference.DataType.UINT8:
return np.uint8
elif pd_dtype == inference.DataType.INT8:
return np.int8
else:
raise TypeError(f"Unsupported data type: {pd_dtype}")
def load_config(file_path):
_, ext = os.path.splitext(file_path)
if ext not in [".yml", ".yaml"]:
raise ValueError(f"only support yaml files for now, got {file_path}")
with open(file_path, "rb") as file:
config = yaml.load(file, Loader=yaml.SafeLoader)
return config
def get_output_tensors(args, mode, predictor):
output_names = predictor.get_output_names()
output_tensors = []
if mode == "rec" and args.rec_algorithm in ["CRNN", "SVTR_LCNet", "SVTR_HGNet"]:
output_name = "softmax_0.tmp_0"
if output_name in output_names:
return [predictor.get_output_handle(output_name)]
else:
for output_name in output_names:
output_tensor = predictor.get_output_handle(output_name)
output_tensors.append(output_tensor)
else:
for output_name in output_names:
output_tensor = predictor.get_output_handle(output_name)
output_tensors.append(output_tensor)
return output_tensors
def get_infer_gpuid():
"""
Get the GPU ID to be used for inference.
Returns:
int: The GPU ID to be used for inference.
"""
logger = get_logger()
if not paddle.device.is_compiled_with_rocm:
gpu_id_str = os.environ.get("CUDA_VISIBLE_DEVICES", "0")
else:
gpu_id_str = os.environ.get("HIP_VISIBLE_DEVICES", "0")
gpu_ids = gpu_id_str.split(",")
logger.warning(
"The first GPU is used for inference by default, GPU ID: {}".format(gpu_ids[0])
)
return int(gpu_ids[0])
def draw_e2e_res(dt_boxes, strs, img_path):
src_im = cv2.imread(img_path)
for box, str in zip(dt_boxes, strs):
box = box.astype(np.int32).reshape((-1, 1, 2))
cv2.polylines(src_im, [box], True, color=(255, 255, 0), thickness=2)
cv2.putText(
src_im,
str,
org=(int(box[0, 0, 0]), int(box[0, 0, 1])),
fontFace=cv2.FONT_HERSHEY_COMPLEX,
fontScale=0.7,
color=(0, 255, 0),
thickness=1,
)
return src_im
def draw_text_det_res(dt_boxes, img):
for box in dt_boxes:
box = np.array(box).astype(np.int32).reshape(-1, 2)
cv2.polylines(img, [box], True, color=(255, 255, 0), thickness=2)
return img
def resize_img(img, input_size=600):
"""
resize img and limit the longest side of the image to input_size
"""
img = np.array(img)
im_shape = img.shape
im_size_max = np.max(im_shape[0:2])
im_scale = float(input_size) / float(im_size_max)
img = cv2.resize(img, None, None, fx=im_scale, fy=im_scale)
return img
def draw_ocr(
image,
boxes,
txts=None,
scores=None,
drop_score=0.5,
font_path="./doc/fonts/simfang.ttf",
):
"""
Visualize the results of OCR detection and recognition
args:
image(Image|array): RGB image
boxes(list): boxes with shape(N, 4, 2)
txts(list): the texts
scores(list): txxs corresponding scores
drop_score(float): only scores greater than drop_threshold will be visualized
font_path: the path of font which is used to draw text
return(array):
the visualized img
"""
if scores is None:
scores = [1] * len(boxes)
box_num = len(boxes)
for i in range(box_num):
if scores is not None and (scores[i] < drop_score or math.isnan(scores[i])):
continue
box = np.reshape(np.array(boxes[i]), [-1, 1, 2]).astype(np.int64)
image = cv2.polylines(np.array(image), [box], True, (255, 0, 0), 2)
if txts is not None:
img = np.array(resize_img(image, input_size=600))
txt_img = text_visual(
txts,
scores,
img_h=img.shape[0],
img_w=600,
threshold=drop_score,
font_path=font_path,
)
img = np.concatenate([np.array(img), np.array(txt_img)], axis=1)
return img
return image
def draw_ocr_box_txt(
image,
boxes,
txts=None,
scores=None,
drop_score=0.5,
font_path="./doc/fonts/simfang.ttf",
):
h, w = image.height, image.width
img_left = image.copy()
img_right = np.ones((h, w, 3), dtype=np.uint8) * 255
random.seed(0)
draw_left = ImageDraw.Draw(img_left)
if txts is None or len(txts) != len(boxes):
txts = [None] * len(boxes)
for idx, (box, txt) in enumerate(zip(boxes, txts)):
if scores is not None and scores[idx] < drop_score:
continue
color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
draw_left.polygon(box, fill=color)
img_right_text = draw_box_txt_fine((w, h), box, txt, font_path)
pts = np.array(box, np.int32).reshape((-1, 1, 2))
cv2.polylines(img_right_text, [pts], True, color, 1)
img_right = cv2.bitwise_and(img_right, img_right_text)
img_left = Image.blend(image, img_left, 0.5)
img_show = Image.new("RGB", (w * 2, h), (255, 255, 255))
img_show.paste(img_left, (0, 0, w, h))
img_show.paste(Image.fromarray(img_right), (w, 0, w * 2, h))
return np.array(img_show)
def draw_box_txt_fine(img_size, box, txt, font_path="./doc/fonts/simfang.ttf"):
box_height = int(
math.sqrt((box[0][0] - box[3][0]) ** 2 + (box[0][1] - box[3][1]) ** 2)
)
box_width = int(
math.sqrt((box[0][0] - box[1][0]) ** 2 + (box[0][1] - box[1][1]) ** 2)
)
if box_height > 2 * box_width and box_height > 30:
img_text = Image.new("RGB", (box_height, box_width), (255, 255, 255))
draw_text = ImageDraw.Draw(img_text)
if txt:
font = create_font(txt, (box_height, box_width), font_path)
draw_text.text([0, 0], txt, fill=(0, 0, 0), font=font)
img_text = img_text.transpose(Image.ROTATE_270)
else:
img_text = Image.new("RGB", (box_width, box_height), (255, 255, 255))
draw_text = ImageDraw.Draw(img_text)
if txt:
font = create_font(txt, (box_width, box_height), font_path)
draw_text.text([0, 0], txt, fill=(0, 0, 0), font=font)
pts1 = np.float32(
[[0, 0], [box_width, 0], [box_width, box_height], [0, box_height]]
)
pts2 = np.array(box, dtype=np.float32)
M = cv2.getPerspectiveTransform(pts1, pts2)
img_text = np.array(img_text, dtype=np.uint8)
img_right_text = cv2.warpPerspective(
img_text,
M,
img_size,
flags=cv2.INTER_NEAREST,
borderMode=cv2.BORDER_CONSTANT,
borderValue=(255, 255, 255),
)
return img_right_text
def create_font(txt, sz, font_path="./doc/fonts/simfang.ttf"):
font_size = int(sz[1] * 0.99)
font = ImageFont.truetype(font_path, font_size, encoding="utf-8")
if int(PIL.__version__.split(".")[0]) < 10:
length = font.getsize(txt)[0]
else:
length = font.getlength(txt)
if length > sz[0]:
font_size = int(font_size * sz[0] / length)
font = ImageFont.truetype(font_path, font_size, encoding="utf-8")
return font
def str_count(s):
"""
Count the number of Chinese characters,
a single English character and a single number
equal to half the length of Chinese characters.
args:
s(string): the input of string
return(int):
the number of Chinese characters
"""
import string
count_zh = count_pu = 0
s_len = len(s)
en_dg_count = 0
for c in s:
if c in string.ascii_letters or c.isdigit() or c.isspace():
en_dg_count += 1
elif c.isalpha():
count_zh += 1
else:
count_pu += 1
return s_len - math.ceil(en_dg_count / 2)
def text_visual(
texts, scores, img_h=400, img_w=600, threshold=0.0, font_path="./doc/simfang.ttf"
):
"""
create new blank img and draw txt on it
args:
texts(list): the text will be draw
scores(list|None): corresponding score of each txt
img_h(int): the height of blank img
img_w(int): the width of blank img
font_path: the path of font which is used to draw text
return(array):
"""
if scores is not None:
assert len(texts) == len(
scores
), "The number of txts and corresponding scores must match"
def create_blank_img():
blank_img = np.ones(shape=[img_h, img_w], dtype=np.uint8) * 255
blank_img[:, img_w - 1 :] = 0
blank_img = Image.fromarray(blank_img).convert("RGB")
draw_txt = ImageDraw.Draw(blank_img)
return blank_img, draw_txt
blank_img, draw_txt = create_blank_img()
font_size = 20
txt_color = (0, 0, 0)
font = ImageFont.truetype(font_path, font_size, encoding="utf-8")
gap = font_size + 5
txt_img_list = []
count, index = 1, 0
for idx, txt in enumerate(texts):
index += 1
if scores[idx] < threshold or math.isnan(scores[idx]):
index -= 1
continue
first_line = True
while str_count(txt) >= img_w // font_size - 4:
tmp = txt
txt = tmp[: img_w // font_size - 4]
if first_line:
new_txt = str(index) + ": " + txt
first_line = False
else:
new_txt = " " + txt
draw_txt.text((0, gap * count), new_txt, txt_color, font=font)
txt = tmp[img_w // font_size - 4 :]
if count >= img_h // gap - 1:
txt_img_list.append(np.array(blank_img))
blank_img, draw_txt = create_blank_img()
count = 0
count += 1
if first_line:
new_txt = str(index) + ": " + txt + " " + "%.3f" % (scores[idx])
else:
new_txt = " " + txt + " " + "%.3f" % (scores[idx])
draw_txt.text((0, gap * count), new_txt, txt_color, font=font)
# whether add new blank img or not
if count >= img_h // gap - 1 and idx + 1 < len(texts):
txt_img_list.append(np.array(blank_img))
blank_img, draw_txt = create_blank_img()
count = 0
count += 1
txt_img_list.append(np.array(blank_img))
if len(txt_img_list) == 1:
blank_img = np.array(txt_img_list[0])
else:
blank_img = np.concatenate(txt_img_list, axis=1)
return np.array(blank_img)
def base64_to_cv2(b64str):
import base64
data = base64.b64decode(b64str.encode("utf8"))
data = np.frombuffer(data, np.uint8)
data = cv2.imdecode(data, cv2.IMREAD_COLOR)
return data
def draw_boxes(image, boxes, scores=None, drop_score=0.5):
if scores is None:
scores = [1] * len(boxes)
for box, score in zip(boxes, scores):
if score < drop_score:
continue
box = np.reshape(np.array(box), [-1, 1, 2]).astype(np.int64)
image = cv2.polylines(np.array(image), [box], True, (255, 0, 0), 2)
return image
def get_rotate_crop_image(img, points):
"""
img_height, img_width = img.shape[0:2]
left = int(np.min(points[:, 0]))
right = int(np.max(points[:, 0]))
top = int(np.min(points[:, 1]))
bottom = int(np.max(points[:, 1]))
img_crop = img[top:bottom, left:right, :].copy()
points[:, 0] = points[:, 0] - left
points[:, 1] = points[:, 1] - top
"""
assert len(points) == 4, "shape of points must be 4*2"
img_crop_width = int(
max(
np.linalg.norm(points[0] - points[1]), np.linalg.norm(points[2] - points[3])
)
)
img_crop_height = int(
max(
np.linalg.norm(points[0] - points[3]), np.linalg.norm(points[1] - points[2])
)
)
pts_std = np.float32(
[
[0, 0],
[img_crop_width, 0],
[img_crop_width, img_crop_height],
[0, img_crop_height],
]
)
M = cv2.getPerspectiveTransform(points, pts_std)
dst_img = cv2.warpPerspective(
img,
M,
(img_crop_width, img_crop_height),
borderMode=cv2.BORDER_REPLICATE,
flags=cv2.INTER_CUBIC,
)
dst_img_height, dst_img_width = dst_img.shape[0:2]
if dst_img_height * 1.0 / dst_img_width >= 1.5:
dst_img = np.rot90(dst_img)
return dst_img
def get_minarea_rect_crop(img, points):
bounding_box = cv2.minAreaRect(np.array(points).astype(np.int32))
points = sorted(list(cv2.boxPoints(bounding_box)), key=lambda x: x[0])
index_a, index_b, index_c, index_d = 0, 1, 2, 3
if points[1][1] > points[0][1]:
index_a = 0
index_d = 1
else:
index_a = 1
index_d = 0
if points[3][1] > points[2][1]:
index_b = 2
index_c = 3
else:
index_b = 3
index_c = 2
box = [points[index_a], points[index_b], points[index_c], points[index_d]]
crop_img = get_rotate_crop_image(img, np.array(box))
return crop_img
def slice_generator(image, horizontal_stride, vertical_stride, maximum_slices=500):
if not isinstance(image, np.ndarray):
image = np.array(image)
image_h, image_w = image.shape[:2]
vertical_num_slices = (image_h + vertical_stride - 1) // vertical_stride
horizontal_num_slices = (image_w + horizontal_stride - 1) // horizontal_stride
assert (
vertical_num_slices > 0
), f"Invalid number ({vertical_num_slices}) of vertical slices"
assert (
horizontal_num_slices > 0
), f"Invalid number ({horizontal_num_slices}) of horizontal slices"
if vertical_num_slices >= maximum_slices:
recommended_vertical_stride = max(1, image_h // maximum_slices) + 1
assert (
False
), f"Too computationally expensive with {vertical_num_slices} slices, try a higher vertical stride (recommended minimum: {recommended_vertical_stride})"
if horizontal_num_slices >= maximum_slices:
recommended_horizontal_stride = max(1, image_w // maximum_slices) + 1
assert (
False
), f"Too computationally expensive with {horizontal_num_slices} slices, try a higher horizontal stride (recommended minimum: {recommended_horizontal_stride})"
for v_slice_idx in range(vertical_num_slices):
v_start = max(0, (v_slice_idx * vertical_stride))
v_end = min(((v_slice_idx + 1) * vertical_stride), image_h)
vertical_slice = image[v_start:v_end, :]
for h_slice_idx in range(horizontal_num_slices):
h_start = max(0, (h_slice_idx * horizontal_stride))
h_end = min(((h_slice_idx + 1) * horizontal_stride), image_w)
horizontal_slice = vertical_slice[:, h_start:h_end]
yield (horizontal_slice, v_start, h_start)
def calculate_box_extents(box):
min_x = box[0][0]
max_x = box[1][0]
min_y = box[0][1]
max_y = box[2][1]
return min_x, max_x, min_y, max_y
def merge_boxes(box1, box2, x_threshold, y_threshold):
min_x1, max_x1, min_y1, max_y1 = calculate_box_extents(box1)
min_x2, max_x2, min_y2, max_y2 = calculate_box_extents(box2)
if (
abs(min_y1 - min_y2) <= y_threshold
and abs(max_y1 - max_y2) <= y_threshold
and abs(max_x1 - min_x2) <= x_threshold
):
new_xmin = min(min_x1, min_x2)
new_xmax = max(max_x1, max_x2)
new_ymin = min(min_y1, min_y2)
new_ymax = max(max_y1, max_y2)
return [
[new_xmin, new_ymin],
[new_xmax, new_ymin],
[new_xmax, new_ymax],
[new_xmin, new_ymax],
]
else:
return None
def merge_fragmented(boxes, x_threshold=10, y_threshold=10):
merged_boxes = []
visited = set()
for i, box1 in enumerate(boxes):
if i in visited:
continue
merged_box = [point[:] for point in box1]
for j, box2 in enumerate(boxes[i + 1 :], start=i + 1):
if j not in visited:
merged_result = merge_boxes(
merged_box, box2, x_threshold=x_threshold, y_threshold=y_threshold
)
if merged_result:
merged_box = merged_result
visited.add(j)
merged_boxes.append(merged_box)
if len(merged_boxes) == len(boxes):
return np.array(merged_boxes)
else:
return merge_fragmented(merged_boxes, x_threshold, y_threshold)
def check_gpu(use_gpu):
if use_gpu and (
not paddle.is_compiled_with_cuda() or paddle.device.get_device() == "cpu"
):
use_gpu = False
return use_gpu
if __name__ == "__main__":
pass