xslandClaude Opus 4.8
8d3b145111
worker 侧从 Mock 替换为真实算法： - face_analysis 包：detector(MediaPipe 478点) / pose(solvePnP 姿态) / calibration(虹膜直径法) / hair_segmenter+bisenet_model(方案B 头发分割) / measure(方案A兜底+B/A决策+七眼+换算) / annotation(numpy渐变线+中文标注) - app.py：/api/v1/face/measure 接真实实现，返回 annotated_image_base64 （不落盘不拼URL，落盘由网关做）；加 X-Internal-Token 鉴权、/health 就绪态、 可配置分辨率门槛、异常兜底 - 部署：start.sh/run_worker.sh/hair-worker.service 监听 8187；worker_config 示例 - 测试 tests/：Tier1合成真值<1e-6 + Tier2缩放不变 + Tier3叠加 + 错误码集成 + 数值回归，pytest 24 项全绿 - 文档补实测基线表 + RTX5090/torch 说明 注：worker 为 RTX 5090(sm_120)，pinned torch 2.2.2(cu121) 只到 sm_90， BiSeNet 已自动回退 CPU（方案B 正常）；要用 GPU 需换 torch cu128(≥2.7)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
216 lines
8.1 KiB
Python
216 lines
8.1 KiB
Python
"""BiSeNet (face-parsing.PyTorch) 网络结构，vendored。
|
||
|
||
源自 zllrunning/face-parsing.PyTorch，结构与权重 `79999_iter.pth`（CelebAMask-HQ
|
||
19 类）严格对应，仅修改 resnet18 骨干加载为「优先本地权重」以适配内网离线。
|
||
hair 类别索引 = 17。
|
||
"""
|
||
import os
|
||
|
||
import torch
|
||
import torch.nn as nn
|
||
import torch.nn.functional as F
|
||
import torch.utils.model_zoo as modelzoo
|
||
|
||
resnet18_url = "https://download.pytorch.org/models/resnet18-5c106cde.pth"
|
||
_LOCAL_RESNET18 = os.path.join(os.path.dirname(__file__), "weights", "resnet18-5c106cde.pth")
|
||
|
||
|
||
def conv3x3(in_planes, out_planes, stride=1):
|
||
return nn.Conv2d(in_planes, out_planes, kernel_size=3, stride=stride, padding=1, bias=False)
|
||
|
||
|
||
class BasicBlock(nn.Module):
|
||
def __init__(self, in_chan, out_chan, stride=1):
|
||
super(BasicBlock, self).__init__()
|
||
self.conv1 = conv3x3(in_chan, out_chan, stride)
|
||
self.bn1 = nn.BatchNorm2d(out_chan)
|
||
self.conv2 = conv3x3(out_chan, out_chan)
|
||
self.bn2 = nn.BatchNorm2d(out_chan)
|
||
self.relu = nn.ReLU(inplace=True)
|
||
self.downsample = None
|
||
if in_chan != out_chan or stride != 1:
|
||
self.downsample = nn.Sequential(
|
||
nn.Conv2d(in_chan, out_chan, kernel_size=1, stride=stride, bias=False),
|
||
nn.BatchNorm2d(out_chan),
|
||
)
|
||
|
||
def forward(self, x):
|
||
residual = self.conv1(x)
|
||
residual = F.relu(self.bn1(residual))
|
||
residual = self.conv2(residual)
|
||
residual = self.bn2(residual)
|
||
shortcut = x
|
||
if self.downsample is not None:
|
||
shortcut = self.downsample(x)
|
||
out = shortcut + residual
|
||
out = self.relu(out)
|
||
return out
|
||
|
||
|
||
def create_layer_basic(in_chan, out_chan, bnum, stride=1):
|
||
layers = [BasicBlock(in_chan, out_chan, stride=stride)]
|
||
for _ in range(bnum - 1):
|
||
layers.append(BasicBlock(out_chan, out_chan, stride=1))
|
||
return nn.Sequential(*layers)
|
||
|
||
|
||
class Resnet18(nn.Module):
|
||
def __init__(self):
|
||
super(Resnet18, self).__init__()
|
||
self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False)
|
||
self.bn1 = nn.BatchNorm2d(64)
|
||
self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
|
||
self.layer1 = create_layer_basic(64, 64, bnum=2, stride=1)
|
||
self.layer2 = create_layer_basic(64, 128, bnum=2, stride=2)
|
||
self.layer3 = create_layer_basic(128, 256, bnum=2, stride=2)
|
||
self.layer4 = create_layer_basic(256, 512, bnum=2, stride=2)
|
||
self.init_weight()
|
||
|
||
def forward(self, x):
|
||
x = self.conv1(x)
|
||
x = F.relu(self.bn1(x))
|
||
x = self.maxpool(x)
|
||
x = self.layer1(x)
|
||
feat8 = self.layer2(x) # 1/8
|
||
feat16 = self.layer3(feat8) # 1/16
|
||
feat32 = self.layer4(feat16) # 1/32
|
||
return feat8, feat16, feat32
|
||
|
||
def init_weight(self):
|
||
# 优先本地骨干权重（内网离线），缺失才回退 torch model_zoo（会查缓存）。
|
||
if os.path.isfile(_LOCAL_RESNET18):
|
||
state_dict = torch.load(_LOCAL_RESNET18, map_location="cpu")
|
||
else:
|
||
state_dict = modelzoo.load_url(resnet18_url)
|
||
self_state_dict = self.state_dict()
|
||
for k, v in state_dict.items():
|
||
if "fc" in k:
|
||
continue
|
||
self_state_dict.update({k: v})
|
||
self.load_state_dict(self_state_dict)
|
||
|
||
|
||
class ConvBNReLU(nn.Module):
|
||
def __init__(self, in_chan, out_chan, ks=3, stride=1, padding=1):
|
||
super(ConvBNReLU, self).__init__()
|
||
self.conv = nn.Conv2d(in_chan, out_chan, kernel_size=ks, stride=stride,
|
||
padding=padding, bias=False)
|
||
self.bn = nn.BatchNorm2d(out_chan)
|
||
|
||
def forward(self, x):
|
||
x = self.conv(x)
|
||
x = F.relu(self.bn(x))
|
||
return x
|
||
|
||
|
||
class BiSeNetOutput(nn.Module):
|
||
def __init__(self, in_chan, mid_chan, n_classes):
|
||
super(BiSeNetOutput, self).__init__()
|
||
self.conv = ConvBNReLU(in_chan, mid_chan, ks=3, stride=1, padding=1)
|
||
self.conv_out = nn.Conv2d(mid_chan, n_classes, kernel_size=1, bias=False)
|
||
|
||
def forward(self, x):
|
||
x = self.conv(x)
|
||
x = self.conv_out(x)
|
||
return x
|
||
|
||
|
||
class AttentionRefinementModule(nn.Module):
|
||
def __init__(self, in_chan, out_chan):
|
||
super(AttentionRefinementModule, self).__init__()
|
||
self.conv = ConvBNReLU(in_chan, out_chan, ks=3, stride=1, padding=1)
|
||
self.conv_atten = nn.Conv2d(out_chan, out_chan, kernel_size=1, bias=False)
|
||
self.bn_atten = nn.BatchNorm2d(out_chan)
|
||
self.sigmoid_atten = nn.Sigmoid()
|
||
|
||
def forward(self, x):
|
||
feat = self.conv(x)
|
||
atten = F.avg_pool2d(feat, feat.size()[2:])
|
||
atten = self.conv_atten(atten)
|
||
atten = self.bn_atten(atten)
|
||
atten = self.sigmoid_atten(atten)
|
||
out = torch.mul(feat, atten)
|
||
return out
|
||
|
||
|
||
class ContextPath(nn.Module):
|
||
def __init__(self):
|
||
super(ContextPath, self).__init__()
|
||
self.resnet = Resnet18()
|
||
self.arm16 = AttentionRefinementModule(256, 128)
|
||
self.arm32 = AttentionRefinementModule(512, 128)
|
||
self.conv_head32 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
|
||
self.conv_head16 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
|
||
self.conv_avg = ConvBNReLU(512, 128, ks=1, stride=1, padding=0)
|
||
|
||
def forward(self, x):
|
||
feat8, feat16, feat32 = self.resnet(x)
|
||
h8, w8 = feat8.size()[2:]
|
||
h16, w16 = feat16.size()[2:]
|
||
h32, w32 = feat32.size()[2:]
|
||
|
||
avg = F.avg_pool2d(feat32, feat32.size()[2:])
|
||
avg = self.conv_avg(avg)
|
||
avg_up = F.interpolate(avg, (h32, w32), mode="nearest")
|
||
|
||
feat32_arm = self.arm32(feat32)
|
||
feat32_sum = feat32_arm + avg_up
|
||
feat32_up = F.interpolate(feat32_sum, (h16, w16), mode="nearest")
|
||
feat32_up = self.conv_head32(feat32_up)
|
||
|
||
feat16_arm = self.arm16(feat16)
|
||
feat16_sum = feat16_arm + feat32_up
|
||
feat16_up = F.interpolate(feat16_sum, (h8, w8), mode="nearest")
|
||
feat16_up = self.conv_head16(feat16_up)
|
||
|
||
return feat8, feat16_up, feat32_up # feat8 未用，保持与权重结构一致
|
||
|
||
|
||
class FeatureFusionModule(nn.Module):
|
||
def __init__(self, in_chan, out_chan):
|
||
super(FeatureFusionModule, self).__init__()
|
||
self.convblk = ConvBNReLU(in_chan, out_chan, ks=1, stride=1, padding=0)
|
||
self.conv1 = nn.Conv2d(out_chan, out_chan // 4, kernel_size=1, stride=1,
|
||
padding=0, bias=False)
|
||
self.conv2 = nn.Conv2d(out_chan // 4, out_chan, kernel_size=1, stride=1,
|
||
padding=0, bias=False)
|
||
self.relu = nn.ReLU(inplace=True)
|
||
self.sigmoid = nn.Sigmoid()
|
||
|
||
def forward(self, fsp, fcp):
|
||
fcat = torch.cat([fsp, fcp], dim=1)
|
||
feat = self.convblk(fcat)
|
||
atten = F.avg_pool2d(feat, feat.size()[2:])
|
||
atten = self.conv1(atten)
|
||
atten = self.relu(atten)
|
||
atten = self.conv2(atten)
|
||
atten = self.sigmoid(atten)
|
||
feat_atten = torch.mul(feat, atten)
|
||
feat_out = feat_atten + feat
|
||
return feat_out
|
||
|
||
|
||
class BiSeNet(nn.Module):
|
||
def __init__(self, n_classes):
|
||
super(BiSeNet, self).__init__()
|
||
# 该权重(79999_iter.pth)的变体无独立 SpatialPath：
|
||
# 直接用 ContextPath 的 resnet feat8（128ch）作为空间路径特征。
|
||
self.cp = ContextPath()
|
||
self.ffm = FeatureFusionModule(256, 256)
|
||
self.conv_out = BiSeNetOutput(256, 256, n_classes)
|
||
self.conv_out16 = BiSeNetOutput(128, 64, n_classes)
|
||
self.conv_out32 = BiSeNetOutput(128, 64, n_classes)
|
||
|
||
def forward(self, x):
|
||
h, w = x.size()[2:]
|
||
feat_res8, feat_cp8, feat_cp16 = self.cp(x)
|
||
feat_fuse = self.ffm(feat_res8, feat_cp8)
|
||
|
||
feat_out = self.conv_out(feat_fuse)
|
||
feat_out16 = self.conv_out16(feat_cp8)
|
||
feat_out32 = self.conv_out32(feat_cp16)
|
||
feat_out = F.interpolate(feat_out, (h, w), mode="bilinear", align_corners=True)
|
||
feat_out16 = F.interpolate(feat_out16, (h, w), mode="bilinear", align_corners=True)
|
||
feat_out32 = F.interpolate(feat_out32, (h, w), mode="bilinear", align_corners=True)
|
||
return feat_out, feat_out16, feat_out32
|