基于集对理论的视觉图片隐私计算模型与算法
隐私实时检测GUI设计实现
readme.md
# 作者：HuBingqi
# 时间：2021.7.1
# 说明
本程序主要实现的功能是对家居环境中人体实时语义分割以及
人体隐私检测。实现的方法主要是基于深度学习框架pytorch搭建
实时语义分割模型，并基于集对分析理论对人体隐私进行计算。本
人通过python语言完成对模型的训练测试，并完成对算法的集成，
最后通过python中扩展包tkinker实现对界面GUI的设计。
# 主要组成部分
1 人中所使用的语义分割模型是采用RefineNet-Light model.
该模型属于轻量化设计，具有可实时分割的效果，该网络的
backbone部分由resnet50组成，具有良好的特征提取能力，之后增
加网络基数上多尺度提取网络的特征，充分利用了网络的细节信息，
具备良好的细节分割能力。该网络将人体分割为18个部位（前景）。
其中分割的id与彩色图像素的对应关系在model_process.py中。若
有需要可自行更改。其分割的源代码大部分借鉴于前人github上：
https://github.com/DrSleep/light-weight-refinenet
2在完成了人体的语义分割之后，通过集对分析实现对隐私的量
化，其中该程序只针对人体中胸，大腿（左右），人脸，胳膊（左
右），手（左右）进行隐私度量：其中权重关系如下:
胸部 0.3
大腿 0.3
人脸 0.3
胳膊 0.05
手 0.05
在实际的测试中隐私的阈值为1.当大于1属于隐私，反之不属于隐
私，其中该阈值由集对分析理论中集合关系推导可得。此处不加详
述。对隐私的权重大小可集合实际进行调整，其中调节地点于
GUI.py中privacy_measure函数出进行修改。
# 效果
本文集合语义分割与隐私测量实时检测，一次在GUI设计时采用
了多线程技术进行处理，在修改源代码时多加注意。在使用时“依
次”点击开始、分割、测量的按钮，其中可能出现一下问题：
1、在显示区域未出现摄像头检测图像、首先应当检测是否插入
摄像头设备以及是否存在多个，程序未能检测到（更改GUI.py中
read_camera函数处读取摄像头索引号）
2、未能出现分割的图像，主要考虑两个方向会出问题：1、模
型权重+预训练权重（resnet50）是否存在于该文件中合适地点
2、拷贝代码模型时是否缺少文件或者相应的库。当你使用自己训练
的模型时，在保存模型时要考虑不同版本使用的兼容性。
3、当测量隐私时未能实现显示测试的结果图像，当认真检测
GUI.py中privacy_measure函数部分，是否是因为调节GUI尺寸导
致的显示问题。
ckpt
args.json
checkpoint.pth.tar
args.json
{
"ckpt_path": "./ckpt/checkpoint.pth.tar",
"enc": "50",
"enc_pretrained": true,
"evaluate": false,
"ignore_label": 255,
"num_stages": 3,
"num_workers": 6,
"optim_dec": "sgd",
"print_every": 10,
"random_seed": 42,
"snapshot_dir": "./ckpt/",
"train_dir": "../Data/train",
"val_dir": "../Data/val"
}
models
mobilenet.py
resnet.py
mobilenet.py
"""RefineNet-LightWeight
RefineNet-LigthWeight PyTorch for non-commercial purposes
Copyright (c) 2018, Vladimir Nekrasov (vladimir.nekrasov@adelaide.edu.au)
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
import torch.nn as nn
from utils.helpers import maybe_download
from utils.layer_factory import conv1x1, conv3x3, convbnrelu, CRPBlock
data_info = {21: "VOC"}
models_urls = {
"mbv2_voc": "https://cloudstor.aarnet.edu.au/plus/s/nQ6wDnTEFhyidot/download",
"mbv2_imagenet": "https://cloudstor.aarnet.edu.au/plus/s/uRgFbkaRjD3qOg5/download",
}
class InvertedResidualBlock(nn.Module):
"""Inverted Residual Block from https://arxiv.org/abs/1801.04381"""
def __init__(self, in_planes, out_planes, expansion_factor, stride=1):
super(InvertedResidualBlock, self).__init__()
intermed_planes = in_planes * expansion_factor
self.residual = (in_planes == out_planes) and (stride == 1)
self.output = nn.Sequential(
convbnrelu(in_planes, intermed_planes, 1),
convbnrelu(
intermed_planes,
intermed_planes,
3,
stride=stride,
groups=intermed_planes,
),
convbnrelu(intermed_planes, out_planes, 1, act=False),
)
def forward(self, x):
residual = x
out = self.output(x)
if self.residual:
return out + residual
else:
return out
class MBv2(nn.Module):
"""Net Definition"""
mobilenet_config = [
[1, 16, 1, 1], # expansion rate, output channels, number of repeats, stride
[6, 24, 2, 2],
[6, 32, 3, 2],
[6, 64, 4, 2],
[6, 96, 3, 1],
[6, 160, 3, 2],
[6, 320, 1, 1],
]
in_planes = 32 # number of input channels
num_layers = len(mobilenet_config)
def __init__(self, num_classes):
super(MBv2, self).__init__()
self.layer1 = convbnrelu(3, self.in_planes, kernel_size=3, stride=2)
c_layer = 2
for t, c, n, s in self.mobilenet_config:
layers = []
for idx in range(n):
layers.append(
InvertedResidualBlock(
self.in_planes,
c,
expansion_factor=t,
stride=s if idx == 0 else 1,
)
)
self.in_planes = c
setattr(self, "layer{}".format(c_layer), nn.Sequential(*layers))
c_layer += 1
## Light-Weight RefineNet ##
self.conv8 = conv1x1(320, 256, bias=False)
self.conv7 = conv1x1(160, 256, bias=False)
self.conv6 = conv1x1(96, 256, bias=False)
self.conv5 = conv1x1(64, 256, bias=False)
self.conv4 = conv1x1(32, 256, bias=False)
self.conv3 = conv1x1(24, 256, bias=False)
self.crp4 = self._make_crp(256, 256, 4)
self.crp3 = self._make_crp(256, 256, 4)
self.crp2 = self._make_crp(256, 256, 4)
self.crp1 = self._make_crp(256, 256, 4)
self.conv_adapt4 = conv1x1(256, 256, bias=False)
self.conv_adapt3 = conv1x1(256, 256, bias=False)
self.conv_adapt2 = conv1x1(256, 256, bias=False)
self.segm = conv3x3(256, num_classes, bias=True)
self.relu = nn.ReLU6(inplace=True)
def forward(self, x):
x = self.layer1(x)
x = self.layer2(x) # x / 2
l3 = self.layer3(x) # 24, x / 4
l4 = self.layer4(l3) # 32, x / 8
l5 = self.layer5(l4) # 64, x / 16
l6 = self.layer6(l5) # 96, x / 16
l7 = self.layer7(l6) # 160, x / 32
l8 = self.layer8(l7) # 320, x / 32
l8 = self.conv8(l8)
l7 = self.conv7(l7)
l7 = self.relu(l8 + l7)
l7 = self.crp4(l7)
l7 = self.conv_adapt4(l7)
l7 = nn.Upsample(size=l6.size()[2:], mode="bilinear", align_corners=True)(l7)
l6 = self.conv6(l6)
l5 = self.conv5(l5)
l5 = self.relu(l5 + l6 + l7)
l5 = self.crp3(l5)
l5 = self.conv_adapt3(l5)
l5 = nn.Upsample(size=l4.size()[2:], mode="bilinear", align_corners=True)(l5)
l4 = self.conv4(l4)
l4 = self.relu(l5 + l4)
l4 = self.crp2(l4)
l4 = self.conv_adapt2(l4)
l4 = nn.Upsample(size=l3.size()[2:], mode="bilinear", align_corners=True)(l4)
l3 = self.conv3(l3)
l3 = self.relu(l3 + l4)
l3 = self.crp1(l3)
out_segm = self.segm(l3)
return out_segm
def _initialize_weights(self):
for m in self.modules():
if isinstance(m, nn.Conv2d):
m.weight.data.normal_(0, 0.01)
if m.bias is not None:
m.bias.data.zero_()
elif isinstance(m, nn.BatchNorm2d):
m.weight.data.fill_(1)
m.bias.data.zero_()
def _make_crp(self, in_planes, out_planes, stages):
layers = [CRPBlock(in_planes, out_planes, stages)]
return nn.Sequential(*layers)
def mbv2(num_classes, imagenet=False, pretrained=True, **kwargs):
"""Constructs the network.
Args:
num_classes (int): the number of classes for the segmentation head to output.
"""
model = MBv2(num_classes, **kwargs)
if imagenet:
key = "mbv2_imagenet"
url = models_urls[key]
model.load_state_dict(maybe_download(key, url), strict=False)
elif pretrained:
dataset = data_info.get(num_classes, None)
if dataset:
bname = "mbv2_" + dataset.lower()
key = "rf_lw" + bname
url = models_urls[bname]
model.load_state_dict(maybe_download(key, url), strict=False)
return model
resnet.py
"""RefineNet-LightWeight
RefineNet-LigthWeight PyTorch for non-commercial purposes
Copyright (c) 2018, Vladimir Nekrasov (vladimir.nekrasov@adelaide.edu.au)
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
import torch.nn as nn
import torch.nn.functional as F
from utils.helpers import maybe_download
from utils.layer_factory import conv1x1, conv3x3, CRPBlock
data_info = {7: "Person", 21: "VOC", 40: "NYU", 60: "Context"}
models_urls = {
"50_person": "https://cloudstor.aarnet.edu.au/plus/s/mLA7NxVSPjNL7Oo/download",
"101_person": "https://cloudstor.aarnet.edu.au/plus/s/f1tGGpwdCnYS3xu/download",
"152_person": "https://cloudstor.aarnet.edu.au/plus/s/Ql64rWqiTvWGAA0/download",
"50_voc": "https://cloudstor.aarnet.edu.au/plus/s/xp7GcVKC0GbxhTv/download",
"101_voc": "https://cloudstor.aarnet.edu.au/plus/s/CPRKWiaCIDRdOwF/download",
"152_voc": "https://cloudstor.aarnet.edu.au/plus/s/2w8bFOd45JtPqbD/download",
"50_nyu": "https://cloudstor.aarnet.edu.au/plus/s/gE8dnQmHr9svpfu/download",
"101_nyu": "https://cloudstor.aarnet.edu.au/plus/s/VnsaSUHNZkuIqeB/download",
"152_nyu": "https://cloudstor.aarnet.edu.au/plus/s/EkPQzB2KtrrDnKf/download",
"101_context": "https://cloudstor.aarnet.edu.au/plus/s/hqmplxWOBbOYYjN/download",
"152_context": "https://cloudstor.aarnet.edu.au/plus/s/O84NszlYlsu00fW/download",
"50_imagenet": "https://download.pytorch.org/models/resnet50-19c8e357.pth",
"101_imagenet": "https://download.pytorch.org/models/resnet101-5d3b4d8f.pth",
"152_imagenet": "https://download.pytorch.org/models/resnet152-b121ed2d.pth",
}
stages_suffixes = {0: "_conv", 1: "_conv_relu_varout_dimred"}
class BasicBlock(nn.Module):
expansion = 1
def __init__(self, inplanes, planes, stride=1, downsample=None):
super(BasicBlock, self).__init__()
self.conv1 = conv3x3(inplanes, planes, stride)
self.bn1 = nn.BatchNorm2d(planes)
self.relu = nn.ReLU(inplace=True)
self.conv2 = conv3x3(planes, planes)
self.bn2 = nn.BatchNorm2d(planes)
self.downsample = downsample
self.stride = stride
def forward(self, x):
residual = x
out = self.conv1(x)
out = self.bn1(out)
out = self.relu(out)
out = self.conv2(out)
out = self.bn2(out)
if self.downsample is not None:
residual = self.downsample(x)
out += residual
out = self.relu(out)
return out
class Bottleneck(nn.Module):
expansion = 4
def __init__(self, inplanes, planes, stride=1, downsample=None):
super(Bottleneck, self).__init__()
self.conv1 = nn.Conv2d(inplanes, planes, kernel_size=1, bias=False)
self.bn1 = nn.BatchNorm2d(planes)
self.conv2 = nn.Conv2d(
planes, planes, kernel_size=3, stride=stride, padding=1, bias=False
)
self.bn2 = nn.BatchNorm2d(planes)
self.conv3 = nn.Conv2d(planes, planes * 4, kernel_size=1, bias=False)
self.bn3 = nn.BatchNorm2d(planes * 4)
self.relu = nn.ReLU(inplace=True)
self.downsample = downsample
self.stride = stride
def forward(self, x):
residual = x
out = self.conv1(x)
out = self.bn1(out)
out = self.relu(out)
out = self.conv2(out)
out = self.bn2(out)
out = self.relu(out)
out = self.conv3(out)
out = self.bn3(out)
if self.downsample is not None:
residual = self.downsample(x)
out += residual
out = self.relu(out)
return out
class ResNetLW(nn.Module):
def __init__(self, block, layers, num_classes=21):
self.inplanes = 64
super(ResNetLW, self).__init__()
self.do = nn.Dropout(p=0.5)
self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False)
self.bn1 = nn.BatchNorm2d(64)
self.relu = nn.ReLU(inplace=True)
self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
self.layer1 = self._make_layer(block, 64, layers[0])
self.layer2 = self._make_layer(block, 128, layers[1], stride=2)
self.layer3 = self._make_layer(block, 256, layers[2], stride=2)
self.layer4 = self._make_layer(block, 512, layers[3], stride=2)
self.p_ims1d2_outl1_dimred = conv1x1(2048, 512, bias=False)
self.mflow_conv_g1_pool = self._make_crp(512, 512, 4)
self.mflow_conv_g1_b3_joint_varout_dimred = conv1x1(512, 256, bias=False)
self.p_ims1d2_outl2_dimred = conv1x1(1024, 256, bias=False)
self.adapt_stage2_b2_joint_varout_dimred = conv1x1(256, 256, bias=False)
self.mflow_conv_g2_pool = self._make_crp(256, 256, 4)
self.mflow_conv_g2_b3_joint_varout_dimred = conv1x1(256, 256, bias=False)
self.p_ims1d2_outl3_dimred = conv1x1(512, 256, bias=False)
self.adapt_stage3_b2_joint_varout_dimred = conv1x1(256, 256, bias=False)
self.mflow_conv_g3_pool = self._make_crp(256, 256, 4)
self.mflow_conv_g3_b3_joint_varout_dimred = conv1x1(256, 256, bias=False)
self.p_ims1d2_outl4_dimred = conv1x1(256, 256, bias=False)
self.adapt_stage4_b2_joint_varout_dimred = conv1x1(256, 256, bias=False)
self.mflow_conv_g4_pool = self._make_crp(256, 256, 4)
self.clf_conv = nn.Conv2d(
256, num_classes, kernel_size=3, stride=1, padding=1, bias=True
)
def _make_crp(self, in_planes, out_planes, stages):
layers = [CRPBlock(in_planes, out_planes, stages)]
return nn.Sequential(*layers)
def _make_layer(self, block, planes, blocks, stride=1):
downsample = None
if stride != 1 or self.inplanes != planes * block.expansion:
downsample = nn.Sequential(
nn.Conv2d(
self.inplanes,
planes * block.expansion,
kernel_size=1,
stride=stride,
bias=False,
),
nn.BatchNorm2d(planes * block.expansion),
)
layers = []
layers.append(block(self.inplanes, planes, stride, downsample))
self.inplanes = planes * block.expansion
for i in range(1, blocks):
layers.append(block(self.inplanes, planes))
return nn.Sequential(*layers)
def forward(self, x):
x = self.conv1(x)
x = self.bn1(x)
x = self.relu(x)
x = self.maxpool(x)
l1 = self.layer1(x)
l2 = self.layer2(l1)
l3 = self.layer3(l2)
l4 = self.layer4(l3)
l4 = self.do(l4)
l3 = self.do(l3)
x4 = self.p_ims1d2_outl1_dimred(l4)
x4 = self.relu(x4)
x4 = self.mflow_conv_g1_pool(x4)
x4 = self.mflow_conv_g1_b3_joint_varout_dimred(x4)
x4 = nn.Upsample(size=l3.size()[2:], mode="bilinear", align_corners=True)(x4)
x3 = self.p_ims1d2_outl2_dimred(l3)
x3 = self.adapt_stage2_b2_joint_varout_dimred(x3)
x3 = x3 + x4
x3 = F.relu(x3)
x3 = self.mflow_conv_g2_pool(x3)
x3 = self.mflow_conv_g2_b3_joint_varout_dimred(x3)
x3 = nn.Upsample(size=l2.size()[2:], mode="bilinear", align_corners=True)(x3)
x2 = self.p_ims1d2_outl3_dimred(l2)
x2 = self.adapt_stage3_b2_joint_varout_dimred(x2)
x2 = x2 + x3
x2 = F.relu(x2)
x2 = self.mflow_conv_g3_pool(x2)
x2 = self.mflow_conv_g3_b3_joint_varout_dimred(x2)
x2 = nn.Upsample(size=l1.size()[2:], mode="bilinear", align_corners=True)(x2)
x1 = self.p_ims1d2_outl4_dimred(l1)
x1 = self.adapt_stage4_b2_joint_varout_dimred(x1)
x1 = x1 + x2
x1 = F.relu(x1)
x1 = self.mflow_conv_g4_pool(x1)
out = self.clf_conv(x1)
return out
def rf_lw50(num_classes, imagenet=False, pretrained=True, **kwargs):
model = ResNetLW(Bottleneck, [3, 4, 6, 3], num_classes=num_classes, **kwargs)
if imagenet:
key = "50_imagenet"
url = models_urls[key]
model.load_state_dict(maybe_download(key, url), strict=False)
elif pretrained:
assert 3==4
dataset = data_info.get(num_classes, None)
if dataset:
bname = "50_" + dataset.lower()
key = "rf_lw" + bname
url = models_urls[bname]
model.load_state_dict(maybe_download(key, url), strict=False)
return model
def rf_lw101(num_classes, imagenet=False, pretrained=True, **kwargs):
model = ResNetLW(Bottleneck, [3, 4, 23, 3], num_classes=num_classes, **kwargs)
if imagenet:
key = "101_imagenet"
url = models_urls[key]
model.load_state_dict(maybe_download(key, url), strict=False)
elif pretrained:
assert 1==2
dataset = data_info.get(num_classes, None)
if dataset:
bname = "101_" + dataset.lower()
key = "rf_lw" + bname
url = models_urls[bname]
model.load_state_dict(maybe_download(key, url), strict=False)
return model
def rf_lw152(num_classes, imagenet=False, pretrained=True, **kwargs):
model = ResNetLW(Bottleneck, [3, 8, 36, 3], num_classes=num_classes, **kwargs)
if imagenet:
key = "152_imagenet"
url = models_urls[key]
model.load_state_dict(maybe_download(key, url), strict=False)
elif pretrained:
assert 5==6
dataset = data_info.get(num_classes, None)
if dataset:
bname = "152_" + dataset.lower()
key = "rf_lw" + bname
url = models_urls[bname]
model.load_state_dict(maybe_download(key, url), strict=False)
return model
utils
helpers.py
layer_factory.py
cmap.npy
cs_cmap.npy
helpers.py
import numpy as np
import torch
IMG_SCALE = 1.0 / 255
IMG_MEAN = np.array([0.485, 0.456, 0.406]).reshape((1, 1, 3))
IMG_STD = np.array([0.229, 0.224, 0.225]).reshape((1, 1, 3))
def maybe_download(model_name, model_url, model_dir=None, map_location=None):
import os
import sys
from six.moves import urllib
if model_dir is None:
torch_home = os.path.expanduser(os.getenv("TORCH_HOME", "~/.torch"))
model_dir = os.getenv("TORCH_MODEL_ZOO", os.path.join(torch_home, "models"))
if not os.path.exists(model_dir):
os.makedirs(model_dir)
filename = "{}.pth.tar".format(model_name)
cached_file = os.path.join(model_dir, filename)
if not os.path.exists(cached_file):
url = model_url
sys.stderr.write('Downloading: "{}" to {}\n'.format(url, cached_file))
urllib.request.urlretrieve(url, cached_file)
return torch.load(cached_file, map_location=map_location)
def prepare_img(img):
return (img * IMG_SCALE - IMG_MEAN) / IMG_STD
layer_factory.py
"""RefineNet-LightWeight-CRP Block
RefineNet-LigthWeight PyTorch for non-commercial purposes
Copyright (c) 2018, Vladimir Nekrasov (vladimir.nekrasov@adelaide.edu.au)
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
import torch.nn as nn
def batchnorm(in_planes):
"batch norm 2d"
return nn.BatchNorm2d(in_planes, affine=True, eps=1e-5, momentum=0.1)
def conv3x3(in_planes, out_planes, stride=1, bias=False):
"3x3 convolution with padding"
return nn.Conv2d(
in_planes, out_planes, kernel_size=3, stride=stride, padding=1, bias=bias
)
def conv1x1(in_planes, out_planes, stride=1, bias=False):
"1x1 convolution"
return nn.Conv2d(
in_planes, out_planes, kernel_size=1, stride=stride, padding=0, bias=bias
)
def convbnrelu(in_planes, out_planes, kernel_size, stride=1, groups=1, act=True):
"conv-batchnorm-relu"
if act:
return nn.Sequential(
nn.Conv2d(
in_planes,
out_planes,
kernel_size,
stride=stride,
padding=int(kernel_size / 2.0),
groups=groups,
bias=False,
),
batchnorm(out_planes),
nn.ReLU6(inplace=True),
)
else:
return nn.Sequential(
nn.Conv2d(
in_planes,
out_planes,
kernel_size,
stride=stride,
padding=int(kernel_size / 2.0),
groups=groups,
bias=False,
),
batchnorm(out_planes),
)
class CRPBlock(nn.Module):
def __init__(self, in_planes, out_planes, n_stages):
super(CRPBlock, self).__init__()
for i in range(n_stages):
setattr(
self,
"{}_{}".format(i + 1, "outvar_dimred"),
conv1x1(
in_planes if (i == 0) else out_planes,
out_planes,
stride=1,
bias=False,
),
)
self.stride = 1
self.n_stages = n_stages
self.maxpool = nn.MaxPool2d(kernel_size=5, stride=1, padding=2)
def forward(self, x):
top = x
for i in range(self.n_stages):
top = self.maxpool(top)
top = getattr(self, "{}_{}".format(i + 1, "outvar_dimred"))(top)
x = top + x
return x
config.py
import numpy as np
# DATASET PARAMETERS
TRAIN_DIR = "../Data/train"
VAL_DIR = "../Data/val"
TRAIN_LIST = ["../Data/train/train_list.txt"] * 3
VAL_LIST = ["../Data/val/val_list.txt"] * 3
SHORTER_SIDE = [128] * 3
CROP_SIZE = [256] * 3
NORMALISE_PARAMS = [
1.0 / 255, # SCALE
np.array([0.485, 0.456, 0.406]).reshape((1, 1, 3)), # MEAN
np.array([0.229, 0.224, 0.225]).reshape((1, 1, 3)),
] # STD
BATCH_SIZE = [64] * 3
NUM_WORKERS = 6
NUM_CLASSES = [18] * 3
LOW_SCALE = [0.5] * 3
HIGH_SCALE = [2.0] * 3
IGNORE_LABEL = 255
# ENCODER PARAMETERS
ENC = "50"
ENC_PRETRAINED = True # pre-trained on ImageNet or randomly initialised
# GENERAL
EVALUATE = False
FREEZE_BN = [True] * 3
NUM_SEGM_EPOCHS = [100] * 3
PRINT_EVERY = 10
RANDOM_SEED = 42
SNAPSHOT_DIR = "./ckpt/"
CKPT_PATH = "./ckpt/checkpoint.pth.tar"
VAL_EVERY = [5] * 3 # how often to record validation scores
# OPTIMISERS' PARAMETERS
LR_ENC = [5e-4, 2.5e-4, 1e-4] # TO FREEZE, PUT 0
LR_DEC = [5e-3, 2.5e-3, 1e-3]
MOM_ENC = [0.9] * 3 # TO FREEZE, PUT 0
MOM_DEC = [0.9] * 3
WD_ENC = [1e-5] * 3 # TO FREEZE, PUT 0
WD_DEC = [1e-5] * 3
OPTIM_DEC = "sgd"
datasets.py
"""RefineNet-LightWeight
RefineNet-LigthWeight PyTorch for non-commercial purposes
Copyright (c) 2018, Vladimir Nekrasov (vladimir.nekrasov@adelaide.edu.au)
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
from __future__ import print_function, division
import os
import warnings
import cv2
import numpy as np
import torch
from PIL import Image
from torch.utils.data import Dataset
warnings.filterwarnings("ignore")
class Pad(object):
"""Pad image and mask to the desired size
Args:
size (int) : minimum length/width
img_val (array) : image padding value
msk_val (int) : mask padding value
"""
def __init__(self, size, img_val, msk_val):
self.size = size
self.img_val = img_val
self.msk_val = msk_val
def __call__(self, sample):
image, mask = sample["image"], sample["mask"]
h, w = image.shape[:2]
h_pad = int(np.clip(((self.size - h) + 1) // 2, 0, 1e6))
w_pad = int(np.clip(((self.size - w) + 1) // 2, 0, 1e6))
pad = ((h_pad, h_pad), (w_pad, w_pad))
image = np.stack(
[
np.pad(
image[:, :, c],
pad,
mode="constant",
constant_values=self.img_val[c],
)
for c in range(3)
],
axis=2,
)
mask = np.pad(mask, pad, mode="constant", constant_values=self.msk_val)
return {"image": image, "mask": mask}
class RandomCrop(object):
"""Crop randomly the image in a sample.
Args:
output_size (tuple or int): Desired output size. If int, square crop
is made.
"""
def __init__(self, crop_size):
assert isinstance(crop_size, int)
self.crop_size = crop_size
if self.crop_size % 2 != 0:
self.crop_size -= 1
def __call__(self, sample):
image, mask = sample["image"], sample["mask"]
h, w = image.shape[:2]
new_h = min(h, self.crop_size)
new_w = min(w, self.crop_size)
top = np.random.randint(0, h - new_h + 1)
left = np.random.randint(0, w - new_w + 1)
image = image[top : top + new_h, left : left + new_w]
mask = mask[top : top + new_h, left : left + new_w]
return {"image": image, "mask": mask}
class ResizeShorterScale(object):
"""Resize shorter side to a given value and randomly scale."""
def __init__(self, shorter_side, low_scale, high_scale):
assert isinstance(shorter_side, int)
self.shorter_side = shorter_side
self.low_scale = low_scale
self.high_scale = high_scale
def __call__(self, sample):
image, mask = sample["image"], sample["mask"]
min_side = min(image.shape[:2])
scale = np.random.uniform(self.low_scale, self.high_scale)
if min_side * scale < self.shorter_side:
scale = self.shorter_side * 1.0 / min_side
image = cv2.resize(
image, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC
)
mask = cv2.resize(
mask, None, fx=scale, fy=scale, interpolation=cv2.INTER_NEAREST
)
return {"image": image, "mask": mask}
class RandomMirror(object):
"""Randomly flip the image and the mask"""
def __init__(self):
pass
def __call__(self, sample):
image, mask = sample["image"], sample["mask"]
do_mirror = np.random.randint(2)
if do_mirror:
image = cv2.flip(image, 1)
mask = cv2.flip(mask, 1)
return {"image": image, "mask": mask}
class Normalise(object):
"""Normalise a tensor image with mean and standard deviation.
Given mean: (R, G, B) and std: (R, G, B),
will normalise each channel of the torch.*Tensor, i.e.
channel = (channel - mean) / std
Args:
mean (sequence): Sequence of means for R, G, B channels respecitvely.
std (sequence): Sequence of standard deviations for R, G, B channels
respecitvely.
"""
def __init__(self, scale, mean, std):
self.scale = scale
self.mean = mean
self.std = std
def __call__(self, sample):
image = sample["image"]
return {
"image": (self.scale * image - self.mean) / self.std,
"mask": sample["mask"],
}
class ToTensor(object):
"""Convert ndarrays in sample to Tensors."""
def __call__(self, sample):
image, mask = sample["image"], sample["mask"]
# swap color axis because
# numpy image: H x W x C
# torch image: C X H X W
image = image.transpose((2, 0, 1))
return {"image": torch.from_numpy(image), "mask": torch.from_numpy(mask)}
class NYUDataset(Dataset):
"""NYUv2-40"""
def __init__(self, data_file, data_dir, transform_trn=None, transform_val=None):
"""
Args:
data_file (string): Path to the data file with annotations.
data_dir (string): Directory with all the images.
transform_{trn, val} (callable, optional): Optional transform to be applied
on a sample.
"""
with open(data_file, "rb") as f:
datalist = f.readlines()
self.datalist = [
(k, v)
for k, v in map(
lambda x: x.decode("utf-8").strip("\n").split("\t"), datalist
)
]
self.root_dir = data_dir
self.transform_trn = transform_trn
self.transform_val = transform_val
self.stage = "train"
def set_stage(self, stage):
self.stage = stage
def __len__(self):
return len(self.datalist)
def __getitem__(self, idx):
# img_name = os.path.join(self.root_dir, self.datalist[idx][0])
# msk_name = os.path.join(self.root_dir, self.datalist[idx][1])
img_name = self.root_dir + self.datalist[idx][0]
msk_name = self.root_dir + self.datalist[idx][1]
def read_image(x):
img_arr = np.array(Image.open(x))
if len(img_arr.shape) == 2: # grayscale
img_arr = np.tile(img_arr, [3, 1, 1]).transpose(1, 2, 0)
return img_arr
image = read_image(img_name)
mask = np.array(Image.open(msk_name))
try:
mask = np.array(Image.open(msk_name))
except:
mask = torch.randn(256,256,dtype = int)
if img_name != msk_name:
assert len(mask.shape) == 2, "Masks must be encoded without colourmap"
if self.stage == "val":
if self.transform_val:
image = self.transform_val(image)
try:
mask = self.transform_val(mask)
except:
mask = mask
sample = {"image": image, "mask": mask}
if self.stage == "train":
if self.transform_trn:
sample = self.transform_trn(sample)
'''
elif self.stage == "val":
if self.transform_val:
sample = self.transform_val(sample)
'''
return sample
ghjkl.py
import numpy as np
import matplotlib.pyplot as plt
import mpl_toolkits.axisartist as axisartist
from matplotlib.backends.backend_agg import FigureCanvasAgg
import cv2
plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False # 用来正常显示负号
from matplotlib.pyplot import MultipleLocator
# 创建画布
fig = plt.figure(figsize=(6, 6))
# 创建一个绘图区对象
ax = axisartist.Subplot(fig, 111)
# 将对象添加到画布中
fig.add_axes(ax)
# 设置绘图区所有坐标轴隐藏
ax.axis[:].set_visible(False)
# 添加新的坐标轴
ax.axis["x"] = ax.new_floating_axis(0, 0) # 第一个0代表平行直线，第二个0代表经过0点
# 添加箭头
ax.axis["x"].set_axisline_style("-|>", size=1.0) # ->表示空箭头
ax.axis["y"] = ax.new_floating_axis(1, 0) # 竖直直线经过0点
ax.axis["y"].set_axisline_style("-|>", size=1.0) # -|> 表示实心箭头
# 设置x,y轴上刻度显示方向
ax.axis["x"].set_visible(True) # 设置是否可见
ax.axis["x"].set_axis_direction("bottom") # 刻度在上面还是下面
ax.axis["y"].set_axis_direction("left") # 刻度在左边还是右边
# 设置坐标轴的名称及格式
fontstyle = {'family': 'Times New Roman',
'weight': 'normal',
'size': 30, }
ax.axis["x"].label.set_text("times")
# ax.axis["x"].label.set_color("black") # 设置颜色
ax.axis["x"].label.set_fontsize(10) # 设置字体大小
# ax.axis["x"].label.set_rotation(45) # 设置旋转角度
ax.axis["y"].label.set_text("privacy")
ax.axis["y"].label.set_fontsize(10) # 设置字体大小
# 设置坐标轴刻度范围
list = list(range(1, 50))
length = len(list)
for i in range(length):
list[i] = list[i]/length*5
plt.xticks(np.arange(0, int(length*2), 10))
plt.yticks(np.arange(-1, max(), 0.5))
# 设置底边距
plt.subplots_adjust(bottom=0.10)
# 开启网格
# ax.grid(True, linestyle='-.') # 开启网格并设置样式
x = range(length)
plt.plot(x,list)
plt.xlim(0,int(length*2))
canvas = FigureCanvasAgg(plt.gcf())
canvas.draw()
img = np.array(canvas.renderer.buffer_rgba())
print(img.shape)
cv2.imshow('1',img)
cv2.waitKey(0)
# plt.show()
miou_utils.py
import numpy as np
import os
def fast_cm(preds, gt, numclass):
cm = np.zeros((numclass,numclass),dtype=int)
n = gt.shape[0]
for i in range(n):
a = gt[i]
p = preds[i]
cm[a, p] += 1
return cm
def compute_iu(confuse):
numclass = confuse.shape[0]
iu = np.ones(numclass)
for i in range(numclass):
pi = sum(confuse[:, i])
gi = sum(confuse[i, :])
ii = confuse[i, i]
denom = pi + gi - ii
if denom > 0:
iu[i] = ii / denom
return iu
model_process.py
# general libs
import argparse
import logging
import os
import random
import re
import time
# misc
import cv2
import numpy as np
# pytorch libs
import torch
import torch.nn as nn
# custom libs
import sys
sys.path.append('./privacy_measure/')
from config import *
from PIL import Image
from miou_utils import compute_iu, fast_cm
from util import *
from datasets import ToTensor,Normalise
from torchvision import transforms
frame = None
color_pic = None
def create_segmenter(net, pretrained, num_classes):
"""Create Encoder; for now only ResNet [50,101,152]"""
from models.resnet import rf_lw50, rf_lw101, rf_lw152
if str(net) == "50":
return rf_lw50(num_classes, imagenet=pretrained)
elif str(net) == "101":
return rf_lw101(num_classes, imagenet=pretrained)
elif str(net) == "152":
return rf_lw152(num_classes, imagenet=pretrained)
else:
raise ValueError("{} is not supported".format(str(net)))
def load_ckpt(ckpt_path, ckpt_dict):
best_val = epoch_start = 0
if os.path.exists(ckpt_path):
ckpt = torch.load(ckpt_path)
for (k, v) in ckpt_dict.items():
if k in ckpt:
v.load_state_dict(ckpt[k])
best_val = ckpt.get("best_val", 0)
epoch_start = ckpt.get("epoch_start", 0)
return best_val, epoch_start
def transform_color(id):
label = [
[0,0,0],
[255,255,255],
[192,192,192],
[128,128,128],
[255,0,0],
[128,0,0],
[255,255,0],
[128,128,0],
[0,255,0],
[0,128,0],
[0,255,255],
[0,128,128],
[0,0,255],
[0,0,128],
[255,0,255],
[128,0,128],
[0,128,255],
[128,0,255]
]
color = np.array(label)
color_pic = color[id.astype(int)]
color_pic = color_pic.astype(np.uint8)
return color_pic
def build_model():
enc = "50"
enc_pretrained = True
segmenter = nn.DataParallel(
create_segmenter(enc, enc_pretrained, 18)
).cuda()
ckpt_path = "./ckpt/checkpoint.pth.tar"
load_ckpt(ckpt_path, {"segmenter": segmenter})
enc_params = []
dec_params = []
for k, v in segmenter.named_parameters():
if bool(re.match(".*conv1.*|.*bn1.*|.*layer.*", k)):
enc_params.append(v)
else:
dec_params.append(v)
return segmenter
def proccess(image):
pred_transforms = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
Image_pred = np.array(image/255.0)
pred = pred_transforms(Image_pred).unsqueeze(0)
return pred
def read_camera():
ret, frame = cap.read()
if ret:
input = proccess(frame)
with torch.no_grad():
input_var = torch.autograd.Variable(input).float().cuda()
output = segmenter(input_var)
output = (
cv2.resize(
output[0, :num_classes].data.cpu().numpy().transpose(1, 2, 0),
input.shape[2:4][::-1],
interpolation=cv2.INTER_CUBIC,
)
.argmax(axis=2)
.astype(np.uint8)
)
color_pic = transform_color(output)
current_image = Image.fromarray(frame) # 将图像转换成Image对象
imgtk = ImageTk.PhotoImage(image=current_image)
pannel1.imgtk = imgtk
pannel1.config(image=imgtk)
segment = Image.fromarray(color_pic) # 将图像转换成Image对象
segment = ImageTk.PhotoImage(image=segment)
pannel2.imgtk = segment
pannel2.config(image=segment)
pannel3.imgtk = segment
pannel3.config(image=segment)
pannel4.imgtk = imgtk
pannel4.config(image=imgtk)
window.after(1,read_camera)
if __name__ == "__main__":
# index = 0
# read_camera(index)
from tkinter import *
import cv2
from PIL import Image, ImageTk
import tkinter
# 加载并初始化模型
segmenter = build_model()
segmenter.eval()
#打开摄像头
cap = cv2.VideoCapture(0)
#分割的类别数
num_classes = 18
# 建立可视化界面并命名
window = tkinter.Tk()
window.title("隐私度量和保护系统")
# 获取显示屏大小
width = window.winfo_screenwidth()
height = window.winfo_screenheight()
# 控件的窗口大小
window.geometry("%dx%d"%(width,height))
# 添加 启动 、 分割 、 测量 、 监护 按钮功能
# Button_height = 5
#
# start = Button(window,text = "开始" ,commond = "",height= Button_height)
# start.pack()
#
# segment = Button(window,text = "分割" , commond = "" ,heght = Button_height)
# segment.pack()
#
# measure = Button(window,text = "测量" ,commond = "",height = Button_height)
# measure.pack()
#
# custody = Button(window ,text = "监护" ,command = "" , height = Button_height)
# custody.pack()
# 设置读取图像的大小
m = 2 # 横向视频数目
n = 2 # 纵向视频数目
cap.set(3, 200) # 设置图像分辨率
cap.set(4, 200) # 设置图像分辨率
pannel1 = Label(window)
pannel1.grid(row=0,column=0)
pannel2 = Label(window)
pannel2.grid(row=0,column=1)
pannel3 = Label(window)
pannel3.grid(row=1,column=0)
pannel4 = Label(window)
pannel4.grid(row=1,column=1)
read_camera()
window.mainloop()
cap.release()
cv2.destroyAllWindows()
setup.py
from distutils.core import setup
from Cython.Build import cythonize
import numpy
setup(
ext_modules=cythonize("./src/*.pyx"), include_dirs=[numpy.get_include()],
)
test.py
# general libs
import argparse
import logging
import os
import random
import re
import time
# misc
import cv2
import numpy as np
# pytorch libs
import torch
import torch.nn as nn
# custom libs
from config import *
from miou_utils import compute_iu, fast_cm
from util import *
def get_arguments():
"""Parse all the arguments provided from the CLI.
Returns:
A list of parsed arguments.
"""
parser = argparse.ArgumentParser(description="Full Pipeline Training")
# Dataset
parser.add_argument(
"--train-dir",
type=str,
default=TRAIN_DIR,
help="Path to the training set directory.",
)
parser.add_argument(
"--val-dir",
type=str,
default=VAL_DIR,
help="Path to the validation set directory.",
)
parser.add_argument(
"--train-list",
type=str,
nargs="+",
default=TRAIN_LIST,
help="Path to the training set list.",
)
parser.add_argument(
"--val-list",
type=str,
nargs="+",
default=VAL_LIST,
help="Path to the validation set list.",
)
parser.add_argument(
"--shorter-side",
type=int,
nargs="+",
default=SHORTER_SIDE,
help="Shorter side transformation.",
)
parser.add_argument(
"--crop-size",
type=int,
nargs="+",
default=CROP_SIZE,
help="Crop size for training,",
)
parser.add_argument(
"--normalise-params",
type=list,
default=NORMALISE_PARAMS,
help="Normalisation parameters [scale, mean, std],",
)
parser.add_argument(
"--batch-size",
type=int,
nargs="+",
default=BATCH_SIZE,
help="Batch size to train the segmenter model.",
)
parser.add_argument(
"--num-workers",
type=int,
default=NUM_WORKERS,
help="Number of workers for pytorch's dataloader.",
)
parser.add_argument(
"--num-classes",
type=int,
nargs="+",
default=NUM_CLASSES,
help="Number of output classes for each task.",
)
parser.add_argument(
"--low-scale",
type=float,
nargs="+",
default=LOW_SCALE,
help="Lower bound for random scale",
)
parser.add_argument(
"--high-scale",
type=float,
nargs="+",
default=HIGH_SCALE,
help="Upper bound for random scale",
)
parser.add_argument(
"--ignore-label",
type=int,
default=IGNORE_LABEL,
help="Label to ignore during training",
)
# Encoder
parser.add_argument("--enc", type=str, default=ENC, help="Encoder net type.")
parser.add_argument(
"--enc-pretrained",
type=bool,
default=ENC_PRETRAINED,
help="Whether to init with imagenet weights.",
)
# General
parser.add_argument(
"--evaluate",
type=bool,
default=EVALUATE,
help="If true, only validate segmentation.",
)
parser.add_argument(
"--freeze-bn",
type=bool,
nargs="+",
default=FREEZE_BN,
help="Whether to keep batch norm statistics intact.",
)
parser.add_argument(
"--num-segm-epochs",
type=int,
nargs="+",
default=NUM_SEGM_EPOCHS,
help="Number of epochs to train for segmentation network.",
)
parser.add_argument(
"--print-every",
type=int,
default=PRINT_EVERY,
help="Print information every often.",
)
parser.add_argument(
"--random-seed",
type=int,
default=RANDOM_SEED,
help="Seed to provide (near-)reproducibility.",
)
parser.add_argument(
"--snapshot-dir",
type=str,
default=SNAPSHOT_DIR,
help="Path to directory for storing checkpoints.",
)
parser.add_argument(
"--ckpt-path", type=str, default=CKPT_PATH, help="Path to the checkpoint file."
)
parser.add_argument(
"--val-every",
nargs="+",
type=int,
default=VAL_EVERY,
help="How often to validate current architecture.",
)
# Optimisers
parser.add_argument(
"--lr-enc",
type=float,
nargs="+",
default=LR_ENC,
help="Learning rate for encoder.",
)
parser.add_argument(
"--lr-dec",
type=float,
nargs="+",
default=LR_DEC,
help="Learning rate for decoder.",
)
parser.add_argument(
"--mom-enc",
type=float,
nargs="+",
default=MOM_ENC,
help="Momentum for encoder.",
)
parser.add_argument(
"--mom-dec",
type=float,
nargs="+",
default=MOM_DEC,
help="Momentum for decoder.",
)
parser.add_argument(
"--wd-enc",
type=float,
nargs="+",
default=WD_ENC,
help="Weight decay for encoder.",
)
parser.add_argument(
"--wd-dec",
type=float,
nargs="+",
default=WD_DEC,
help="Weight decay for decoder.",
)
parser.add_argument(
"--optim-dec",
type=str,
default=OPTIM_DEC,
help="Optimiser algorithm for decoder.",
)
return parser.parse_args()
def create_segmenter(net, pretrained, num_classes):
"""Create Encoder; for now only ResNet [50,101,152]"""
from models.resnet import rf_lw50, rf_lw101, rf_lw152
if str(net) == "50":
return rf_lw50(num_classes, imagenet=pretrained)
elif str(net) == "101":
return rf_lw101(num_classes, imagenet=pretrained)
elif str(net) == "152":
return rf_lw152(num_classes, imagenet=pretrained)
else:
raise ValueError("{} is not supported".format(str(net)))
def create_loaders(
test_dir,
test_list,
shorter_side,
crop_size,
low_scale,
high_scale,
normalise_params,
batch_size,
num_workers,
ignore_label,
):
# Torch libraries
from torchvision import transforms
from torch.utils.data import DataLoader
# Custom libraries
from datasets import NYUDataset as Dataset
from datasets import (
Pad,
RandomCrop,
RandomMirror,
ResizeShorterScale,
ToTensor,
Normalise,
)
composed_val = transforms.Compose([Normalise(*normalise_params), ToTensor()])
testset = Dataset(
data_file=test_list,
data_dir=test_dir,
transform_trn=None,
transform_val=composed_val,
)
test_loader = DataLoader(
testset, batch_size=1, shuffle=False, num_workers=0, pin_memory=True
)
return test_loader
def load_ckpt(ckpt_path, ckpt_dict):
best_val = epoch_start = 0
if os.path.exists(args.ckpt_path):
ckpt = torch.load(ckpt_path)
for (k, v) in ckpt_dict.items():
if k in ckpt:
v.load_state_dict(ckpt[k])
best_val = ckpt.get("best_val", 0)
epoch_start = ckpt.get("epoch_start", 0)
logger.info(
" Found checkpoint at {} with best_val {:.4f} at epoch {}".format(
ckpt_path, best_val, epoch_start
)
)
return best_val, epoch_start
def transform_color(id):
label = [
[0,0,0],
[255,255,255],
[192,192,192],
[128,128,128],
[255,0,0],
[128,0,0],
[255,255,0],
[128,128,0],
[0,255,0],
[0,128,0],
[0,255,255],
[0,128,128],
[0,0,255],
[0,0,128],
[255,0,255],
[128,0,128],
[0,128,255],
[128,0,255]
]
color = np.array(label)
color_pic = color[id.astype(int)]
color_pic = color_pic.astype(np.uint8)
return color_pic
def test(segmenter, test_loader, num_classes=-1):
test_loader.dataset.set_stage("val")
segmenter.eval()
with torch.no_grad():
for i, sample in enumerate(test_loader):
input = sample["image"]
input_var = torch.autograd.Variable(input).float().cuda()
# Compute output
output = segmenter(input_var)
output = (
cv2.resize(
output[0, :num_classes].data.cpu().numpy().transpose(1, 2, 0),
input.shape[2:4][::-1],
interpolation=cv2.INTER_CUBIC,
)
.argmax(axis=2)
.astype(np.uint8)
)
# Compute IoU
color_pic = transform_color(output)
name = str(i) + ".png"
save_path = "./pred/color/" + name
cv2.imwrite(save_path,color_pic)
def main():
global args, logger
args = get_arguments()
logger = logging.getLogger(__name__)
## Add args ##
args.num_stages = len(args.num_classes)
## Set random seeds ##
torch.backends.cudnn.deterministic = True
torch.manual_seed(args.random_seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(args.random_seed)
np.random.seed(args.random_seed)
random.seed(args.random_seed)
## Generate Segmenter ##
segmenter = nn.DataParallel(
create_segmenter(args.enc, args.enc_pretrained, args.num_classes[0])
).cuda()
## Restore if any ##
best_val, epoch_start = load_ckpt(args.ckpt_path, {"segmenter": segmenter})
#########################
test_dir = "./pred"
test_list = "./pred/test_list.txt"
########################
start = time.time()
torch.cuda.empty_cache()
## Create dataloaders ##
test_loader = create_loaders(
test_dir,
test_list,
args.shorter_side[0],
args.crop_size[0],
args.low_scale[0],
args.high_scale[0],
args.normalise_params,
args.batch_size[0],
args.num_workers,
args.ignore_label,
)
enc_params = []
dec_params = []
for k, v in segmenter.named_parameters():
if bool(re.match(".*conv1.*|.*bn1.*|.*layer.*", k)):
enc_params.append(v)
logger.info(" Enc. parameter: {}".format(k))
else:
dec_params.append(v)
logger.info(" Dec. parameter: {}".format(k))
test(
segmenter,
test_loader,
num_classes = 18
)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
main()
train.py
# general libs
import argparse
import logging
import os
import random
import re
import time
# misc
import cv2
import numpy as np
# pytorch libs
import torch
import torch.nn as nn
# custom libs
from config import *
from miou_utils import compute_iu, fast_cm
from util import *
import os
os.environ['CUDA_VISIBLE_DEVICES']= "0,1"
def get_arguments():
"""Parse all the arguments provided from the CLI.
Returns:
A list of parsed arguments.
"""
parser = argparse.ArgumentParser(description="Full Pipeline Training")
# Dataset
parser.add_argument(
"--train-dir",
type=str,
default=TRAIN_DIR,
help="Path to the training set directory.",
)
parser.add_argument(
"--val-dir",
type=str,
default=VAL_DIR,
help="Path to the validation set directory.",
)
parser.add_argument(
"--train-list",
type=str,
nargs="+",
default=TRAIN_LIST,
help="Path to the training set list.",
)
parser.add_argument(
"--val-list",
type=str,
nargs="+",
default=VAL_LIST,
help="Path to the validation set list.",
)
parser.add_argument(
"--shorter-side",
type=int,
nargs="+",
default=SHORTER_SIDE,
help="Shorter side transformation.",
)
parser.add_argument(
"--crop-size",
type=int,
nargs="+",
default=CROP_SIZE,
help="Crop size for training,",
)
parser.add_argument(
"--normalise-params",
type=list,
default=NORMALISE_PARAMS,
help="Normalisation parameters [scale, mean, std],",
)
parser.add_argument(
"--batch-size",
type=int,
nargs="+",
default=BATCH_SIZE,
help="Batch size to train the segmenter model.",
)
parser.add_argument(
"--num-workers",
type=int,
default=NUM_WORKERS,
help="Number of workers for pytorch's dataloader.",
)
parser.add_argument(
"--num-classes",
type=int,
nargs="+",
default=NUM_CLASSES,
help="Number of output classes for each task.",
)
parser.add_argument(
"--low-scale",
type=float,
nargs="+",
default=LOW_SCALE,
help="Lower bound for random scale",
)
parser.add_argument(
"--high-scale",
type=float,
nargs="+",
default=HIGH_SCALE,
help="Upper bound for random scale",
)
parser.add_argument(
"--ignore-label",
type=int,
default=IGNORE_LABEL,
help="Label to ignore during training",
)
# Encoder
parser.add_argument("--enc", type=str, default=ENC, help="Encoder net type.")
parser.add_argument(
"--enc-pretrained",
type=bool,
default=ENC_PRETRAINED,
help="Whether to init with imagenet weights.",
)
# General
parser.add_argument(
"--evaluate",
type=bool,
default=EVALUATE,
help="If true, only validate segmentation.",
)
parser.add_argument(
"--freeze-bn",
type=bool,
nargs="+",
default=FREEZE_BN,
help="Whether to keep batch norm statistics intact.",
)
parser.add_argument(
"--num-segm-epochs",
type=int,
nargs="+",
default=NUM_SEGM_EPOCHS,
help="Number of epochs to train for segmentation network.",
)
parser.add_argument(
"--print-every",
type=int,
default=PRINT_EVERY,
help="Print information every often.",
)
parser.add_argument(
"--random-seed",
type=int,
default=RANDOM_SEED,
help="Seed to provide (near-)reproducibility.",
)
parser.add_argument(
"--snapshot-dir",
type=str,
default=SNAPSHOT_DIR,
help="Path to directory for storing checkpoints.",
)
parser.add_argument(
"--ckpt-path", type=str, default=CKPT_PATH, help="Path to the checkpoint file."
)
parser.add_argument(
"--val-every",
nargs="+",
type=int,
default=VAL_EVERY,
help="How often to validate current architecture.",
)
# Optimisers
parser.add_argument(
"--lr-enc",
type=float,
nargs="+",
default=LR_ENC,
help="Learning rate for encoder.",
)
parser.add_argument(
"--lr-dec",
type=float,
nargs="+",
default=LR_DEC,
help="Learning rate for decoder.",
)
parser.add_argument(
"--mom-enc",
type=float,
nargs="+",
default=MOM_ENC,
help="Momentum for encoder.",
)
parser.add_argument(
"--mom-dec",
type=float,
nargs="+",
default=MOM_DEC,
help="Momentum for decoder.",
)
parser.add_argument(
"--wd-enc",
type=float,
nargs="+",
default=WD_ENC,
help="Weight decay for encoder.",
)
parser.add_argument(
"--wd-dec",
type=float,
nargs="+",
default=WD_DEC,
help="Weight decay for decoder.",
)
parser.add_argument(
"--optim-dec",
type=str,
default=OPTIM_DEC,
help="Optimiser algorithm for decoder.",
)
return parser.parse_args()
def create_segmenter(net, pretrained, num_classes):
"""Create Encoder; for now only ResNet [50,101,152]"""
from models.resnet import rf_lw50, rf_lw101, rf_lw152
if str(net) == "50":
return rf_lw50(num_classes, imagenet=pretrained)
elif str(net) == "101":
return rf_lw101(num_classes, imagenet=pretrained)
elif str(net) == "152":
return rf_lw152(num_classes, imagenet=pretrained)
else:
raise ValueError("{} is not supported".format(str(net)))
def create_loaders(
train_dir,
val_dir,
train_list,
val_list,
shorter_side,
crop_size,
low_scale,
high_scale,
normalise_params,
batch_size,
num_workers,
ignore_label,
):
"""
Returns:
train_loader, val loader
"""
# Torch libraries
from torchvision import transforms
from torch.utils.data import DataLoader
# Custom libraries
from datasets import NYUDataset as Dataset
from datasets import (
Pad,
RandomCrop,
RandomMirror,
ResizeShorterScale,
ToTensor,
Normalise,
)
## Transformations during training ##
composed_trn = transforms.Compose(
[
ResizeShorterScale(shorter_side, low_scale, high_scale),
Pad(crop_size, [123.675, 116.28, 103.53], ignore_label),
RandomMirror(),
RandomCrop(crop_size),
Normalise(*normalise_params),
ToTensor(),
]
)
composed_val = transforms.Compose([Normalise(*normalise_params), ToTensor()])
## Training and validation sets ##
trainset = Dataset(
data_file=train_list,
data_dir=train_dir,
transform_trn=composed_trn,
transform_val=composed_val,
)
valset = Dataset(
data_file=val_list,
data_dir=val_dir,
transform_trn=None,
transform_val=composed_val,
)
logger.info(
" Created train set = {} examples, val set = {} examples".format(
len(trainset), len(valset)
)
)
## Training and validation loaders ##
train_loader = DataLoader(
trainset,
batch_size=batch_size,
shuffle=True,
num_workers=num_workers,
pin_memory=True,
drop_last=True,
)
val_loader = DataLoader(
valset, batch_size=1, shuffle=False, num_workers=0, pin_memory=True
)
return train_loader, val_loader
def create_optimisers(
lr_enc, lr_dec, mom_enc, mom_dec, wd_enc, wd_dec, param_enc, param_dec, optim_dec
):
"""Create optimisers for encoder, decoder and controller"""
optim_enc = torch.optim.SGD(
param_enc, lr=lr_enc, momentum=mom_enc, weight_decay=wd_enc
)
if optim_dec == "sgd":
optim_dec = torch.optim.SGD(
param_dec, lr=lr_dec, momentum=mom_dec, weight_decay=wd_dec
)
elif optim_dec == "adam":
optim_dec = torch.optim.Adam(
param_dec, lr=lr_dec, weight_decay=wd_dec, eps=1e-3
)
return optim_enc, optim_dec
def load_ckpt(ckpt_path, ckpt_dict):
best_val = epoch_start = 0
if os.path.exists(args.ckpt_path):
ckpt = torch.load(ckpt_path)
for (k, v) in ckpt_dict.items():
if k in ckpt:
v.load_state_dict(ckpt[k])
best_val = ckpt.get("best_val", 0)
epoch_start = ckpt.get("epoch_start", 0)
logger.info(
" Found checkpoint at {} with best_val {:.4f} at epoch {}".format(
ckpt_path, best_val, epoch_start
)
)
return best_val, epoch_start
def train_segmenter(
segmenter, train_loader, optim_enc, optim_dec, epoch, segm_crit, freeze_bn
):
"""Training segmenter
"""
train_loader.dataset.set_stage("train")
segmenter.train()
if freeze_bn:
for m in segmenter.modules():
if isinstance(m, nn.BatchNorm2d):
m.eval()
batch_time = AverageMeter()
losses = AverageMeter()
for i, sample in enumerate(train_loader):
start = time.time()
input = sample["image"].cuda()
target = sample["mask"].cuda()
input_var = torch.autograd.Variable(input).float()
target_var = torch.autograd.Variable(target).long()
# Compute output
output = segmenter(input_var)
output = nn.functional.interpolate(
output, size=target_var.size()[1:], mode="bilinear", align_corners=False
)
soft_output = nn.LogSoftmax()(output)
# Compute loss and backpropagate
loss = segm_crit(soft_output, target_var)
optim_enc.zero_grad()
optim_dec.zero_grad()
loss.backward()
optim_enc.step()
optim_dec.step()
losses.update(loss.item())
batch_time.update(time.time() - start)
if i % args.print_every == 0:
logger.info(
" Train epoch: {} [{}/{}]\t"
"Avg. Loss: {:.3f}\t"
"Avg. Time: {:.3f}".format(
epoch, i, len(train_loader), losses.avg, batch_time.avg
)
)
def validate(segmenter, val_loader, epoch, num_classes=-1):
"""Validate segmenter
Args:
segmenter (nn.Module) : segmentation network
val_loader (DataLoader) : training data iterator
epoch (int) : current epoch
num_classes (int) : number of classes to consider
Returns:
Mean IoU (float)
"""
val_loader.dataset.set_stage("val")
segmenter.eval()
cm = np.zeros((num_classes, num_classes), dtype=int)
with torch.no_grad():
for i, sample in enumerate(val_loader):
input = sample["image"]
target = sample["mask"]
input_var = torch.autograd.Variable(input).float().cuda()
# Compute output
output = segmenter(input_var)
output = (
cv2.resize(
output[0, :num_classes].data.cpu().numpy().transpose(1, 2, 0),
target.size()[1:][::-1],
interpolation=cv2.INTER_CUBIC,
)
.argmax(axis=2)
.astype(np.uint8)
)
# Compute IoU
gt = target[0].data.cpu().numpy().astype(np.uint8)
gt_idx = (
gt < num_classes
) # Ignore every class index larger than the number of classes
cm += fast_cm(output[gt_idx], gt[gt_idx], num_classes)
if i % args.print_every == 0:
logger.info(
" Val epoch: {} [{}/{}]\t"
"Mean IoU: {:.3f}".format(
epoch, i, len(val_loader), compute_iu(cm).mean()
)
)
ious = compute_iu(cm)
logger.info(" IoUs: {}".format(ious))
miou = np.mean(ious)
logger.info(" Val epoch: {}\tMean IoU: {:.3f}".format(epoch, miou))
return miou
def main():
global args, logger
args = get_arguments()
logger = logging.getLogger(__name__)
## Add args ##
args.num_stages = len(args.num_classes)
## Set random seeds ##
torch.backends.cudnn.deterministic = True
torch.manual_seed(args.random_seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(args.random_seed)
np.random.seed(args.random_seed)
random.seed(args.random_seed)
## Generate Segmenter ##
segmenter = nn.DataParallel(
create_segmenter(args.enc, args.enc_pretrained, args.num_classes[0])
).cuda()
logger.info(
" Loaded Segmenter {}, ImageNet-Pre-Trained={}, #PARAMS={:3.2f}M".format(
args.enc, args.enc_pretrained, compute_params(segmenter) / 1e6
)
)
## Restore if any ##
best_val, epoch_start = load_ckpt(args.ckpt_path, {"segmenter": segmenter})
## Criterion ##
segm_crit = nn.NLLLoss2d(ignore_index=args.ignore_label).cuda()
## Saver ##
saver = Saver(
args=vars(args),
ckpt_dir=args.snapshot_dir,
best_val=best_val,
condition=lambda x, y: x > y,
) # keep checkpoint with the best validation score
logger.info(" Training Process Starts")
for task_idx in range(args.num_stages):
start = time.time()
torch.cuda.empty_cache()
## Create dataloaders ##
train_loader, val_loader = create_loaders(
args.train_dir,
args.val_dir,
args.train_list[task_idx],
args.val_list[task_idx],
args.shorter_side[task_idx],
args.crop_size[task_idx],
args.low_scale[task_idx],
args.high_scale[task_idx],
args.normalise_params,
args.batch_size[task_idx],
args.num_workers,
args.ignore_label,
)
if args.evaluate:
return validate(
segmenter, val_loader, 0, num_classes=args.num_classes[task_idx]
)
logger.info(" Training Stage {}".format(str(task_idx)))
## Optimisers ##
enc_params = []
dec_params = []
for k, v in segmenter.named_parameters():
if bool(re.match(".*conv1.*|.*bn1.*|.*layer.*", k)):
enc_params.append(v)
logger.info(" Enc. parameter: {}".format(k))
else:
dec_params.append(v)
logger.info(" Dec. parameter: {}".format(k))
optim_enc, optim_dec = create_optimisers(
args.lr_enc[task_idx],
args.lr_dec[task_idx],
args.mom_enc[task_idx],
args.mom_dec[task_idx],
args.wd_enc[task_idx],
args.wd_dec[task_idx],
enc_params,
dec_params,
args.optim_dec,
)
for epoch in range(args.num_segm_epochs[task_idx]):
train_segmenter(
segmenter,
train_loader,
optim_enc,
optim_dec,
epoch_start,
segm_crit,
args.freeze_bn[task_idx],
)
if (epoch + 1) % (args.val_every[task_idx]) == 0:
miou = validate(
segmenter, val_loader, epoch_start, args.num_classes[task_idx]
)
# assert 1==2
saver.save(
miou,
{"segmenter": segmenter.state_dict(), "epoch_start": epoch_start},
logger,
)
epoch_start += 1
logger.info(
"Stage {} finished, time spent {:.3f}min".format(
task_idx, (time.time() - start) / 60.0
)
)
logger.info(
"All stages are now finished. Best Val is {:.3f}".format(saver.best_val)
)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
main()
GUI.py
# encoding:utf-8
import tkinter.font as tkFont
import tkinter.messagebox
import tkinter.ttk as ttk
import time
import threading
from tkinter import *
import cv2
from PIL import ImageTk,Image
import inspect
import ctypes
import sys
from model_process import proccess,build_model,transform_color
import torch
import numpy as np
import copy
import numpy as np
import matplotlib.pyplot as plt
import mpl_toolkits.axisartist as axisartist
from matplotlib.backends.backend_agg import FigureCanvasAgg
class GUI(Frame):
def __init__(self, master=None):
super().__init__(master)
self.master = master
self.initComponent(master)
# 初始化变量
self.thread0 = 1 # 读取视频流线程
self.thread3 = 1 # 分割线程
self.thread4 = 1 # 测试线程
self.ret = False # 摄像头打开标识符
#self.frame = None # 图像
self.pred_index = None #
self.numclass = 18 # 分割类别数
self.ID_index = [11, 12, 13, 14, 15]
self.Weight_index = [0.4,0.2,0.2,0.1,0.1]
self.Privacy = [] # 记录隐私值
self.show_h = 470
self.show_w = 470
self.privacy_num = 25
self.current_privacy = 0
self.segmenter = None
def initComponent(self, master):
master.rowconfigure(0, weight=1)
master.columnconfigure(0, weight=1)
self.ft = tkFont.Font(family='微软雅黑', size=12, weight='bold')
self.grid(row=0, column=0, sticky=NSEW)
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
self.panewin = ttk.Panedwindow(self, orient=HORIZONTAL)
self.panewin.grid(row=0, column=0, sticky=NSEW)
self.frm_left = ttk.Frame(self.panewin, relief=SUNKEN, padding=0)
self.frm_left.grid(row=0, column=0, sticky=NS)
self.panewin.add(self.frm_left)
self.computelist()
self.frm_right = ttk.Frame(self.panewin, relief=SUNKEN)
self.frm_right.grid(row=0, column=1, sticky=NSEW)
self.panewin.add(self.frm_right)
self.video_play()
def computelist(self):
self.frm_left.rowconfigure(0, weight=1)
self.frm_left.columnconfigure(0, weight=1)
self.Button_height = 5
self.Button_width = 10
self.Button_bd = 5
self.Button_color1 = 'red'
self.Button_color2 = 'green'
self.start = Button(self.frm_left, text="开始", command=self.start_fun, bd=self.Button_bd, bg=self.Button_color2,
width=self.Button_width) #
self.start.grid(row = 0,column=0,sticky=W)
self.segment = Button(self.frm_left, text="分割", command=self.segment_fun, bd=self.Button_bd, bg=self.Button_color2,
width=self.Button_width)
self.segment.grid(row = 1,column=0,sticky=W)
self.measure = Button(self.frm_left, text="测量", command=self.measure_fun, bd=self.Button_bd, bg=self.Button_color2,
width=self.Button_width)
self.measure.grid(row = 2,column=0,sticky=W)
# 设置摄像头的索引滑动条
self.index_label = Label(self.frm_left, text="摄像头编号", font='Helvetica -12 bold',width=self.Button_width)
self.index_label.grid(row = 3,column=0,sticky=W)
self.scale = Scale(self.frm_left, from_=0, to=4, orient=tkinter.HORIZONTAL, command="通过scale.get()获取值",width=self.Button_width)
self.scale.set(2)
self.scale.grid(row = 4,column=0,sticky=W)
# 设置展示时间的label
self.times_label = Label(self.frm_left, text="时间",font='Helvetica -12 bold',width=self.Button_width)
self.times_label.grid(row = 5,column=0,sticky=W)
self.times = Label(self.frm_left,width=self.Button_width)
self.times.grid(row = 6,column=0,sticky=W)
for i in range(7): # 为每列添加权重值以便水平拉伸
self.frm_left.rowconfigure(i, weight=1)
#self.frm_left.rowconfigure(2,weight = 1)
def video_play(self):
# 展示视频的label
self.Label_old = Label(self.frm_right)
self.Label_old.pack(side = LEFT,fill="y")
#print(self.frm_right.size)
self.Label_segment = Label(self.frm_right)
self.Label_segment.pack(side = LEFT,fill = "y")
self.Label_measure = Label(self.frm_right)
self.Label_measure.pack(side = LEFT,fill = "y")
'''
def video_play(self):
self.Label_old = Label(self.frm_right)
self.label_introduce = Label(self.frm_right)
text = u"~该隐私度量系统中，所用语义分割技术是基于refineNet的深度学习网络，同时该系统主要集合集对分析理论等技术实现对图像中人体隐私数学模型的建立，并通过摄像头实现对人体隐私的实时度量。~"
font =("华文楷体",20)
fg = "blue"
self.label_introduce.config(text=text.encode("gb2312"),font = font,fg = fg)
#self.label_introduce.text = "12121212"
self.label_introduce.pack()
self.Label_old.pack(side = LEFT,fill="y")
#print(self.frm_right.size)
self.Label_segment = Label(self.frm_right)
self.Label_segment.pack(side = LEFT,fill = "y")
self.Label_measure = Label(self.frm_right)
self.Label_measure.pack(side = LEFT,fill = "y")
'''
def read_camera(self):
self.index = self.scale.get()
#self.cap = cv2.VideoCapture(self.index + cv2.CAP_DSHOW)
self.cap = cv2.VideoCapture(0)
self.cap.set(5, 60) # 设置帧率
while True:
self.ret, camare_readpic = self.cap.read()
if self.ret:
self.frame = cv2.resize(camare_readpic, (int(camare_readpic.shape[1]/2), int(camare_readpic.shape[0]/2)), cv2.INTER_CUBIC)
self.segment.config(state=NORMAL)
self.measure.config(state=NORMAL)
camare_readpic = cv2.resize(camare_readpic,(self.show_w,self.show_h))
cv2image = cv2.cvtColor(camare_readpic, cv2.COLOR_BGR2RGB) # 转换颜色从BGR到RGBA
image = Image.fromarray(cv2image)
imgtks = ImageTk.PhotoImage(image=image)
self.Label_old.config(image = imgtks)
self.Label_old.imgtk = imgtks
else:
self.frame = None
self.frame = None
self.segment.config(state=DISABLED)
self.measure.config(state=DISABLED)
# 控件函数
def start_fun(self):
try:
self.stop_thread(self.thread0)
time.sleep(1)
print("have stop！")
except:
pass
try:
if self.thread0.isAlive():
pass
except:
self.thread0 = threading.Thread(target=self.read_camera)
self.thread0.setDaemon(True)
self.thread0.start()
print("restart")
self.thread1 = threading.Thread(target=self.clock)
self.thread1.setDaemon(True)
self.thread1.start()
def segment_net(self):
while True:
if self.frame is not None:
input = proccess(self.frame)
with torch.no_grad():
input_var = torch.autograd.Variable(input).float()
output = self.segmenter(input_var)
self.pred_index = (
cv2.resize(
output[0, :self.numclass].data.cpu().numpy().transpose(1, 2, 0),
input.shape[2:4][::-1],
interpolation=cv2.INTER_CUBIC,
)
.argmax(axis=2)
.astype(np.uint8)
)
color_pic = transform_color(self.pred_index)
color_pic = cv2.resize(color_pic, (self.show_w, self.show_h))
color_pic = cv2.cvtColor(color_pic, cv2.COLOR_BGR2RGBA)
seg_pic = Image.fromarray(color_pic)
seg_pic = ImageTk.PhotoImage(image=seg_pic)
self.Label_segment.config(image=seg_pic)
self.Label_segment.imgtk = seg_pic
def segment_fun(self):
try:
self.stop_thread(self.thread3)
time.sleep(1)
print("have stop！")
except:
self.segmenter = build_model()
self.segmenter.eval()
try:
if self.thread3.isAlive():
pass
except:
self.thread3 = threading.Thread(target=self.segment_net)
self.thread3.setDaemon(True)
self.thread3.start()
def privacy_measure(self):
t1 = time.time()
ID_index = [4,6,11,14,15] #4胸，6裤子，11人脸，14胳膊，15手
Weight_index = [0.3,0.3,0.3,0.05,0.05]
assert len(Weight_index) == len(ID_index)
while True:
t2 = time.time()
if (t2 - t1) > 0:
if self.frame is not None and self.pred_index is not None:
#转通道
new = cv2.cvtColor(self.frame,cv2.COLOR_RGB2YCrCb)
Cr = new[:,:,1]
Cb = new[:,:,2]
k1 = Cr>0
k2 = (Cb>133) * (Cb<173)
skin_bineary = (k1*k2).astype(np.float)
p = 1.0
A = 0.0
C = 0.0
cont = 0
for index in ID_index:
sumall = np.sum(self.pred_index == index)
if sumall > 0:
a = np.sum((self.pred_index*skin_bineary) ==index)/sumall
else:
a = 0.0
b = p * a * (1-a)
c = 1 - a - b
A += Weight_index[cont] * a
C += Weight_index[cont] * c
cont += 1
self.Privacy.append(A/C)
print("隐私值%.2f"%(A/C))
else:
continue
t1 = t2
else:
continue
if len(self.Privacy) > self.privacy_num:
self.Privacy = self.Privacy[len(self.Privacy)-self.privacy_num:]
#time.sleep(0.2)
# self.Privacy = 1
####################################################################################
plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False # 用来正常显示负号
from matplotlib.pyplot import MultipleLocator
# 创建画布
fig = plt.figure(figsize=(6, 6))
# 创建一个绘图区对象
ax = axisartist.Subplot(fig, 111)
# 将对象添加到画布中
fig.add_axes(ax)
# 设置绘图区所有坐标轴隐藏
ax.axis[:].set_visible(False)
# 添加新的坐标轴
ax.axis["x"] = ax.new_floating_axis(0, 0) # 第一个0代表平行直线，第二个0代表经过0点
# 添加箭头
ax.axis["x"].set_axisline_style("-|>", size=1.0) # ->表示空箭头
ax.axis["y"] = ax.new_floating_axis(1, 0) # 竖直直线经过0点
ax.axis["y"].set_axisline_style("-|>", size=1.0) # -|> 表示实心箭头
# 设置x,y轴上刻度显示方向
ax.axis["x"].set_visible(True) # 设置是否可见
ax.axis["x"].set_axis_direction("bottom") # 刻度在上面还是下面
ax.axis["y"].set_axis_direction("left") # 刻度在左边还是右边
# 设置坐标轴的名称及格式
fontstyle = {'family': 'Times New Roman',
'weight': 'normal',
'size': 30, }
ax.axis["x"].label.set_text("times")
# ax.axis["x"].label.set_color("black") # 设置颜色
ax.axis["x"].label.set_fontsize(10) # 设置字体大小
# ax.axis["x"].label.set_rotation(45) # 设置旋转角度
ax.axis["y"].label.set_text("privacy")
ax.axis["y"].label.set_fontsize(10) # 设置字体大小
# 设置坐标轴刻度范围
plt.xticks(np.arange(0, int(self.privacy_num*2), int(self.privacy_num*2)/10))
plt.yticks(np.arange(-1, 3, 0.5))
# 设置底边距
plt.subplots_adjust(bottom=0.10)
# 开启网格
# ax.grid(True, linestyle='-.') # 开启网格并设置样式
if self.Privacy[-1] > plt.axis()[3]:
self.Privacy[-1] = plt.axis()[3]
if self.Privacy[-1] > 1:
color = 'r'
else:
color = 'g'
plt.plot(self.Privacy,'o:')
#linex = np.arange(0, int(self.privacy_num*2)
#liney = [1]*len(linex)
linex = np.array([0, int(self.privacy_num*2)])
liney = np.array([self.Privacy[-1], self.Privacy[-1]])
plt.plot(linex, liney,color=color)
plt.xlim(0,int(self.privacy_num * 2))
plt.ylim(-1,3)
canvas = FigureCanvasAgg(plt.gcf())
canvas.draw()
image = np.array(canvas.renderer.buffer_rgba())
######################################################################################
#curve = self.plot_curve(self.Privacy).astype(np.uint8)
image = cv2.resize(image,(self.show_w,self.show_h))
#measure_curve = cv2.cvtColor(image, cv2.COLOR_RGB2RGBA)
measure_curve = Image.fromarray(image)
measure_curve = ImageTk.PhotoImage(image=measure_curve)
self.Label_measure.config(image= measure_curve)
self.Label_measure.imgtk = measure_curve
def measure_fun(self):
#if self.pred_index is not None:
if True:
self.thread4 = threading.Thread(target=self.privacy_measure)
self.thread4.setDaemon(True)
self.thread4.start()
else:
self.Label_measure.config(state = DISABLED)
def clock(self):
global systimes
current_times = time.strftime("%H:%M:%S")
if systimes != current_times:
systimes = current_times
self.times.config(text=current_times)
# calls itself every 200 milliseconds
# to update the time display as needed
# could use >200 ms, but display gets jerky
self.times.after(200, self.clock)
# 线程关闭函数
def _async_raise(self,tid, exctype):
"""raises the exception, performs cleanup if needed"""
tid = ctypes.c_long(tid)
if not inspect.isclass(exctype):
exctype = type(exctype)
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
if res == 0:
raise ValueError("invalid thread id")
elif res != 1:
ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
raise SystemError("PyThreadState_SetAsyncExc failed")
def stop_thread(self,thread):
self._async_raise(thread.ident, SystemExit)
if __name__ == '__main__':
systimes = ''
root = Tk()
root.geometry('1300x400')
root.title('隐私度量系统')
# root.option_add("*Font", "宋体")
root.minsize(1300, 400)
#root.maxsize(1300, 400)
app = GUI(root)
root.mainloop()
cv2.destroyAllWindows()
util.py
"""Helper definitions"""
import json
import os
import torch
def compute_params(model):
"""Compute number of parameters"""
n_total_params = 0
for name, m in model.named_parameters():
n_elem = m.numel()
n_total_params += n_elem
return n_total_params
# Adopted from https://raw.githubusercontent.com/pytorch/examples/master/imagenet/main.py
class AverageMeter(object):
"""Computes and stores the average and current value"""
def __init__(self):
self.reset()
def reset(self):
self.val = 0
self.avg = 0
self.sum = 0
self.count = 0
def update(self, val, n=1):
self.val = val
self.sum += val * n
self.count += n
self.avg = self.sum / self.count
class Saver:
"""Saver class for managing parameters"""
def __init__(self, args, ckpt_dir, best_val=0, condition=lambda x, y: x > y):
"""
Args:
args (dict): dictionary with arguments.
ckpt_dir (str): path to directory in which to store the checkpoint.
best_val (float): initial best value.
condition (function): how to decide whether to save the new checkpoint
by comparing best value and new value (x,y).
"""
if not os.path.exists(ckpt_dir):
os.makedirs(ckpt_dir)
with open("{}/args.json".format(ckpt_dir), "w") as f:
json.dump(
{k: v for k, v in args.items() if isinstance(v, (int, float, str))},
f,
sort_keys=True,
indent=4,
ensure_ascii=False,
)
self.ckpt_dir = ckpt_dir
self.best_val = best_val
self.condition = condition
self._counter = 0
def _do_save(self, new_val):
"""Check whether need to save"""
return self.condition(new_val, self.best_val)
def save(self, new_val, dict_to_save, logger):
"""Save new checkpoint"""
self._counter += 1
if self._do_save(new_val):
logger.info(
" New best value {:.4f}, was {:.4f}".format(new_val, self.best_val)
)
self.best_val = new_val
dict_to_save["best_val"] = new_val
torch.save(dict_to_save, "{}/checkpoint.pth.tar".format(self.ckpt_dir))
return True
return False
下载
点击此处下载，下载前请留下您的信息方便与我们沟通-
关于我们
贵州大学现代制造技术教育部重点实验室杨观赐教授所带领的这支研究智能家居移动服务机器人的团队是由一群醉心于学术、勤勉于实践、热爱生活的研究生们组成的优秀团队。