mitterrand_dev

.github/workflows/pr_test_cpu.yml

@@ -78,6 +78,16 @@ jobs:
         cd enhance
         python3 -m pytest
         docker container rm -f deepstack
+    - name: Test Landmark Detection
+      run: |
+        docker run -d --name deepstack -e VISION_LANDMARK=True -p 80:5000 deepquestai/deepstack:cpu
+        cd tests
+        export TEST_DATA_DIR=$PWD"/test_data"
+        export TEST_DEEPSTACK_URL="http://localhost:80"
+        cd landmarkdetection
+        python3 -m pytest
+        docker container rm -f deepstack
+
   windows-build:
     runs-on: windows-2019
     steps:

Dockerfile.cpu

@@ -42,6 +42,7 @@ COPY ./sharedfiles/facerec-high.model /app/sharedfiles/facerec-high.model
 COPY ./sharedfiles/scene.pt /app/sharedfiles/scene.pt
 COPY ./sharedfiles/categories_places365.txt /app/sharedfiles/categories_places365.txt
 COPY ./sharedfiles/bebygan_x4.pth /app/sharedfiles/bebygan_x4.pth
+COPY ./sharedfiles/facelandmark.pth /app/sharedfiles/facelandmark.pth
 
 RUN mkdir /app/server
 COPY ./server /app/server

intelligencelayer/shared/facelandmark/config.py

@@ -0,0 +1,18 @@
+from easydict import EasyDict as edict
+
+class Config:
+    MODEL=edict()
+    MODEL.SCALE=4
+    MODEL.IN_CHANNEL=3
+    MODEL.OUT_CHANNEL=3
+    MODEL.N_FEATURE=64
+    MODEL.N_BLOCK=6
+    MODEL.DEVICE='cuda'
+
+
+
+
+
+
+
+config=Config()

intelligencelayer/shared/facelandmark/network.py

@@ -0,0 +1,210 @@
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torch.optim as optim
+from torch.autograd import Variable
+from torchvision import transforms
+
+
+
+''' Creating the Xception Net Model'''
+
+''' Depthwise Separable Convolution'''
+
+class DepthewiseSeperableConv2d(nn.Module):
+    def __init__(self, input_channels, output_channels, kernel_size, **kwargs):
+        super(DepthewiseSeperableConv2d, self).__init__()
+
+        self.depthwise = nn.Conv2d(input_channels, input_channels, kernel_size, groups = input_channels, bias = False, **kwargs)
+        self.pointwise = nn.Conv2d(input_channels, output_channels, 1, bias = False)
+
+    def forward(self, x):
+        x = self.depthwise(x)
+        x = self.pointwise(x)
+
+        return x
+
+
+
+''' Entry block'''
+
+
+class EntryBlock(nn.Module):
+    def __init__(self):
+        super(EntryBlock, self).__init__()
+
+        self.conv1 = nn.Sequential(
+            nn.Conv2d(1, 32, 3, padding = 1, bias = False),
+            nn.BatchNorm2d(32),
+            nn.LeakyReLU(0.2)
+        )
+
+        self.conv2 = nn.Sequential(
+            nn.Conv2d(32, 64, 3, padding = 1, bias = False),
+            nn.BatchNorm2d(64),
+            nn.LeakyReLU(0.2)
+        )
+
+        self.conv3_residual = nn.Sequential(
+            DepthewiseSeperableConv2d(64, 64, 3, padding = 1),
+            nn.BatchNorm2d(64),
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(64, 128, 3, padding = 1),
+            nn.BatchNorm2d(128),
+            nn.MaxPool2d(3, stride = 2, padding = 1),
+        )
+
+        self.conv3_direct = nn.Sequential(
+            nn.Conv2d(64, 128, 1, stride = 2),
+            nn.BatchNorm2d(128),
+        )
+
+        self.conv4_residual = nn.Sequential(
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(128, 128, 3, padding = 1),
+            nn.BatchNorm2d(128),
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(128, 256, 3, padding = 1),
+            nn.BatchNorm2d(256),
+            nn.MaxPool2d(3, stride = 2, padding = 1)
+        )
+
+        self.conv4_direct = nn.Sequential(
+            nn.Conv2d(128, 256, 1, stride = 2),
+            nn.BatchNorm2d(256),
+        )
+
+    def forward(self, x):
+        x = self.conv1(x)
+        x = self.conv2(x)
+
+        residual = self.conv3_residual(x)
+        direct = self.conv3_direct(x)
+        x = residual + direct
+
+        residual = self.conv4_residual(x)
+        direct = self.conv4_direct(x)
+        x = residual + direct
+
+        return x
+
+''' Middle basic Block'''
+class MiddleBasicBlock(nn.Module):
+    def __init__(self):
+        super(MiddleBasicBlock, self).__init__()
+
+        self.conv1 = nn.Sequential(
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(256, 256, 3, padding = 1),
+            nn.BatchNorm2d(256)
+        )
+        self.conv2 = nn.Sequential(
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(256, 256, 3, padding = 1),
+            nn.BatchNorm2d(256)
+        )
+        self.conv3 = nn.Sequential(
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(256, 256, 3, padding = 1),
+            nn.BatchNorm2d(256)
+        )
+
+    def forward(self, x):
+        residual = self.conv1(x)
+        residual = self.conv2(residual)
+        residual = self.conv3(residual)
+
+        return x + residual
+
+
+class MiddleBlock(nn.Module):
+    def __init__(self, num_blocks):
+        super().__init__()
+
+        self.block = nn.Sequential(*[MiddleBasicBlock() for _ in range(num_blocks)])
+
+    def forward(self, x):
+        x = self.block(x)
+
+        return x
+
+''' Exit Block'''
+
+class ExitBlock(nn.Module):
+    def __init__(self):
+        super(ExitBlock, self).__init__()
+
+        self.residual = nn.Sequential(
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(256, 256, 3, padding = 1),
+            nn.BatchNorm2d(256),
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(256, 512, 3, padding = 1),
+            nn.BatchNorm2d(512),
+            nn.MaxPool2d(3, stride = 2, padding = 1)
+        )
+
+        self.direct = nn.Sequential(
+            nn.Conv2d(256, 512, 1, stride = 2),
+            nn.BatchNorm2d(512)
+        )
+
+        self.conv = nn.Sequential(
+            DepthewiseSeperableConv2d(512, 512, 3, padding = 1),
+            nn.BatchNorm2d(512),
+            nn.LeakyReLU(0.2),
+            DepthewiseSeperableConv2d(512, 1024, 3, padding = 1),
+            nn.BatchNorm2d(1024),
+            nn.LeakyReLU(0.2)
+        )
+
+        self.dropout = nn.Dropout(0.3)
+
+        self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
+
+    def forward(self, x):
+        direct = self.direct(x)
+        residual = self.residual(x)
+        x = direct + residual
+
+        x = self.conv(x)
+        x = self.avgpool(x)
+        x = self.dropout(x)
+
+        return x
+
+''' Xception Model Final'''
+class XceptionNet(nn.Module):
+    def __init__(self, num_middle_blocks = 6):
+        super(XceptionNet, self).__init__()
+
+        self.entry_block = EntryBlock()
+        self.middel_block = MiddleBlock(num_middle_blocks)
+        self.exit_block = ExitBlock()
+
+        self.fc = nn.Linear(1024, 136)
+
+    def forward(self, x):
+        x = self.entry_block(x)
+        x = self.middel_block(x)
+        x = self.exit_block(x)
+
+        x = x.view(x.size(0), -1)
+
+        x = self.fc(x)
+
+        return x
+
+
+
+
+
+
+
+
+
+
+
+
+

intelligencelayer/shared/facelandmark/preprocess.py

@@ -0,0 +1,126 @@
+import os
+import numpy as np
+from PIL import Image
+import cv2
+import matplotlib.pyplot as plt
+#from skimage import io, transform
+#from tqdm.auto import tqdm
+
+import torch
+#import torchvision
+#import torch.nn as nn
+#import torch.optim as optim
+import random
+from math import *
+import imutils
+
+import xml.etree.ElementTree as ET
+
+from torch.utils.data import Dataset, DataLoader, random_split
+from torchvision import transforms, utils
+from torchsummary import summary
+import torchvision.transforms.functional as TF
+from __future__ import print_function, division
+import pandas as pd
+import time
+import copy
+
+
+
+''' Creating transforms'''
+
+class Transforms():
+    def __init__(self):
+        pass
+
+    def rotate(self, image, landmarks, angle):
+        angle = random.uniform(-angle, +angle)
+
+        transformation_matrix = torch.tensor([
+            [+cos(radians(angle)), -sin(radians(angle))], 
+            [+sin(radians(angle)), +cos(radians(angle))]
+        ])
+
+        image = imutils.rotate(np.array(image), angle)
+
+        landmarks = landmarks - 0.5
+        new_landmarks = np.matmul(landmarks, transformation_matrix)
+        new_landmarks = new_landmarks + 0.5
+        return Image.fromarray(image), new_landmarks
+
+    def resize(self, image, landmarks, img_size):
+        image = TF.resize(image, img_size)
+        return image, landmarks
+
+    def color_jitter(self, image, landmarks):
+        color_jitter = transforms.ColorJitter(brightness=0.3, 
+                                              contrast=0.3,
+                                              saturation=0.3, 
+                                              hue=0.1)
+        image = color_jitter(image)
+        return image, landmarks
+
+    def crop_face(self, image, landmarks, crops):
+        left = int(crops['left'])
+        top = int(crops['top'])
+        width = int(crops['width'])
+        height = int(crops['height'])
+
+        image = TF.crop(image, top, left, height, width)
+
+        img_shape = np.array(image).shape
+        landmarks = torch.tensor(landmarks) - torch.tensor([[left, top]])
+        landmarks = landmarks / torch.tensor([img_shape[1], img_shape[0]])
+        return image, landmarks
+
+    def __call__(self, image, landmarks, crops):
+        image = Image.fromarray(image)
+        image, landmarks = self.crop_face(image, landmarks, crops)
+        image, landmarks = self.resize(image, landmarks, (224, 224))
+        image, landmarks = self.color_jitter(image, landmarks)
+        image, landmarks = self.rotate(image, landmarks, angle=10)
+
+        image = TF.to_tensor(image)
+        image = TF.normalize(image, [0.5], [0.5])
+        return image, landmarks
+
+''' Create face landmark dataset'''
+class FaceLandmarkDataset(Dataset):
+    def __init__(self, transform=None):
+        tree=ET.parse('path/to/annotations.xml')
+        root=tree.getroot()
+
+        self.image_filenames=[]
+        self.landmarks=[]
+        self.crops=[]
+        self.transform=transform
+        self.root_dir='path/to/images/'
+
+        for filename in root[2]:
+            self.image_filenames.append(os.path.join(self.root_dir, filename.attrib['file']))
+            self.crops.append(filename[0].attrib)
+
+            landmark=[]
+
+            for num in range(68):
+                x_coord=int(filename[0][num].attrib['x'])
+                y_coord=int(filename[0][num].attrib['y'])
+
+                landmark.append([x_coord, y_coord])
+            self.landmarks.append(landmark)
+
+        self.landmarks=np.array(self.landmarks).astype('float32')
+
+        assert len(self.image_filenames)==len(self.landmarks)
+    def __len__(self):
+        return len(self.image_filenames)
+
+    def __getitem__(self, idx):
+        image=cv2.imread(self.image_filenames[idx],0)
+        landmarks=self.landmarks[idx]
+
+        if self.transform:
+            image, landmarks=self.transform(image, landmarks, self.crops[idx])
+
+        landmarks=landmarks - 0.5
+        return image, landmarks