<jupyter_start><jupyter_text>------------------------- Author: <NAME> Date: 2019 August 14-------------------------  Import<jupyter_code>import os
from shutil import copyfile, move
from PIL import Image
import six.moves.urllib as urllib
import tarfile
import tensorflow as tf

from ObjectDetection import *

from utils import dataset_util<jupyter_output><empty_output><jupyter_text>Define functions and classes Build labelmap fileLabelmap files provide the associated name object to an index.<jupyter_code>########################################################################################
def newLabelmapItem(index, name):

    item  = "item {\n"
    item += "    id: {}\n".format(index)
    item += "    name: '{}'\n".format(name)
    item += "}\n"

    return item

############################################################################################
class Labelmap:
    
    ########################################################################################
    def __init__(self, filename_src, filename_dst=None):
        
        self.filename_src = filename_src
        
        if filename_dst == None:
            filename_dst = filename_src.replace(".txt", ".pbtxt")
        self.filename_dst = filename_dst
        
    
    ########################################################################################
    def GetCategories(self):

        with open(self.filename_src, "r") as file:
            self.categories = file.read().split('\n')
            self.categories.remove("")

        return self.categories
    
    ########################################################################################
    def Build(self):

        with open(self.filename_dst, "w") as file:
            
            for i, category in enumerate(self.GetCategories()):
                item = newLabelmapItem(i+1, category)
                file.write(item)<jupyter_output><empty_output><jupyter_text>Exporting data to TFRecordTFRecord is a file format where all data are stored including both images and annotations. Get annotations from a directory<jupyter_code>########################################################################################
class Annotation(dict):
    
    ####################################################################################
    def __init__(self, filename, classes):
        
        self["filename"] = filename
        self.classes = classes
        
        
    ####################################################################################
    def SetArrays(self):
        
        self["indexes"] = []
        self["classes"] = []
        self["xmins"] = []
        self["ymins"] = []
        self["xmaxs"] = []
        self["ymaxs"] = []
        
    
    ####################################################################################
    def AppendLine(self, line):
        
        index, center_x, center_y, width, height = line.split(" ")
    
        self["indexes"].append(int(index)+1)
        self["classes"].append(self.classes[int(index)].encode())
        self["xmins"].append(float(center_x)-float(width)/2)
        self["ymins"].append(float(center_y)-float(height)/2)
        self["xmaxs"].append(float(center_x)+float(width)/2)
        self["ymaxs"].append(float(center_y)+float(height)/2)
        
        
    ####################################################################################
    def Build(self):
        
        self["image_name"] = self["filename"].replace(".txt",".jpg")
        self["image_format"] = b"jpg"
        self["width"], self["height"] = Image.open(self["image_name"]).size
        
        with open(self["image_name"], "rb") as f:
            self["encoded_jpg"] = f.read()
        
        with open(self["filename"], "r") as file:
            lines = filter(None, file.read().split('\n'))
            
        self.SetArrays()
        for line in lines:
            self.AppendLine(line)
    


########################################################################################
def IsAnnotationFile(filename):
    
    if filename.endswith(".txt") and filename != "classes.txt":
        return True
    else:
        return False
    
    
########################################################################################
def GetAnnotationFilenames(path):
    
    return [os.path.abspath(os.path.join(path, filenames)) for filenames in os.listdir(path) if IsAnnotationFile(filenames)]


########################################################################################
def BuildAnnotationList(path, classes):
    
    annotationList = []
    for filename in GetAnnotationFilenames(path):
        annotation = Annotation(filename, classes)
        annotation.Build()
        annotationList.append(annotation)
        
    return annotationList<jupyter_output><empty_output><jupyter_text>Writing .record (TFRecord) file<jupyter_code>########################################################################################
def WriteTFRecord(filename, annotations):
    
    with tf.python_io.TFRecordWriter(filename) as writer:
        
        for annotation in annotations:

            features = tf.train.Features(feature={
                'image/height': dataset_util.int64_feature(annotation["height"]),
                'image/width': dataset_util.int64_feature(annotation["width"]),
                
                'image/filename': dataset_util.bytes_feature(annotation["filename"].encode()),
                'image/source_id': dataset_util.bytes_feature(annotation["filename"].encode()),
                'image/encoded': dataset_util.bytes_feature(annotation["encoded_jpg"]),
                'image/format': dataset_util.bytes_feature(annotation["image_format"]),

                'image/object/bbox/xmin': dataset_util.float_list_feature(annotation["xmins"]),
                'image/object/bbox/xmax': dataset_util.float_list_feature(annotation["xmaxs"]),
                'image/object/bbox/ymin': dataset_util.float_list_feature(annotation["ymins"]),
                'image/object/bbox/ymax': dataset_util.float_list_feature(annotation["ymaxs"]),
                'image/object/class/text': dataset_util.bytes_list_feature(annotation["classes"]),
                'image/object/class/label': dataset_util.int64_list_feature(annotation["indexes"]),
            })

            writer.write(tf.train.Example(features=features).SerializeToString())<jupyter_output><empty_output><jupyter_text>Download modelIf it was not already used, the object detector directory will be downloaded on the tensorflow download base. The directory provide a config file to use for the training.<jupyter_code>########################################################################################
def extract(model_name, target_dir):
    
    MODEL_ZIPFILE = model_name + '.tar.gz'
    MODEL_PATH = os.path.join("cnn", MODEL_ZIPFILE)
    
    tar_file = tarfile.open(MODEL_PATH)
    for file in tar_file.getmembers():
        tar_file.extract(file, target_dir)
        

########################################################################################
def download(model_name):
    
    DOWNLOAD_BASE = 'http://download.tensorflow.org/models/object_detection/'
    MODEL_ZIPFILE = model_name + '.tar.gz'
    MODEL_URL = DOWNLOAD_BASE + MODEL_ZIPFILE
    
    opener = urllib.request.URLopener()
    opener.retrieve(MODEL_URL, MODEL_ZIPFILE)
    opener.cleanup()
    opener.close()
    

########################################################################################
def need_to_download(model_name):
    
    MODEL_ZIPFILE = model_name + '.tar.gz'
    
    if MODEL_ZIPFILE not in os.listdir("cnn"):
        return True
    else:
        return False
        

########################################################################################
def move_to_cnn_dir(model_name):

    SRC = model_name + '.tar.gz'
    DST = os.path.join("cnn", model_name + '.tar.gz')
    move(SRC, DST)
    
    
########################################################################################
def download_if_needed(model_name):
    
    if need_to_download(model_name):
        download(model_name)
        move_to_cnn_dir(model_name)<jupyter_output><empty_output><jupyter_text>Complete the .config fileThe .config file set the trainning by specify parameters like loss function or number of evaluation to do. Path to train and test tfrecord file must be completed.<jupyter_code>########################################################################################
def complete_val(lines, param, val, start):
    
    for i in range(start, len(lines)):
        
        if param in lines[i] :
            lines[i] = param + ': ' + val
            return i+1
        
    return i+1

########################################################################################
def complete_values(path, parameters):
    
    # read the file to get its lines
    with open(path) as file:
        lines = file.read().split("\n")
    
    # loop over lines for setting parameters
    i = 0
    for param, value in parameters:
        
        if type(value) == str:
            val = '"{}"'.format(value.replace("\\", "/")) # backslash must be replace in config file and " inserted
        elif type(value) == bool:
            val = str(value).lower()
        else:
            val = str(value)
        
        i = complete_val(lines, param, val, i)
        if i == len(lines): break
            
    # update the file
    with open(path, "w") as file:
        for line in lines:
            file.write(line+"\n")<jupyter_output><empty_output><jupyter_text>Add scripts<jupyter_code>########################################################################################
def add_train_module(path):
    
    TRAIN_SCRIPT_SRC = os.path.join(object_detection_path(), "legacy", "train.py")
    TRAIN_SCRIPT_DST = os.path.join(path, "train.py")
    
    with open(TRAIN_SCRIPT_DST, "w") as dst:
        dst.write("from ObjectDetection import *\n")
        
        # open read only
        with open(TRAIN_SCRIPT_SRC, "r") as src:
            dst.write(src.read())
            

########################################################################################
def gen_train_bat(work_dir, training_dir, config_path):
    
    BAT_FILE = os.path.join(work_dir, "train.bat")
    TRAIN_SCRIPT_PATH = os.path.join(object_detection_path(), "legacy", "train.py")

    CMD = "python train.py --logtostderr --train_dir={} --pipeline_config_path={}".format(training_dir, config_path)
    with open(BAT_FILE, "w") as file: file.write(CMD)<jupyter_output><empty_output><jupyter_text>Main Variables<jupyter_code>WORK_DIR = "WORK_DIR"

TRAIN_DIR = os.path.join(WORK_DIR, "train")
TEST_DIR = os.path.join(WORK_DIR, "test")

TRAINING_DIR = os.path.join(WORK_DIR, "training")

MODEL_NAME = "MODEL_NAME"
CONFIG_FILE = "CONFIG_FILE"<jupyter_output><empty_output><jupyter_text>Create training directory<jupyter_code>try:
    os.mkdir(TRAINING_DIR)
except:
    pass<jupyter_output><empty_output><jupyter_text>Labelmaps<jupyter_code># train filenames
TRAIN_CLASSES = os.path.join(TRAIN_DIR, "classes.txt")
TRAIN_LABELMAP = os.path.join(TRAINING_DIR, "train.pbtxt")

# test filenames
TEST_CLASSES = os.path.join(TEST_DIR, "classes.txt")
TEST_LABELMAP = os.path.join(TRAINING_DIR, "test.pbtxt")

# build labelmaps
train_labelmap = Labelmap(TRAIN_CLASSES, TRAIN_LABELMAP)
train_labelmap.Build()

test_labelmap = Labelmap(TEST_CLASSES, TEST_LABELMAP)
test_labelmap.Build()<jupyter_output><empty_output><jupyter_text>TFRecord<jupyter_code>TRAIN_RECORD = os.path.join(TRAINING_DIR, "train.record")
TEST_RECORD = os.path.join(TRAINING_DIR, "test.record")

train_annotations = BuildAnnotationList(TRAIN_DIR, train_labelmap.GetCategories())
test_annotations = BuildAnnotationList(TEST_DIR, test_labelmap.GetCategories())
    
WriteTFRecord(TRAIN_RECORD, train_annotations)
WriteTFRecord(TEST_RECORD, test_annotations)<jupyter_output><empty_output><jupyter_text>Model importation and config Pretrained model<jupyter_code>download_if_needed(MODEL_NAME)
extract(MODEL_NAME, TRAINING_DIR)<jupyter_output><empty_output><jupyter_text>Config file<jupyter_code>CONFIG_SRC = os.path.join(object_detection_path(), "samples/configs", CONFIG_FILE)
CONFIG_DST = os.path.join(TRAINING_DIR, CONFIG_FILE)

copyfile(CONFIG_SRC, CONFIG_DST)<jupyter_output><empty_output><jupyter_text>Checkpoint file<jupyter_code>CHECK_FILE = os.path.join(TRAINING_DIR, MODEL_NAME, "model.ckpt")

complete_values(CONFIG_DST, [
                ("num_classes", len(train_labelmap.categories)),
                ("fine_tune_checkpoint", CHECK_FILE),
                ("input_path", TRAIN_RECORD),
                ("label_map_path", TRAIN_LABELMAP),
                ("num_examples", 150),
                ("input_path", TEST_RECORD),
                ("label_map_path", TEST_LABELMAP),
                ("shuffle", True)
                ])<jupyter_output><empty_output><jupyter_text>Train files (.py and .bat)<jupyter_code># copy object detection initializer module
OBJECT_DETECTION_DST = os.path.join(WORK_DIR, "ObjectDetection.py")
copyfile("ObjectDetection.py", OBJECT_DETECTION_DST)

add_train_module(WORK_DIR)
gen_train_bat(WORK_DIR, TRAINING_DIR, CONFIG_DST)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.25598194130925506, 'fraction_numerical': 0.0018058690744920992, 'mean_word_length': 3.8401310997815004, 'pattern_counts': {'":': 1, '<': 62, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '407433', 'license': 'Apache-2.0', 'n_tokens_mistral': 3862, 'n_tokens_neox': 3695, 'n_words': 661, 'path': 'PC/Tensorflow/TrainingConfigurator.ipynb', 'repo_name': 'limvi-licef/NATAssistanceAR', 'size': '13297'}