mirror of
https://github.com/RapidAI/RapidOCR.git
synced 2026-11-10 21:10:55 +08:00
feat: support PP-OCRv6 PyTor as the inference engine
This commit is contained in:
1 parent
c18432567f
commit
370f01d0a8
11 files changed
+1100
-288
No files matched your search
+324
-274
File diff suppressed because it is too large.
Load diff
@@ -1,3 +1,133 @@
|
||||
PP-OCRv6_rec_tiny:
|
||||
model_type: rec
|
||||
algorithm: SVTR_LCNet
|
||||
Transform:
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
model_size: tiny
|
||||
Head:
|
||||
name: MultiHead
|
||||
out_channels_list:
|
||||
CTCLabelDecode: 6906 #'blank' + ...(6904) + ' '
|
||||
head_list:
|
||||
- CTCHead:
|
||||
Neck:
|
||||
name: reshape
|
||||
Head:
|
||||
mid_channels: 80
|
||||
use_guide: true
|
||||
fc_decay: 0.00001
|
||||
- NRTRHead:
|
||||
nrtr_dim: 384
|
||||
max_text_length: 25
|
||||
|
||||
PP-OCRv6_rec_small:
|
||||
model_type: rec
|
||||
algorithm: SVTR_LCNet
|
||||
Transform:
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
model_size: small
|
||||
Head:
|
||||
name: MultiHead
|
||||
out_channels_list:
|
||||
CTCLabelDecode: 18710 #'blank' + ...(18708) + ' '
|
||||
head_list:
|
||||
- CTCHead:
|
||||
Neck:
|
||||
name: lightsvtr
|
||||
dims: 120
|
||||
depth: 2
|
||||
mlp_ratio: 2.0
|
||||
local_kernel: 7
|
||||
Head:
|
||||
fc_decay: 0.00001
|
||||
- NRTRHead:
|
||||
nrtr_dim: 384
|
||||
max_text_length: 25
|
||||
|
||||
PP-OCRv6_rec_medium:
|
||||
model_type: rec
|
||||
algorithm: SVTR_LCNet
|
||||
Transform:
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
model_size: medium
|
||||
Head:
|
||||
name: MultiHead
|
||||
out_channels_list:
|
||||
CTCLabelDecode: 18710 #'blank' + ...(18708) + ' '
|
||||
head_list:
|
||||
- CTCHead:
|
||||
Neck:
|
||||
name: lightsvtr
|
||||
dims: 192
|
||||
depth: 2
|
||||
mlp_ratio: 4.0
|
||||
local_kernel: 7
|
||||
use_guide: false
|
||||
Head:
|
||||
fc_decay: 0.00001
|
||||
- NRTRHead:
|
||||
nrtr_dim: 512
|
||||
max_text_length: 25
|
||||
|
||||
PP-OCRv6_det_tiny:
|
||||
model_type: det
|
||||
algorithm: DB
|
||||
Transform: null
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
det: true
|
||||
model_size: tiny
|
||||
Neck:
|
||||
name: RepLKFPN
|
||||
out_channels: 64
|
||||
dilated_kernel_size: 5
|
||||
shortcut: true
|
||||
Head:
|
||||
name: DBHead
|
||||
k: 50
|
||||
fix_nan: true
|
||||
aux_in_channels: 64
|
||||
|
||||
PP-OCRv6_det_small:
|
||||
model_type: det
|
||||
algorithm: DB
|
||||
Transform: null
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
det: true
|
||||
model_size: small
|
||||
Neck:
|
||||
name: RepLKFPN
|
||||
out_channels: 96
|
||||
dilated_kernel_size: 7
|
||||
shortcut: true
|
||||
Head:
|
||||
name: DBHead
|
||||
k: 50
|
||||
fix_nan: true
|
||||
aux_in_channels: 96
|
||||
|
||||
PP-OCRv6_det_medium:
|
||||
model_type: det
|
||||
algorithm: DB
|
||||
Transform: null
|
||||
Backbone:
|
||||
name: PPLCNetV4
|
||||
det: true
|
||||
model_size: medium
|
||||
Neck:
|
||||
name: RepLKPAN
|
||||
out_channels: 256
|
||||
intracl: true
|
||||
Head:
|
||||
name: DBHead
|
||||
k: 50
|
||||
fix_nan: true
|
||||
aux_in_channels: 256
|
||||
|
||||
ch_PP-OCRv5_rec_mobile:
|
||||
model_type: rec
|
||||
algorithm: SVTR_LCNet
|
||||
|
||||
@@ -3,6 +3,7 @@ from torch import nn
|
||||
from ..backbones import build_backbone
|
||||
from ..heads import build_head
|
||||
from ..necks import build_neck
|
||||
from ..transforms import build_transform
|
||||
|
||||
|
||||
class BaseModel(nn.Module):
|
||||
@@ -16,6 +17,19 @@ class BaseModel(nn.Module):
|
||||
|
||||
in_channels = config.get("in_channels", 3)
|
||||
model_type = config["model_type"]
|
||||
# build transfrom,
|
||||
# for rec, transfrom can be TPS,None
|
||||
# for det and cls, transfrom shoule to be None,
|
||||
# if you make model differently, you can use transfrom in det and cls
|
||||
if "Transform" not in config or config["Transform"] is None:
|
||||
self.use_transform = False
|
||||
else:
|
||||
self.use_transform = True
|
||||
config["Transform"]["in_channels"] = in_channels
|
||||
self.transform = build_transform(config["Transform"])
|
||||
in_channels = self.transform.out_channels
|
||||
# raise NotImplementedError
|
||||
|
||||
# build backbone, backbone is need for del, rec and cls
|
||||
if "Backbone" not in config or config["Backbone"] is None:
|
||||
self.use_backbone = False
|
||||
@@ -70,6 +84,8 @@ class BaseModel(nn.Module):
|
||||
|
||||
def forward(self, x):
|
||||
y = dict()
|
||||
if self.use_transform:
|
||||
x = self.transform(x)
|
||||
if self.use_backbone:
|
||||
x = self.backbone(x)
|
||||
if isinstance(x, dict):
|
||||
|
||||
@@ -20,6 +20,7 @@ def build_backbone(config, model_type):
|
||||
from .det_mobilenet_v3 import MobileNetV3
|
||||
from .rec_hgnet import PPHGNet_small
|
||||
from .rec_lcnetv3 import PPLCNetV3
|
||||
from .rec_lcnetv4 import PPLCNetV4
|
||||
from .rec_pphgnetv2 import PPHGNetV2_B4
|
||||
|
||||
support_dict = [
|
||||
@@ -30,13 +31,18 @@ def build_backbone(config, model_type):
|
||||
"PPLCNetV3",
|
||||
"PPHGNet_small",
|
||||
"PPHGNetV2_B4",
|
||||
"PPLCNetV4",
|
||||
]
|
||||
elif model_type == "rec" or model_type == "cls":
|
||||
from .rec_hgnet import PPHGNet_small
|
||||
from .rec_lcnetv3 import PPLCNetV3
|
||||
from .rec_lcnetv4 import PPLCNetV4
|
||||
from .rec_mobilenet_v3 import MobileNetV3
|
||||
from .rec_mv1_enhance import MobileNetV1Enhance
|
||||
from .rec_pphgnetv2 import PPHGNetV2_B4
|
||||
from .rec_pphgnetv2 import (
|
||||
PPHGNetV2_B4,
|
||||
)
|
||||
from .rec_pplcnet import PPLCNet
|
||||
from .rec_svtrnet import SVTRNet
|
||||
|
||||
support_dict = [
|
||||
@@ -52,6 +58,8 @@ def build_backbone(config, model_type):
|
||||
"PPLCNetV3",
|
||||
"PPHGNet_small",
|
||||
"PPHGNetV2_B4",
|
||||
"PPLCNetV4",
|
||||
"PPLCNet",
|
||||
]
|
||||
else:
|
||||
raise NotImplementedError
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
from torch import nn
|
||||
|
||||
|
||||
class Hswish(nn.Module):
|
||||
|
||||
@@ -10,9 +10,28 @@ class CTCHead(nn.Module):
|
||||
fc_decay=0.0004,
|
||||
mid_channels=None,
|
||||
return_feats=False,
|
||||
**kwargs
|
||||
use_guide=False,
|
||||
**kwargs,
|
||||
):
|
||||
super(CTCHead, self).__init__()
|
||||
self.use_guide = use_guide
|
||||
if use_guide:
|
||||
# Depthwise-separable Conv1d block:
|
||||
# DW Conv1d(k=5, groups=in_channels) → BN → ReLU → PW Conv1d(k=1) → BN
|
||||
self.guide_layer = nn.Sequential(
|
||||
nn.Conv1d(
|
||||
in_channels,
|
||||
in_channels,
|
||||
5,
|
||||
padding=2,
|
||||
groups=in_channels,
|
||||
bias=True,
|
||||
),
|
||||
nn.BatchNorm1d(in_channels),
|
||||
nn.ReLU(),
|
||||
nn.Conv1d(in_channels, in_channels, 1, bias=True),
|
||||
nn.BatchNorm1d(in_channels),
|
||||
)
|
||||
if mid_channels is None:
|
||||
self.fc = nn.Linear(
|
||||
in_channels,
|
||||
@@ -36,6 +55,12 @@ class CTCHead(nn.Module):
|
||||
self.return_feats = return_feats
|
||||
|
||||
def forward(self, x, labels=None):
|
||||
if self.use_guide:
|
||||
# x is [B, W, C], guide_layer is Conv1d on [B, C, W]
|
||||
x = x.permute(0, 2, 1) # [B, C, W]
|
||||
x = self.guide_layer(x)
|
||||
x = x.permute(0, 2, 1) # [B, W, C]
|
||||
|
||||
if self.mid_channels is None:
|
||||
predicts = self.fc(x)
|
||||
else:
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
from torch import nn
|
||||
|
||||
from ..necks.rnn import Im2Seq, SequenceEncoder
|
||||
from .rec_ctc_head import CTCHead
|
||||
|
||||
|
||||
class FCTranspose(nn.Module):
|
||||
@@ -42,8 +41,7 @@ class MultiHead(nn.Module):
|
||||
head_args = self.head_list[idx][name].get("Head", {})
|
||||
if head_args is None:
|
||||
head_args = {}
|
||||
|
||||
self.ctc_head = CTCHead(
|
||||
self.ctc_head = eval(name)(
|
||||
in_channels=self.ctc_encoder.out_channels,
|
||||
out_channels=out_channels_list["CTCLabelDecode"],
|
||||
**head_args,
|
||||
@@ -54,8 +52,7 @@ class MultiHead(nn.Module):
|
||||
def forward(self, x, data=None):
|
||||
ctc_encoder = self.ctc_encoder(x)
|
||||
ctc_out = self.ctc_head(ctc_encoder)
|
||||
|
||||
head_out = {}
|
||||
head_out = dict()
|
||||
head_out["ctc"] = ctc_out
|
||||
head_out["res"] = ctc_out
|
||||
head_out["ctc_neck"] = ctc_encoder
|
||||
|
||||
@@ -16,10 +16,23 @@ __all__ = ["build_neck"]
|
||||
|
||||
|
||||
def build_neck(config):
|
||||
from .db_fpn import DBFPN, LKPAN, RSEFPN
|
||||
from .db_fpn import DBFPN, LKPAN, RSEFPN, RepLKFPN, RepLKPAN
|
||||
from .rnn import SequenceEncoder
|
||||
|
||||
support_dict = ["DBFPN", "SequenceEncoder", "RSEFPN", "LKPAN"]
|
||||
support_dict = [
|
||||
"FPN",
|
||||
"DBFPN",
|
||||
"EASTFPN",
|
||||
"SASTFPN",
|
||||
"SequenceEncoder",
|
||||
"PGFPN",
|
||||
"TableFPN",
|
||||
"RSEFPN",
|
||||
"LKPAN",
|
||||
"FCEFPN",
|
||||
"RepLKFPN",
|
||||
"RepLKPAN",
|
||||
]
|
||||
|
||||
module_name = config.pop("name")
|
||||
assert module_name in support_dict, Exception(
|
||||
|
||||
@@ -454,3 +454,459 @@ class ASFBlock(nn.Module):
|
||||
for i in range(self.out_features_num):
|
||||
out_list.append(attention_scores[:, i : i + 1] * features_list[i])
|
||||
return torch.cat(out_list, dim=1)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# PP-OCRv6 new components: DilatedReparamBlock, RepLKFPN, RepLKPAN
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class DilatedReparamBlock(nn.Module):
|
||||
"""
|
||||
Dilated Reparam Block from UniRepLKNet.
|
||||
Reference: https://github.com/AILab-CVC/UniRepLKNet
|
||||
|
||||
Training: uses multiple parallel dilated depthwise convolutions + BN
|
||||
Inference: all branches merge into a single large-kernel depthwise conv
|
||||
|
||||
For kernel_size=7, the branches are:
|
||||
- origin: 7x7 DW Conv (dil=1)
|
||||
- branch1: 5x5 DW Conv (dil=1, equiv RF=5)
|
||||
- branch2: 3x3 DW Conv (dil=2, equiv RF=5)
|
||||
- branch3: 3x3 DW Conv (dil=3, equiv RF=7)
|
||||
"""
|
||||
|
||||
def __init__(self, channels, kernel_size=7, deploy=False):
|
||||
super(DilatedReparamBlock, self).__init__()
|
||||
self.channels = channels
|
||||
self.kernel_size = kernel_size
|
||||
self.is_repped = deploy
|
||||
|
||||
if kernel_size == 9:
|
||||
self.kernel_sizes = [5, 5, 3, 3]
|
||||
self.dilates = [1, 2, 3, 4]
|
||||
elif kernel_size == 7:
|
||||
self.kernel_sizes = [5, 3, 3]
|
||||
self.dilates = [1, 2, 3]
|
||||
elif kernel_size == 5:
|
||||
self.kernel_sizes = [3, 3]
|
||||
self.dilates = [1, 2]
|
||||
elif kernel_size == 11:
|
||||
self.kernel_sizes = [5, 5, 3, 3, 3]
|
||||
self.dilates = [1, 2, 3, 4, 5]
|
||||
elif kernel_size == 13:
|
||||
self.kernel_sizes = [5, 7, 3, 3, 3]
|
||||
self.dilates = [1, 2, 3, 4, 5]
|
||||
else:
|
||||
raise ValueError(
|
||||
"DilatedReparamBlock requires kernel_size in [5,7,9,11,13], "
|
||||
"but got {}".format(kernel_size)
|
||||
)
|
||||
|
||||
if not self.is_repped:
|
||||
self.lk_origin = nn.Conv2d(
|
||||
in_channels=channels,
|
||||
out_channels=channels,
|
||||
kernel_size=kernel_size,
|
||||
stride=1,
|
||||
padding=kernel_size // 2,
|
||||
groups=channels,
|
||||
bias=False,
|
||||
)
|
||||
self.origin_bn = nn.BatchNorm2d(channels)
|
||||
|
||||
for k, r in zip(self.kernel_sizes, self.dilates):
|
||||
equiv_ks = r * (k - 1) + 1
|
||||
p = equiv_ks // 2
|
||||
conv = nn.Conv2d(
|
||||
in_channels=channels,
|
||||
out_channels=channels,
|
||||
kernel_size=k,
|
||||
stride=1,
|
||||
padding=p,
|
||||
dilation=r,
|
||||
groups=channels,
|
||||
bias=False,
|
||||
)
|
||||
bn = nn.BatchNorm2d(channels)
|
||||
setattr(self, "dil_conv_k{}_{}".format(k, r), conv)
|
||||
setattr(self, "dil_bn_k{}_{}".format(k, r), bn)
|
||||
else:
|
||||
self.lk_origin = nn.Conv2d(
|
||||
in_channels=channels,
|
||||
out_channels=channels,
|
||||
kernel_size=kernel_size,
|
||||
stride=1,
|
||||
padding=kernel_size // 2,
|
||||
groups=channels,
|
||||
bias=True,
|
||||
)
|
||||
|
||||
def forward(self, x):
|
||||
if self.is_repped:
|
||||
return self.lk_origin(x)
|
||||
out = self.origin_bn(self.lk_origin(x))
|
||||
for k, r in zip(self.kernel_sizes, self.dilates):
|
||||
conv = getattr(self, "dil_conv_k{}_{}".format(k, r))
|
||||
bn = getattr(self, "dil_bn_k{}_{}".format(k, r))
|
||||
out = out + bn(conv(x))
|
||||
return out
|
||||
|
||||
@staticmethod
|
||||
def _fuse_bn(conv, bn):
|
||||
"""Fuse Conv2d + BatchNorm2d into (weight, bias)."""
|
||||
kernel = conv.weight
|
||||
gamma = bn.weight
|
||||
beta = bn.bias
|
||||
running_mean = bn.running_mean
|
||||
running_var = bn.running_var
|
||||
eps = bn.eps
|
||||
std = torch.sqrt(running_var + eps)
|
||||
fused_weight = kernel * (gamma / std).reshape([-1, 1, 1, 1])
|
||||
fused_bias = beta - running_mean * gamma / std
|
||||
return fused_weight, fused_bias
|
||||
|
||||
@staticmethod
|
||||
def _convert_dilated_to_nondilated(kernel, dilate_rate):
|
||||
"""Convert dilated conv kernel to equivalent non-dilated (sparse) kernel
|
||||
by inserting zeros using transposed convolution."""
|
||||
if dilate_rate == 1:
|
||||
return kernel
|
||||
identity = torch.ones([1, 1, 1, 1], dtype=kernel.dtype, device=kernel.device)
|
||||
# F.conv_transpose2d with stride=dilate_rate inserts zeros
|
||||
C = kernel.shape[0]
|
||||
result_list = []
|
||||
for i in range(C):
|
||||
k_i = kernel[i : i + 1] # (1, 1, kH, kW)
|
||||
dilated = F.conv_transpose2d(k_i, identity, stride=dilate_rate)
|
||||
result_list.append(dilated)
|
||||
return torch.cat(result_list, dim=0)
|
||||
|
||||
@staticmethod
|
||||
def _merge_dilated_into_large_kernel(large_kernel, dilated_kernel, dilated_r):
|
||||
"""Pad dilated equivalent kernel to large kernel size and add."""
|
||||
large_k = large_kernel.shape[2]
|
||||
dilated_k = dilated_kernel.shape[2]
|
||||
equiv_ks = dilated_r * (dilated_k - 1) + 1
|
||||
equiv_kernel = DilatedReparamBlock._convert_dilated_to_nondilated(
|
||||
dilated_kernel, dilated_r
|
||||
)
|
||||
rows_to_pad = large_k // 2 - equiv_ks // 2
|
||||
if rows_to_pad > 0:
|
||||
merged = large_kernel + F.pad(
|
||||
equiv_kernel, [rows_to_pad, rows_to_pad, rows_to_pad, rows_to_pad]
|
||||
)
|
||||
else:
|
||||
merged = large_kernel + equiv_kernel
|
||||
return merged
|
||||
|
||||
@torch.no_grad()
|
||||
def rep(self):
|
||||
"""Merge all parallel branches into a single large-kernel DW conv."""
|
||||
if self.is_repped:
|
||||
return
|
||||
origin_k, origin_b = self._fuse_bn(self.lk_origin, self.origin_bn)
|
||||
for k, r in zip(self.kernel_sizes, self.dilates):
|
||||
conv = getattr(self, "dil_conv_k{}_{}".format(k, r))
|
||||
bn = getattr(self, "dil_bn_k{}_{}".format(k, r))
|
||||
branch_k, branch_b = self._fuse_bn(conv, bn)
|
||||
origin_k = self._merge_dilated_into_large_kernel(origin_k, branch_k, r)
|
||||
origin_b = origin_b + branch_b
|
||||
|
||||
merged_conv = nn.Conv2d(
|
||||
in_channels=self.channels,
|
||||
out_channels=self.channels,
|
||||
kernel_size=self.kernel_size,
|
||||
stride=1,
|
||||
padding=self.kernel_size // 2,
|
||||
groups=self.channels,
|
||||
bias=True,
|
||||
)
|
||||
merged_conv.weight.data.copy_(origin_k)
|
||||
merged_conv.bias.data.copy_(origin_b)
|
||||
self.lk_origin = merged_conv
|
||||
self.is_repped = True
|
||||
|
||||
delattr(self, "origin_bn")
|
||||
for k, r in zip(self.kernel_sizes, self.dilates):
|
||||
delattr(self, "dil_conv_k{}_{}".format(k, r))
|
||||
delattr(self, "dil_bn_k{}_{}".format(k, r))
|
||||
|
||||
|
||||
class DilatedReparamConv(nn.Module):
|
||||
"""
|
||||
A drop-in replacement for standard Conv2d (in_ch -> out_ch, large kernel)
|
||||
using DilatedReparamBlock (depthwise) + 1x1 pointwise convolution.
|
||||
|
||||
Architecture:
|
||||
input(in_ch) -> DilatedReparamBlock(in_ch, DW, kernel_size) -> 1x1 Conv(in_ch->out_ch) -> BN
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, in_channels, out_channels, kernel_size=9, deploy=False, **kwargs
|
||||
):
|
||||
super(DilatedReparamConv, self).__init__()
|
||||
self.is_repped = False
|
||||
self.dw = DilatedReparamBlock(
|
||||
channels=in_channels, kernel_size=kernel_size, deploy=deploy
|
||||
)
|
||||
self.pw = nn.Conv2d(
|
||||
in_channels=in_channels,
|
||||
out_channels=out_channels,
|
||||
kernel_size=1,
|
||||
bias=False,
|
||||
)
|
||||
self.bn = nn.BatchNorm2d(out_channels)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.dw(x)
|
||||
x = self.pw(x)
|
||||
if not self.is_repped:
|
||||
x = self.bn(x)
|
||||
return x
|
||||
|
||||
@torch.no_grad()
|
||||
def rep(self):
|
||||
"""Fuse DW branches + PW Conv + BN for deployment."""
|
||||
if self.is_repped:
|
||||
return
|
||||
self.dw.rep()
|
||||
# Fuse pw(Conv2d, no bias) + bn(BatchNorm2d) into single Conv2d with bias
|
||||
conv, bn = self.pw, self.bn
|
||||
gamma = bn.weight
|
||||
std = torch.sqrt(bn.running_var + bn.eps)
|
||||
scale = gamma / std
|
||||
w = conv.weight * scale[:, None, None, None]
|
||||
b = bn.bias - bn.running_mean * scale
|
||||
fused = nn.Conv2d(
|
||||
conv.in_channels,
|
||||
conv.out_channels,
|
||||
conv.kernel_size,
|
||||
stride=conv.stride,
|
||||
padding=conv.padding,
|
||||
dilation=conv.dilation,
|
||||
groups=conv.groups,
|
||||
)
|
||||
fused.weight.data.copy_(w)
|
||||
fused.bias.data.copy_(b)
|
||||
self.pw = fused
|
||||
del self.bn
|
||||
self.is_repped = True
|
||||
|
||||
|
||||
class RepLKFPN(nn.Module):
|
||||
"""Optimized RSEFPN: replaces 3x3 standard Conv in inp_conv with
|
||||
DilatedReparamBlock (DW, kernel_size) + PWConv 1x1 + SE.
|
||||
|
||||
Changes vs RSEFPN:
|
||||
- inp_conv: RSELayer(3x3 std Conv + SE)
|
||||
-> DilatedReparamBlock(DW kernel_size) + PWConv(1x1) + SE
|
||||
- ins_conv: unchanged (1x1 Conv, no benefit from DW decomposition)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, in_channels, out_channels, shortcut=True, dilated_kernel_size=7, **kwargs
|
||||
):
|
||||
super(RepLKFPN, self).__init__()
|
||||
self.out_channels = out_channels
|
||||
self.is_repped = False
|
||||
self.ins_conv = nn.ModuleList()
|
||||
self.inp_conv_dw = nn.ModuleList()
|
||||
self.inp_conv_pw = nn.ModuleList()
|
||||
self.inp_conv_se = nn.ModuleList()
|
||||
self.shortcut = shortcut
|
||||
|
||||
self.intracl = False
|
||||
if "intracl" in kwargs and kwargs["intracl"] is True:
|
||||
self.intracl = kwargs["intracl"]
|
||||
self.incl1 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl2 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl3 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl4 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
|
||||
for i in range(len(in_channels)):
|
||||
self.ins_conv.append(
|
||||
RSELayer(in_channels[i], out_channels, kernel_size=1, shortcut=shortcut)
|
||||
)
|
||||
|
||||
self.inp_conv_dw.append(
|
||||
DilatedReparamBlock(
|
||||
channels=out_channels, kernel_size=dilated_kernel_size
|
||||
)
|
||||
)
|
||||
|
||||
self.inp_conv_pw.append(
|
||||
nn.Conv2d(
|
||||
in_channels=out_channels,
|
||||
out_channels=out_channels // 4,
|
||||
kernel_size=1,
|
||||
bias=False,
|
||||
)
|
||||
)
|
||||
|
||||
self.inp_conv_se.append(SEModule(out_channels // 4))
|
||||
|
||||
def _inp_forward(self, x, idx):
|
||||
x = self.inp_conv_dw[idx](x)
|
||||
x = self.inp_conv_pw[idx](x)
|
||||
if self.shortcut:
|
||||
x = x + self.inp_conv_se[idx](x)
|
||||
else:
|
||||
x = self.inp_conv_se[idx](x)
|
||||
return x
|
||||
|
||||
def forward(self, x):
|
||||
c2, c3, c4, c5 = x
|
||||
|
||||
in5 = self.ins_conv[3](c5)
|
||||
in4 = self.ins_conv[2](c4)
|
||||
in3 = self.ins_conv[1](c3)
|
||||
in2 = self.ins_conv[0](c2)
|
||||
|
||||
out4 = in4 + F.interpolate(in5, scale_factor=2, mode="nearest") # 1/16
|
||||
out3 = in3 + F.interpolate(out4, scale_factor=2, mode="nearest") # 1/8
|
||||
out2 = in2 + F.interpolate(out3, scale_factor=2, mode="nearest") # 1/4
|
||||
|
||||
p5 = self._inp_forward(in5, 3)
|
||||
p4 = self._inp_forward(out4, 2)
|
||||
p3 = self._inp_forward(out3, 1)
|
||||
p2 = self._inp_forward(out2, 0)
|
||||
|
||||
if self.intracl is True:
|
||||
p5 = self.incl4(p5)
|
||||
p4 = self.incl3(p4)
|
||||
p3 = self.incl2(p3)
|
||||
p2 = self.incl1(p2)
|
||||
|
||||
p5 = F.interpolate(p5, scale_factor=8, mode="nearest")
|
||||
p4 = F.interpolate(p4, scale_factor=4, mode="nearest")
|
||||
p3 = F.interpolate(p3, scale_factor=2, mode="nearest")
|
||||
|
||||
fuse = torch.cat([p5, p4, p3, p2], dim=1)
|
||||
if self.training:
|
||||
return {"fuse": fuse, "aux_p4": out4, "aux_p3": out3, "aux_p2": out2}
|
||||
return fuse
|
||||
|
||||
def rep(self):
|
||||
"""Merge DilatedReparamBlock branches for inference deployment."""
|
||||
if self.is_repped:
|
||||
return
|
||||
for i in range(len(self.inp_conv_dw)):
|
||||
self.inp_conv_dw[i].rep()
|
||||
self.is_repped = True
|
||||
|
||||
|
||||
class RepLKPAN(nn.Module):
|
||||
"""
|
||||
Optimized LKPAN using UniRepLKNet's DilatedReparamBlock.
|
||||
|
||||
Replaces the 8 standard 9x9 Conv2d in LKPAN (4 inp_conv + 4 pan_lat_conv)
|
||||
with DilatedReparamConv (DW large-kernel reparam + 1x1 pointwise).
|
||||
"""
|
||||
|
||||
def __init__(self, in_channels, out_channels, mode="large", **kwargs):
|
||||
super(RepLKPAN, self).__init__()
|
||||
self.out_channels = out_channels
|
||||
self.is_repped = False
|
||||
|
||||
self.ins_conv = nn.ModuleList()
|
||||
self.inp_conv = nn.ModuleList()
|
||||
# pan head
|
||||
self.pan_head_conv = nn.ModuleList()
|
||||
self.pan_lat_conv = nn.ModuleList()
|
||||
|
||||
for i in range(len(in_channels)):
|
||||
self.ins_conv.append(
|
||||
nn.Conv2d(
|
||||
in_channels=in_channels[i],
|
||||
out_channels=self.out_channels,
|
||||
kernel_size=1,
|
||||
bias=False,
|
||||
)
|
||||
)
|
||||
|
||||
self.inp_conv.append(
|
||||
DilatedReparamConv(
|
||||
in_channels=self.out_channels,
|
||||
out_channels=self.out_channels // 4,
|
||||
kernel_size=9,
|
||||
)
|
||||
)
|
||||
|
||||
if i > 0:
|
||||
self.pan_head_conv.append(
|
||||
nn.Conv2d(
|
||||
in_channels=self.out_channels // 4,
|
||||
out_channels=self.out_channels // 4,
|
||||
kernel_size=3,
|
||||
padding=1,
|
||||
stride=2,
|
||||
bias=False,
|
||||
)
|
||||
)
|
||||
self.pan_lat_conv.append(
|
||||
DilatedReparamConv(
|
||||
in_channels=self.out_channels // 4,
|
||||
out_channels=self.out_channels // 4,
|
||||
kernel_size=9,
|
||||
)
|
||||
)
|
||||
|
||||
self.intracl = False
|
||||
if "intracl" in kwargs and kwargs["intracl"] is True:
|
||||
self.intracl = kwargs["intracl"]
|
||||
self.incl1 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl2 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl3 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
self.incl4 = IntraCLBlock(self.out_channels // 4, reduce_factor=2)
|
||||
|
||||
def forward(self, x):
|
||||
c2, c3, c4, c5 = x
|
||||
|
||||
in5 = self.ins_conv[3](c5)
|
||||
in4 = self.ins_conv[2](c4)
|
||||
in3 = self.ins_conv[1](c3)
|
||||
in2 = self.ins_conv[0](c2)
|
||||
|
||||
out4 = in4 + F.interpolate(in5, scale_factor=2, mode="nearest") # 1/16
|
||||
out3 = in3 + F.interpolate(out4, scale_factor=2, mode="nearest") # 1/8
|
||||
out2 = in2 + F.interpolate(out3, scale_factor=2, mode="nearest") # 1/4
|
||||
|
||||
f5 = self.inp_conv[3](in5)
|
||||
f4 = self.inp_conv[2](out4)
|
||||
f3 = self.inp_conv[1](out3)
|
||||
f2 = self.inp_conv[0](out2)
|
||||
|
||||
pan3 = f3 + self.pan_head_conv[0](f2)
|
||||
pan4 = f4 + self.pan_head_conv[1](pan3)
|
||||
pan5 = f5 + self.pan_head_conv[2](pan4)
|
||||
|
||||
p2 = self.pan_lat_conv[0](f2)
|
||||
p3 = self.pan_lat_conv[1](pan3)
|
||||
p4 = self.pan_lat_conv[2](pan4)
|
||||
p5 = self.pan_lat_conv[3](pan5)
|
||||
|
||||
if self.intracl is True:
|
||||
p5 = self.incl4(p5)
|
||||
p4 = self.incl3(p4)
|
||||
p3 = self.incl2(p3)
|
||||
p2 = self.incl1(p2)
|
||||
|
||||
p5 = F.interpolate(p5, scale_factor=8, mode="nearest")
|
||||
p4 = F.interpolate(p4, scale_factor=4, mode="nearest")
|
||||
p3 = F.interpolate(p3, scale_factor=2, mode="nearest")
|
||||
|
||||
fuse = torch.cat([p5, p4, p3, p2], dim=1)
|
||||
if self.training:
|
||||
return {"fuse": fuse, "aux_p4": out4, "aux_p3": out3, "aux_p2": out2}
|
||||
return fuse
|
||||
|
||||
def rep(self):
|
||||
"""Merge all DilatedReparamBlock branches and fuse PW+BN for deployment."""
|
||||
if self.is_repped:
|
||||
return
|
||||
for i in range(len(self.inp_conv)):
|
||||
self.inp_conv[i].rep()
|
||||
for i in range(len(self.pan_lat_conv)):
|
||||
self.pan_lat_conv[i].rep()
|
||||
self.is_repped = True
|
||||
@@ -186,6 +186,110 @@ class EncoderWithSVTR(nn.Module):
|
||||
return z
|
||||
|
||||
|
||||
class EncoderWithLightSVTR(nn.Module):
|
||||
"""Light SVTR neck: attention with lightweight skip connection.
|
||||
|
||||
Replaces heavy guide fusion with a lightweight 1x1 skip conv + add.
|
||||
Adds a DWConv before attention for local context priming.
|
||||
|
||||
When use_guide=True, applies detach to input so that neck
|
||||
gradients do not propagate back to the backbone (implicit regularization).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
in_channels,
|
||||
dims=64,
|
||||
depth=1,
|
||||
num_heads=8,
|
||||
qkv_bias=True,
|
||||
mlp_ratio=4.0,
|
||||
drop_rate=0.1,
|
||||
attn_drop_rate=0.1,
|
||||
drop_path=0.0,
|
||||
qk_scale=None,
|
||||
local_kernel=7,
|
||||
use_guide=False,
|
||||
):
|
||||
super().__init__()
|
||||
self.use_guide = use_guide
|
||||
self.conv_reduce = ConvBNLayer(
|
||||
in_channels,
|
||||
dims,
|
||||
kernel_size=1,
|
||||
act="swish",
|
||||
)
|
||||
self.local_conv = nn.Sequential(
|
||||
nn.Conv2d(
|
||||
dims,
|
||||
dims,
|
||||
[1, local_kernel],
|
||||
padding=[0, local_kernel // 2],
|
||||
groups=dims,
|
||||
bias=False,
|
||||
),
|
||||
nn.BatchNorm2d(dims),
|
||||
nn.SiLU(),
|
||||
)
|
||||
|
||||
self.svtr_block = nn.ModuleList(
|
||||
[
|
||||
Block(
|
||||
dim=dims,
|
||||
num_heads=num_heads,
|
||||
mixer="Global",
|
||||
HW=None,
|
||||
mlp_ratio=mlp_ratio,
|
||||
qkv_bias=qkv_bias,
|
||||
qk_scale=qk_scale,
|
||||
drop=drop_rate,
|
||||
act_layer="swish",
|
||||
attn_drop=attn_drop_rate,
|
||||
drop_path=drop_path,
|
||||
norm_layer="nn.LayerNorm",
|
||||
epsilon=1e-05,
|
||||
prenorm=False,
|
||||
)
|
||||
for _ in range(depth)
|
||||
]
|
||||
)
|
||||
self.norm = nn.LayerNorm(dims, eps=1e-6)
|
||||
|
||||
self.skip_conv = ConvBNLayer(
|
||||
in_channels,
|
||||
dims,
|
||||
kernel_size=1,
|
||||
act="swish",
|
||||
)
|
||||
|
||||
self.out_channels = dims
|
||||
self.apply(self._init_weights)
|
||||
|
||||
def _init_weights(self, m):
|
||||
if isinstance(m, nn.Linear):
|
||||
nn.init.trunc_normal_(m.weight, std=0.02)
|
||||
if m.bias is not None:
|
||||
nn.init.zeros_(m.bias)
|
||||
elif isinstance(m, nn.LayerNorm):
|
||||
nn.init.zeros_(m.bias)
|
||||
nn.init.ones_(m.weight)
|
||||
|
||||
def forward(self, x):
|
||||
if self.use_guide:
|
||||
x = x.clone().detach()
|
||||
skip = self.skip_conv(x)
|
||||
z = self.conv_reduce(x)
|
||||
z = z + self.local_conv(z)
|
||||
B, C, H, W = z.shape
|
||||
z = z.flatten(2).permute(0, 2, 1)
|
||||
for blk in self.svtr_block:
|
||||
z = blk(z)
|
||||
z = self.norm(z)
|
||||
z = z.reshape(-1, H, W, C).permute(0, 3, 1, 2)
|
||||
z = z + skip
|
||||
return z
|
||||
|
||||
|
||||
class SequenceEncoder(nn.Module):
|
||||
def __init__(self, in_channels, encoder_type, hidden_size=48, **kwargs):
|
||||
super(SequenceEncoder, self).__init__()
|
||||
@@ -200,12 +304,13 @@ class SequenceEncoder(nn.Module):
|
||||
"fc": EncoderWithFC,
|
||||
"rnn": EncoderWithRNN,
|
||||
"svtr": EncoderWithSVTR,
|
||||
"lightsvtr": EncoderWithLightSVTR,
|
||||
}
|
||||
assert encoder_type in support_encoder_dict, "{} must in {}".format(
|
||||
encoder_type, support_encoder_dict.keys()
|
||||
)
|
||||
|
||||
if encoder_type == "svtr":
|
||||
if encoder_type in ("svtr", "lightsvtr"):
|
||||
self.encoder = support_encoder_dict[encoder_type](
|
||||
self.encoder_reshape.out_channels, **kwargs
|
||||
)
|
||||
@@ -217,7 +322,7 @@ class SequenceEncoder(nn.Module):
|
||||
self.only_reshape = False
|
||||
|
||||
def forward(self, x):
|
||||
if self.encoder_type != "svtr":
|
||||
if self.encoder_type not in ("svtr", "lightsvtr"):
|
||||
x = self.encoder_reshape(x)
|
||||
if not self.only_reshape:
|
||||
x = self.encoder(x)
|
||||
|
||||
@@ -143,7 +143,13 @@ def test_engine_tensorrt():
|
||||
|
||||
@mark.parametrize(
|
||||
"engine_type",
|
||||
[EngineType.ONNXRUNTIME, EngineType.OPENVINO, EngineType.PADDLE],
|
||||
[
|
||||
EngineType.ONNXRUNTIME,
|
||||
EngineType.OPENVINO,
|
||||
EngineType.PADDLE,
|
||||
EngineType.TORCH,
|
||||
EngineType.MNN,
|
||||
],
|
||||
)
|
||||
def test_ppocrv6_det_small(engine_type):
|
||||
engine = RapidOCR(
|
||||
@@ -161,7 +167,13 @@ def test_ppocrv6_det_small(engine_type):
|
||||
|
||||
@mark.parametrize(
|
||||
"engine_type",
|
||||
[EngineType.ONNXRUNTIME, EngineType.OPENVINO, EngineType.PADDLE],
|
||||
[
|
||||
EngineType.ONNXRUNTIME,
|
||||
EngineType.OPENVINO,
|
||||
EngineType.PADDLE,
|
||||
EngineType.TORCH,
|
||||
EngineType.MNN,
|
||||
],
|
||||
)
|
||||
def test_ppocrv6_rec_small(engine_type):
|
||||
engine = RapidOCR(
|
||||
|
||||
Reference in new issue
Block a user