The code for my project is stored on this repository. Please note this repository may continue to grow even after portfolio submission.
https://github.com/DevkumarPatel/Getting-Started-with-Crowd-Counting
from ctypes import resize
import os
import cv2
import glob
import h5py
import tqdm
import torch
import random
import numpy as np
from PIL import Image
import scipy.io as sio
import matplotlib.cm as CM
import matplotlib.pyplot as plt
from torch.utils.data import Dataset
from scipy.ndimage.filters import gaussian_filter
from model import CSRMSA
def generate_h5py_density( image_path, matrix_path, h5_path ):
""" Generates a density map file in h5py format.
Args:
image_path (str) : The path to the set of images corresponding to matrix file
matrix_path (str): The file paths containing .mat
h5_path (str): The output folder for all h5_files with format *.h5
Returns:
List of image paths.
Example:
train_data_list = generate_h5py_density(
'ShanghaiTech/part_A_final/train_data/images',
'ShanghaiTech/part_A_final/train_data/ground_truth',
'ShanghaiTech/part_A_final/train_data/h5'
)
References:
Adopted from CSRNet pytorch version
Github:
"""
images = glob.glob( image_path+"/*.jpg")
for image in tqdm.tqdm(images):
img = plt.imread(image)
np_img = np.zeros( (img.shape[0],img.shape[1]) )
mat = sio.loadmat( (matrix_path +"/"+ os.path.basename(image).replace('jpg', 'mat').replace('IMG_', 'GT_IMG_')) )
gt_coord = mat["image_info"][0,0][0,0][0]
for i in range( 0, len(gt_coord) ):
if int(gt_coord[i][1]) < img.shape[0]: # checks if the coordinates given are under image dimensions
if int(gt_coord[i][0]) < img.shape[1]:
np_img[ int(gt_coord[i][1]), int(gt_coord[i][0]) ] = 1 # Populate the cell with 1.
np_img = gaussian_filter( np_img, sigma=10 ) # Run a Gausian filter
with h5py.File( (h5_path +"/"+ os.path.basename(image).replace('jpg','h5')), 'w' ) as h_file:
h_file['density'] = np_img
return images
def display_density(path):
"""
Loads a h5py file with dict key 'density'
and then displays it using matplotlib.
Args:
path (str) : the file path to h5py file
Example:
img_gaus_density = display_density(
"ShanghaiTech/part_A_final/train_data/h5/IMG_1.h5")
Returns:
np.array of 'density'
"""
gt = h5py.File(path, 'r')
density = np.asarray(gt['density'])
plt.title("Count: " + str(int( density.sum()) ) )
plt.imshow(density,cmap=CM.jet)
return density
def predict_size(x):
#x = (x+1)//2
#x = (x+1)//2
#x = (x+1)//2
#x = (x+1)//2
return x // 8
def load_data(img_path,model, train = True):
gt_path = img_path.replace('.jpg','.h5').replace('images','ground-truth')
img = Image.open(img_path).convert('RGB')
gt_file = h5py.File(gt_path,'r')
target = np.asarray(gt_file['density'])
crop_size = (int(img.size[0]), int(img.size[1]))
if train:
ratio = 0.5
crop_size = (int(img.size[0]*ratio),int(img.size[1]*ratio))
rdn_value = random.random()
if rdn_value<0.25:
dx = 0
dy = 0
elif rdn_value<0.5:
dx = int(img.size[0]*ratio)
dy = 0
elif rdn_value<0.75:
dx = 0
dy = int(img.size[1]*ratio)
else:
dx = int(img.size[0]*ratio)
dy = int(img.size[1]*ratio)
img = img.crop((dx,dy,crop_size[0]+dx,crop_size[1]+dy))
target = target[dy:(crop_size[1]+dy),dx:(crop_size[0]+dx)]
if random.random()>0.8:
target = np.fliplr(target)
img = img.transpose(Image.FLIP_LEFT_RIGHT)
target = cv2.resize(target,( predict_size(target.shape[1]) ,predict_size(target.shape[0])),interpolation = cv2.INTER_CUBIC)*64
return img,target
class listDataset(Dataset):
def __init__(self, root, shape=None, shuffle=True, transform=None, train=False, seen=0, batch_size=1, num_workers=4, model=CSRMSA()):
# if train:
# root = root *4
random.shuffle(root)
self.nSamples = len(root)
self.lines = root
self.transform = transform
self.train = train
self.shape = shape
self.seen = seen
self.batch_size = batch_size
self.num_workers = num_workers
self.model = model
def __len__(self):
return self.nSamples
def __getitem__(self, index):
assert index <= len(self), 'index range error'
img_path = self.lines[index]
img,target = load_data(img_path,self.model, self.train)
if self.transform is not None:
img = self.transform(img)
return img,target
# code obtained from <https://github.com/xxxnell/how-do-vits-work>
import types
import math
import random
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
from einops.layers.torch import Rearrange
def conv1x1(in_channels, out_channels, stride=1, groups=1):
return convnxn(in_channels, out_channels, kernel_size=1, stride=stride, groups=groups)
def conv3x3(in_channels, out_channels, stride=1, groups=1):
return convnxn(in_channels, out_channels, kernel_size=3, stride=stride, groups=groups, padding=1)
def convnxn(in_channels, out_channels, kernel_size, stride=1, groups=1, padding=0):
return nn.Conv2d(in_channels, out_channels,
kernel_size=kernel_size, stride=stride, padding=padding, groups=groups, bias=False)
def relu():
return nn.ReLU()
def relu6(mx=6.0, mn=0.0):
return Lambda(lambda x: torch.clamp(x, mn, mx))
def bn(dim):
return nn.BatchNorm2d(dim)
def ln1d(dim):
return nn.LayerNorm(dim)
def ln2d(dim):
return nn.Sequential(
Rearrange("b c h w -> b h w c"),
nn.LayerNorm(dim),
Rearrange("b h w c -> b c h w"),
)
def dense(in_features, out_features, bias=True):
return nn.Linear(in_features, out_features, bias)
def blur(in_filters, sfilter=(1, 1), pad_mode="constant"):
if tuple(sfilter) == (1, 1) and pad_mode in ["constant", "zero"]:
layer = nn.AvgPool2d(kernel_size=2, stride=1, padding=1)
else:
layer = Blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
return layer
def drop_path(x, drop_prob=0.0, training=False):
if drop_prob == 0. or not training:
return x
keep_prob = 1 - drop_prob
shape = (x.shape[0],) + (1,) * (x.ndim - 1) # work with diff dim tensors, not just 2D ConvNets
random_tensor = keep_prob + torch.rand(shape, dtype=x.dtype, device=x.device)
random_tensor.floor_() # binarize
output = x.div(keep_prob) * random_tensor
return output
class SamePad(nn.Module):
def __init__(self, filter_size, pad_mode="constant", **kwargs):
super(SamePad, self).__init__()
self.pad_size = [
int((filter_size - 1) / 2.0), int(math.ceil((filter_size - 1) / 2.0)),
int((filter_size - 1) / 2.0), int(math.ceil((filter_size - 1) / 2.0)),
]
self.pad_mode = pad_mode
def forward(self, x):
x = F.pad(x, self.pad_size, mode=self.pad_mode)
return x
def extra_repr(self):
return "pad_size=%s, pad_mode=%s" % (self.pad_size, self.pad_mode)
class Blur(nn.Module):
def __init__(self, in_filters, sfilter=(1, 1), pad_mode="replicate", **kwargs):
super(Blur, self).__init__()
filter_size = len(sfilter)
self.pad = SamePad(filter_size, pad_mode=pad_mode)
self.filter_proto = torch.tensor(sfilter, dtype=torch.float, requires_grad=False)
self.filter = torch.tensordot(self.filter_proto, self.filter_proto, dims=0)
self.filter = self.filter / torch.sum(self.filter)
self.filter = self.filter.repeat([in_filters, 1, 1, 1])
self.filter = torch.nn.Parameter(self.filter, requires_grad=False)
def forward(self, x):
x = self.pad(x)
x = F.conv2d(x, self.filter, groups=x.size()[1])
return x
def extra_repr(self):
return "pad=%s, filter_proto=%s" % (self.pad, self.filter_proto.tolist())
class Downsample(nn.Module):
def __init__(self, strides=(2, 2), **kwargs):
super(Downsample, self).__init__()
if isinstance(strides, int):
strides = (strides, strides)
self.strides = strides
def forward(self, x):
shape = (-(-x.size()[2] // self.strides[0]), -(-x.size()[3] // self.strides[1]))
x = F.interpolate(x, size=shape, mode='nearest')
return x
def extra_repr(self):
return "strides=%s" % repr(self.strides)
class DropPath(nn.Module):
def __init__(self, p, **kwargs):
super().__init__()
self.p = p
def forward(self, x):
x = drop_path(x, self.p, self.training)
return x
def extra_repr(self):
return "p=%s" % repr(self.p)
class Lambda(nn.Module):
def __init__(self, lmd):
super().__init__()
if not isinstance(lmd, types.LambdaType):
raise Exception("`lmd` should be lambda ftn.")
self.lmd = lmd
def forward(self, x):
return self.lmd(x)
import torch
import torch.nn as nn
import smoothing_block as smoothing
import vggnet_mcdo_block
import vggnet_dnn_block
class CSRMSA(nn.Module):
def __init__(self,
block=vggnet_dnn_block.BasicBlock,
num_blocks=[2, 2, 3, 3, 3, 3, 1, 1 ],
sblock=smoothing.TanhBlurBlock,
num_sblocks=(1,1,1,1,1,1, 1, 1),
name="CSRMSA"):
super(CSRMSA, self).__init__()
self.seen = 0
self.name = name
self.layer0 = self._make_layer(block, 3, 64, num_blocks[0], pool=False)
self.layer1 = self._make_layer(block, 64, 128, num_blocks[1], pool=True)
self.layer2 = self._make_layer(block, 128, 256, num_blocks[2], pool=True)
self.layer3 = self._make_layer(block, 256, 512, num_blocks[3], pool=True)
self.layer4 = self._make_layer(block, 512, 512, num_blocks[4], pool=True)
## CSR backend
#self.layer5 = self._make_layer(block, 512, 256, num_blocks[5], pool=False)
#self.layer6 = self._make_layer(block, 256, 128, num_blocks[6], pool=False)
#self.layer7 = self._make_layer(block, 128, 64, num_blocks[7], pool=False)
self.smooth0 = self._make_smooth_layer(sblock, 64, num_sblocks[0])
self.smooth1 = self._make_smooth_layer(sblock, 128, num_sblocks[1])
self.smooth2 = self._make_smooth_layer(sblock, 256, num_sblocks[2])
self.smooth3 = self._make_smooth_layer(sblock, 512, num_sblocks[3])
self.smooth4 = self._make_smooth_layer(sblock, 512, num_sblocks[4])
self.smooth5 = self._make_smooth_layer(sblock, 256, num_sblocks[5])
self.smooth6 = self._make_smooth_layer(sblock, 256, num_sblocks[6])
self.smooth7 = self._make_smooth_layer(sblock, 256, num_sblocks[7])
self.output_layer = self._make_layer(block, 64, 1, 1, pool=False)
self.backend_feat = [512, 512, 512, 256,128,64]
self.backend = self._make_backfeat(self.backend_feat, in_channels = 512, dilation = True)
def forward(self, x):
x = self.layer0(x) #64
x = self.smooth0(x)
x = self.layer1(x) #128
x = self.smooth1(x)
x = self.layer2(x) #256
x = self.smooth2(x)
x = self.layer3(x) #512
x = self.smooth3(x)
#x = self.layer4(x) #512
#x = self.smooth4(x)
#x = self.layer5(x)
#x = self.smooth5(x)
#x = self.layer6(x)
#x = self.smooth7(x)
#x = self.layer7(x)
#x = self.smooth7(x)
x = self.backend(x)
x = self.output_layer(x)
return x
# below static methods retrieved from
# <https://github.com/xxxnell/how-do-vits-work>
@staticmethod
def _make_layer(block, in_channels, out_channels, num_blocks, pool):
layers, channels = [], in_channels
if pool:
layers.append(nn.MaxPool2d(kernel_size=2, stride=2))
for _ in range(num_blocks):
layers.append(block(channels, out_channels))
channels = out_channels
return nn.Sequential(*layers)
@staticmethod
def _make_smooth_layer(sblock, in_filters, num_blocks):
layers = []
for _ in range(num_blocks):
layers.append(sblock(in_filters=in_filters))
return nn.Sequential(*layers)
@staticmethod
def _make_backfeat(cfg, in_channels = 3,batch_norm=False,dilation = False):
if dilation:
d_rate = 2
else:
d_rate = 1
layers = []
for v in cfg:
if v == 'M':
layers += [nn.MaxPool2d(kernel_size=2, stride=2)]
else:
conv2d = nn.Conv2d(in_channels, v, kernel_size=3, padding=d_rate,dilation = d_rate)
if batch_norm:
layers += [conv2d, nn.BatchNorm2d(v), nn.ReLU(inplace=True)]
else:
layers += [conv2d, nn.ReLU(inplace=True)]
in_channels = v
return nn.Sequential(*layers)
# model obtained from CSRNet paper.
class CSRNet(nn.Module):
def __init__(self, load_weights=False):
super(CSRNet, self).__init__()
self.seen = 0
self.frontend_feat = [64, 64, 'M', 128, 128, 'M', 256, 256, 256, 'M', 512, 512, 512]
self.backend_feat = [512, 512, 512,256,128,64]
self.frontend = make_layers(self.frontend_feat)
self.backend = make_layers(self.backend_feat,in_channels = 512,dilation = True)
self.output_layer = nn.Conv2d(64, 1, kernel_size=1)
"""
if not load_weights:
mod = models.vgg16(pretrained = True)
self._initialize_weights()
for i in xrange(len(self.frontend.state_dict().items())):
self.frontend.state_dict().items()[i][1].data[:] = mod.state_dict().items()[i][1].data[:]
"""
def forward(self,x):
x = self.frontend(x)
x = self.backend(x)
x = self.output_layer(x)
return x
def _initialize_weights(self):
for m in self.modules():
if isinstance(m, nn.Conv2d):
nn.init.normal_(m.weight, std=0.01)
if m.bias is not None:
nn.init.constant_(m.bias, 0)
elif isinstance(m, nn.BatchNorm2d):
nn.init.constant_(m.weight, 1)
nn.init.constant_(m.bias, 0)
def make_layers(cfg, in_channels = 3,batch_norm=False,dilation = False):
if dilation:
d_rate = 2
else:
d_rate = 1
layers = []
for v in cfg:
if v == 'M':
layers += [nn.MaxPool2d(kernel_size=2, stride=2)]
else:
conv2d = nn.Conv2d(in_channels, v, kernel_size=3, padding=d_rate,dilation = d_rate)
if batch_norm:
layers += [conv2d, nn.BatchNorm2d(v), nn.ReLU(inplace=True)]
else:
layers += [conv2d, nn.ReLU(inplace=True)]
in_channels = v
return nn.Sequential(*layers)
smoothing_block.py
import torch
import torch.nn as nn
import layers as layers
class TanhBlurBlock(nn.Module):
def __init__(self, in_filters, temp=1e1, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(TanhBlurBlock, self).__init__()
self.temp = temp
self.relu = layers.relu()
self.tanh = nn.Tanh()
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = self.temp * self.tanh(x / self.temp)
x = self.relu(x)
x = self.blur(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class BNTanhBlurBlock(nn.Module):
def __init__(self, in_filters, temp=1e1, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(BNTanhBlurBlock, self).__init__()
self.bn = layers.bn(in_filters)
self.temp = temp
self.relu = layers.relu()
self.tanh = nn.Tanh()
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = self.bn(x)
x = self.temp * self.tanh(x / self.temp)
x = self.relu(x)
x = self.blur(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class TanhBlock(nn.Module):
def __init__(self, temp=1e1, **kwargs):
super(TanhBlock, self).__init__()
self.temp = temp
self.relu = layers.relu()
self.tanh = nn.Tanh()
def forward(self, x):
x = self.temp * self.tanh(x / self.temp)
x = self.relu(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class SigmoidBlurBlock(nn.Module):
def __init__(self, in_filters, temp=1e1, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(SigmoidBlurBlock, self).__init__()
self.temp = temp
self.relu = layers.relu()
self.sigmoid = nn.Sigmoid()
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = 4 * self.temp * (self.sigmoid(x / self.temp) - 0.5)
x = self.relu(x)
x = self.blur(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class SoftmaxBlurBlock(nn.Module):
def __init__(self, in_filters, temp=1e1, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(SoftmaxBlurBlock, self).__init__()
self.temp = temp
self.relu = layers.relu()
self.softmax = nn.Softmax(dim=1)
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = self.temp * self.softmax(x / self.temp)
x = self.relu(x)
x = self.blur(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class ReLuBlurBlock(nn.Module):
def __init__(self, in_filters, temp=6.0, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(ReLuBlurBlock, self).__init__()
self.temp = temp
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = torch.clamp(x, 0.0, self.temp)
x = self.blur(x)
return x
def extra_repr(self):
return "thr=%.3e" % self.thr
class ScalingBlurBlock(nn.Module):
def __init__(self, in_filters, temp=5e0, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(ScalingBlurBlock, self).__init__()
self.temp = temp
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = x / self.temp
x = self.blur(x)
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class ScalingBlock(nn.Module):
def __init__(self, temp=5e0, **kwargs):
super(ScalingBlock, self).__init__()
self.temp = temp
def forward(self, x):
x = x / self.temp
return x
def extra_repr(self):
return "temp=%.3e" % self.temp
class ReLuBlock(nn.Module):
def __init__(self, thr=6.0, **kwargs):
super(ReLuBlock, self).__init__()
self.thr = thr
def forward(self, x):
x = torch.clamp(x, 0.0, self.thr)
return x
def extra_repr(self):
return "thr=%.3e" % self.thr
class BNBlurBlock(nn.Module):
def __init__(self, in_filters, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(BNBlurBlock, self).__init__()
self.bn = layers.bn(in_filters)
self.relu = layers.relu()
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = self.bn(x)
x = self.relu(x)
x = self.blur(x)
return x
class BlurBlock(nn.Module):
def __init__(self, in_filters, sfilter=(1, 1), pad_mode="constant", **kwargs):
super(BlurBlock, self).__init__()
self.blur = layers.blur(in_filters, sfilter=sfilter, pad_mode=pad_mode)
def forward(self, x):
x = self.blur(x)
return x
import h5py
import torch
import shutil
import numpy as np
def save_net(fname, net):
with h5py.File(fname, 'w') as h5f:
for k, v in net.state_dict().items():
h5f.create_dataset(k, data=v.cpu().numpy())
def load_net(fname, net):
with h5py.File(fname, 'r') as h5f:
for k, v in net.state_dict().items():
param = torch.from_numpy(np.asarray(h5f[k]))
v.copy_(param)
def save_checkpoint(state, is_best, filename='checkpoint.pth.tar'):
torch.save(state, filename)
if is_best:
shutil.copyfile(filename, 'model_best.pth.tar')
vggnet_dnn_block.py
# code obtained from <https://github.com/xxxnell/how-do-vits-work>
import torch
import torch.nn as nn
import layers as layers
class BasicBlock(nn.Module):
def __init__(self, in_channels, out_channels, **block_kwargs):
super(BasicBlock, self).__init__()
self.conv = layers.conv3x3(in_channels, out_channels)
self.bn = layers.bn(out_channels)
self.relu = layers.relu()
def forward(self, x):
x = self.conv(x)
x = self.bn(x)
x = self.relu(x)
return x