<jupyter_start><jupyter_text>Week 6 -- Preparing Dataset and Writing an Face Recognition Preprocessor   Required Packages conda install tensorflow\conda install jupyterlab\conda install pip\conda install opencv\conda install matplotlib\conda config --add channels conda-forge\conda install imgaug\pip install requests\  Prepare Dataset - face recognition dataset link: http://vis-www.cs.umass.edu/lfw/  Information:13233 images5749 people1680 people with two or more imagesImage augmentation documentation: https://imgaug.readthedocs.io/en/latest/index.html  code<jupyter_code># import packages
import requests
import tarfile
import os
import numpy as np
import cv2
import keras
import imgaug as ia
from imgaug import augmenters as iaa
from typing import List
print("finish importing!")

# global variable

# install dataset
# note: it could take more than 5 minute to download. If the circle on the upper right of this window is black, DO NOT stop this process untill see
# the output "Downloading finish"
data_url = "http://vis-www.cs.umass.edu/lfw/lfw.tgz"
training_pairs_url = "http://vis-www.cs.umass.edu/lfw/pairsDevTrain.txt"
testing_pairs_url = "http://vis-www.cs.umass.edu/lfw/pairsDevTest.txt"
anchor_lib_url = "https://raw.github.com/tensorflow/models/blob/master/research/object_detection/anchor_generators/grid_anchor_generator.py"
zip_dataset_name = "lfw.tgz"
training_pairs_name = "pairsDevTrain.txt"
testing_pairs_name = "pairsDevTest.txt"
data_dir_name = 'data'
if not os.path.isdir(data_dir_name):
    os.mkdir(data_dir_name)
    print("Downloading start")
    r = requests.get(data_url,stream = True)
    with open(os.path.join(data_dir_name, zip_dataset_name), 'wb') as f:
        f.write(r.raw.read())
    print("Downloading finish")
# download pairsDevTrain.txt and parisDevTest.txt
if not os.path.isfile(training_pairs_name):
    r = requests.get(training_pairs_url)
    with open(training_pairs_name, "wb") as file:
        file.write(r.content)
    print(f"{training_pairs_name} Downloaded")
if not os.path.isfile(testing_pairs_name):
    r = requests.get(testing_pairs_url)
    with open(testing_pairs_name, "wb") as file:
        file.write(r.content)
    print(f"{testing_pairs_name} Downloaded")
#unzip the package
if not len([os.listdir('data')]) > 5700:
    tar = tarfile.open("data/lfw.tgz", "r")
    tar.extractall(data_dir_name)
    tar.close
    print("finish extracting")<jupyter_output>finish extracting<jupyter_text>Please read closely to the README before going foreward<jupyter_code># http://vis-www.cs.umass.edu/lfw/README.txt
# this function return the path to the image with name NAME_ID.jpg
def construct_dir_from_name(name, id):
    path_name = os.path.join(data_dir_name,"lfw",name)
    return os.path.join(path_name,name) + "_" + f"{id}".zfill(4) + ".jpg"
class DataGenerator (keras.utils.Sequence):
    def __init__(self, txt_file_name, batch_size, img_aug = True, anchor_shape = (96,96), channel = 3, img_aug = False, shuffle = True):
        print("initializaing")
        self.batch_size = batch_size
        self.img_aug = img_aug
        self.anchor_shape = anchor_shape
        self.shuffle = shuffle
        self.txt_file_name = txt_file_name
        self.anchar_arr = []
        self.channel = channel
        self.img_aug = img_aug
        self.negative_dir_arr:List[str] = []
        self.positive_dir_arr:List[str] = []
        with open(txt_file_name) as f:
            for line in f:
                line = line.split()
                if len(line) == 3:
                    self.positive_dir_arr.append([construct_dir_from_name(line[0],line[1]), construct_dir_from_name(line[0], line[2])])
                elif len(line) == 4:
                    self.negative_dir_arr.append([construct_dir_from_name(line[0],line[1]),construct_dir_from_name(line[2], line[3])])
        self.on_epoch_end()

    # if self.shuffle == true, shuffle the positive and negative indexes
    def on_epoch_end(self):
        # udpate indexs after each epoch
        self.pos_indexes = np.arange(len(self.positive_dir_arr))
        self.neg_indexes = np.arange(len(self.negative_dir_arr))
        if self.shuffle == True:
            np.random.shuffle(self.pos_indexes)
            np.random.shuffle(self.neg_indexes)

    def __len__(self):
        # Denotes the number of batches per epoch
        return int(np.floor(len(self.pos_indexes) / self.batch_size))

    # __getitem__ is a system-defined function. array[i] is doing same operation as
    # __getitem__. Here we actually override it -- we feed in an index and return
    # a batch of image. See how we call this function in "generate_data" function:
    # "yield self[counter]" is the same thing as "yield self.__getitem_(counter)"
    def __getitem__(self, index):
        pos_indexes = self.pos_indexes[index*self.batch_size:(index+1)*self.batch_size]
        neg_indexes = self.neg_indexes[index*self.batch_size:(index+1)*self.batch_size]
        
        pos_list_temp = [self.positive_dir_arr[i] for i in pos_indexes]
        neg_list_temp = [self.negative_dir_arr[k] for k in neg_indexes]

        return self.__batch_generation(pos_list_temp,neg_list_temp)

    # augment img as a batch
    def images_aumentation(self, batch):
        seq = iaa.Sequential([
            iaa.Crop(px=(0, 16)), # crop images from each side by 0 to 16px (randomly chosen)
            iaa.Fliplr(0.5), # horizontally flip 50% of the images
            iaa.GaussianBlur(sigma=(0, 1.0)),
            # Strengthen or weaken the contrast in each image.
            iaa.LinearContrast((0.75, 1.5)),# blur images with a sigma of 0 to 3.0
        ])
        return seq(images=batch)
            
    
    def __batch_generation(self, pos_index_temp, neg_index_temp):
        # construct positive image (numpy) array
        pos_img_arr = []
        neg_img_arr = []
        anchor_arr = [] # todo
        for dir in pos_index_temp:
            img = cv2.imread(dir[1])
            resized_img = cv2.resize(img, self.anchor_shape)
            pos_img_arr.append(resized_img)
        for dir in neg_index_temp:
            img = cv2.imread(dir[0])
            resized_img = cv2.resize(img, self.anchor_shape)
            neg_img_arr.append(resized_img)
        if self.img_aug == True:
            augmented_pos_img = self.images_aumentation(pos_img_arr)
            augmented_neg_img = self.images_aumentation(neg_img_arr)
            # concat augmented imgaes to batch
            pos_img_arr = np.concatenate((pos_img_arr,augmented_pos_img), axis= 0)
            neg_img_arr = np.concatenate((neg_img_arr,augmented_neg_img), axis= 0)
        return anchor_arr, pos_img_arr, neg_img_arr

    # you should call this function to get a batch of imgs
    def generate_data(self):
        while(True):
            counter = 0
            yield self[counter]
            counter += 1

# for testing only
batchsize = 6
# The default option for image augmentation is False. Please notice that if you choose to make image augmention,
# it will double the batch size. For example, the batch shape after the image augmentation is (12, 96, 96, 3),
# comparing to the one without img aug (6, 96, 96, 3)
data_generator = DataGenerator(txt_file_name=training_pairs_name, batch_size = batchsize, img_aug = True);
anchor_arr, pos_img_arr, neg_img_arr = next(data_generator.generate_data())
print(np.asarray(pos_img_arr).shape)
pos_img_example = pos_img_arr[1]
neg_img_example = neg_img_arr[1]
anchor_arr_1, pos_img_arr_1, neg_img_arr_1 = next(data_generator.generate_data())
pos_img_example_1 = pos_img_arr_1[1]
neg_img_example_1 = neg_img_arr_1[1]
print("testing complete")
import matplotlib.pyplot as plt
# convert BGR to RGB
pos_img_example = pos_img_example[:, :, ::-1]
plt.imshow(pos_img_example)
neg_img_example_1 = neg_img_example_1[:, :, ::-1]
plt.imshow(neg_img_example_1)
aug_neg_example = pos_img_arr[8][:, :, ::-1]
plt.imshow(aug_neg_example)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10182504719949653, 'fraction_numerical': 0.012712397734424166, 'mean_word_length': 3.4994337485843716, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 5, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '638317', 'license': 'MIT', 'n_tokens_mistral': 2733, 'n_tokens_neox': 2629, 'n_words': 618, 'path': 'Week6.ipynb', 'repo_name': 'syKevinPeng/fireweek6', 'size': '7945'}