Arabic_OCR.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "colab_type": "text",
        "id": "view-in-github"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/msfasha/Arabic-Deep-Learning-OCR/blob/master/Arabic_OCR.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "is0mDGfImQh6"
      },
      "source": [
        "Usage:<br>\n",
        "1- goto the dataset repository:<br>\n",
        "https://drive.google.com/drive/u/2/folders/1mRefmN4Yzy60Uh7z3B6cllyyOXaxQrgg\n",
        " \n",
        "and select one of the datasets e.g. 1_nice_60000_rows\n",
        "\n",
        "2- download the related files, for example:<br>\n",
        "a- 1_nice_60000_rows.bin<br>\n",
        "and<br>\n",
        "b- 1_nice_60000_rows.txt<br><br>\n",
        "\n",
        "2- upload the dataset files you downloaded from the dataset repository into your Google Colab drive.\n",
        "if you uploaded the files into Colab drive, they will be deleted once the session is over. Hence, you can upload the downloaded dataset files into your own google drive and mount your google drive in Colab.\n",
        "\n",
        "4- Set the path for the dataset files in config.py below i.e. whether in Colab drive or your own google drive e.g. /content/drive/MyDrive/\n",
        "\n",
        "5- initially, we need to train the model, so we set the \n",
        "**OPERATION_TYPE = OperationType.Training** value in config.py below to Training.\n",
        "\n",
        "6- If your are using the dataset for the first time, make sure to empty your output folder from any previously generated files (dataset split files and model files...etc).\n",
        "\n",
        "7- Set the experiment name via the experiment_name constant below (if required)\n",
        "\n",
        "8- Run the training process, the training process will stop after 5 epoches with no improvements, you can change this value in config.py below.\n",
        "\n",
        "9- After finishing the training process, you can test the model on the testing dataset by setting the **OPERATION_TYPE = OperationType.Testing**\n",
        "\n",
        "10- Finally, you can make inference on single images by setting **OPERATION_TYPE = OperationType.Infer**\n",
        "To make inference, you need to place the required word image file in the designated folder defined by the INDIVIDUAL_TEST_IMAGE_PATH constant value below.\n",
        "You also need to set the filename for the image you are trying to infer in the fnInfer variable below e.g. 0.png.\n",
        "\n",
        "You can find sample files in the dataset repository in Google drive as well as the dataset folder in GitHub (under the sample_dataset_imgages folder)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Config.py"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "%cd /content/drive/MyDrive/ArabicMultiFontsDataset/\n",
        "%ls"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Rqwen4i13iEc"
      },
      "outputs": [],
      "source": [
        "# The location/path of the uploaded dataset files (after downloading them from the dataset repository)\n",
        "# Make sure to mount your Google drive in Colab\n",
        "BASE_PATH = \"/content/drive/MyDrive/ArabicMultiFontsDataset/\"\n",
        "\n",
        "# The name of the dataset files i.e. the binary file and the labels file.\n",
        "BASE_FILENAME = \"1_nice_60000_rows\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hyjcmAyrusoI"
      },
      "outputs": [],
      "source": [
        "%tensorflow_version 1.x\n",
        "\n",
        "from __future__ import division\n",
        "from __future__ import print_function\n",
        "import os\n",
        "from enum import Enum\n",
        "import numpy as np\n",
        "import argparse\n",
        "from datetime import datetime\n",
        "import time\n",
        "import cv2\n",
        "import random\n",
        "import sys\n",
        "from shutil import Error\n",
        "import tensorflow as tf\n",
        "import editdistance\n",
        "\n",
        "class OperationType(Enum):\n",
        "    Training = 1\n",
        "    Validation = 2\n",
        "    Testing = 3\n",
        "    Infer = 4\n",
        "\n",
        "\n",
        "class DecoderType:\n",
        "    BestPath = 0\n",
        "    BeamSearch = 1\n",
        "    WordBeamSearch = 2\n",
        "\n",
        "\n",
        "# Experiment name, to be saved in the audit log.\n",
        "EXPERIMENT_NAME = \"Test Drive Training Process\"\n",
        "\n",
        "# The type of the run session, depending on this type, designated datasets will be loaded.\n",
        "# Set this value to training to run a training process using the testing dataset.\n",
        "# Set this value to testing to run a testing process using the testing dataset.\n",
        "# Set this value to infer, to make an inference for a single word image file, make sure\n",
        "# to place the required word image file in the directory defined by the INDIVIDUAL_TEST_IMAGE_PATH below, and\n",
        "# make sure to set the fnInfer to the name of the image file your are trying to infer\n",
        "OPERATION_TYPE = OperationType.Training\n",
        "\n",
        "DECODER_TYPE = DecoderType.BestPath\n",
        "\n",
        "# Use this value to regenerate the training/validation/test datasets, as well as\n",
        "# the other support files. Usually this is needed when we start the training process\n",
        "# It is not needed during the Testing process so we set it to true\n",
        "# in order to regenerate all the required files, we have to delete to old ones train/validate/test and delete\n",
        "# and then its value to true. After running the app, the files are generated and we can set it back to false unless\n",
        "# we need to generate a new set of data i.e. train/validate/test\n",
        "REGENERATE_CHARLIST_AND_CORPUS = True\n",
        "\n",
        "\n",
        "#You can modify these folder settings according to your preference\n",
        "\n",
        "#Set the path where the train, validate,test datasets are saved\n",
        "DATA_PATH = BASE_PATH \n",
        "#set the path where to save the generated tensorflow model\n",
        "MODEL_PATH = BASE_PATH\n",
        "#set the path for the autogenerated files\n",
        "OUTPUT_PATH = BASE_PATH\n",
        "\n",
        "#Set the path of the single image files that you want to recognize\n",
        "INDIVIDUAL_TEST_IMAGE_PATH = DATA_PATH\n",
        "\n",
        "BASE_IMAGES_FILE = DATA_PATH + BASE_FILENAME + \".bin\"\n",
        "BASE_LABELS_FILE = DATA_PATH + BASE_FILENAME + \".txt\"\n",
        "TRAINING_LABELS_FILE = DATA_PATH + \"TRAINING_DATA_\" + BASE_FILENAME + \".txt\"\n",
        "VALIDATION_LABELS_FILE = DATA_PATH + \\\n",
        "    \"VALIDATION_DATA_\" + BASE_FILENAME + \".txt\"\n",
        "TESTING_LABELS_FILE = DATA_PATH + \"TESTING_DATA_\" + BASE_FILENAME + \".txt\"\n",
        "fnCharList = OUTPUT_PATH + 'charList.txt'\n",
        "fnResult = OUTPUT_PATH + 'result.txt'\n",
        "\n",
        "# define the name of the word image file you want to test/infer\n",
        "# for example, you can select 0.png file from as a sample\n",
        "#https://github.com/msfasha/Arabic-Deep-Learning-OCR/tree/master/dataset/sample_files\n",
        "fnInfer = INDIVIDUAL_TEST_IMAGE_PATH + \"sample_files/\" + \"0.png\"\n",
        "\n",
        "\n",
        "fnCorpus = OUTPUT_PATH + 'corpus.txt'\n",
        "fnwordCharList = OUTPUT_PATH + 'wordCharList.txt'\n",
        "\n",
        "# Number of batches for each epoch = SAMPLES_PER_EPOCH / BATCH_SIZE\n",
        "TRAINING_SAMPLES_PER_EPOCH = 5000\n",
        "BATCH_SIZE = 100\n",
        "VALIDATIOIN_SAMPLES_PER_STEP = (int)(TRAINING_SAMPLES_PER_EPOCH * .2)\n",
        "\n",
        "\n",
        "TRAINING_DATASET_SIZE = .9\n",
        "# .5 of the remaining ==> (Total - TRAINING_DATASET_SIZE) / 2\n",
        "VALIDATION_DATASET_SPLIT_SIZE = .5\n",
        "# stop after no improvements for this number of epochs\n",
        "MAXIMUM_NONIMPROVED_EPOCHS = 5\n",
        "MAXIMUM_MODELS_TO_KEEP = 3  # usually only 1, the last one\n",
        "\n",
        "#IMAGE_SIZE = (128, 32)\n",
        "IMAGE_WIDTH = 128\n",
        "IMAGE_HEIGHT = 32\n",
        "MAX_TEXT_LENGTH = 32\n",
        "RESIZE_IMAGE = True\n",
        "CONVERT_IMAGE_TO_MONOCHROME = False\n",
        "MONOCHROME_BINARY_THRESHOLD = 127\n",
        "AUGMENT_IMAGE = False\n",
        "\n",
        "def auditLog(logStr):\n",
        "    open(fnResult, 'a').write(logStr)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DBKs5s3SmU4P"
      },
      "source": [
        "Model.py"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vfu99pHXl88m"
      },
      "outputs": [],
      "source": [
        "class Model:\n",
        "    \"minimalistic TF model for HTR\"\n",
        "\n",
        "    def __init__(self, decoderType = DecoderType.BestPath, mustRestore=False, dump=False):\n",
        "        \"init model: add CNN, RNN and CTC and initialize TF\"\n",
        "        self.dump = dump\n",
        "        self.charList = open(fnCharList, encoding=\"utf-8\").read()\n",
        "        self.decoderType = decoderType\n",
        "        self.mustRestore = mustRestore\n",
        "        self.snapID = 0\n",
        "\n",
        "        # Whether to use normalization over a batch or a population\n",
        "        self.is_train = tf.placeholder(tf.bool, name='is_train')\n",
        "\n",
        "        # input image batch\n",
        "        self.inputImgs = tf.placeholder(tf.float32, shape=(\n",
        "            None, IMAGE_WIDTH, IMAGE_HEIGHT))\n",
        "\n",
        "        # setup CNN, RNN and CTC\n",
        "        self.setup5LayersCNN()\n",
        "        self.setupRNN()\n",
        "        self.setupCTC()\n",
        "\n",
        "        # setup optimizer to train NN\n",
        "        self.batchesTrained = 0\n",
        "        self.learningRate = tf.placeholder(tf.float32, shape=[])\n",
        "        self.update_ops = tf.get_collection(tf.GraphKeys.UPDATE_OPS)\n",
        "        with tf.control_dependencies(self.update_ops):\n",
        "            self.optimizer = tf.train.RMSPropOptimizer(\n",
        "                self.learningRate).minimize(self.loss)\n",
        "\n",
        "        self.auditModelDetails()\n",
        "        # initialize TF\n",
        "        (self.sess, self.saver) = self.setupTF()\n",
        "\n",
        "    def auditModelDetails(self):\n",
        "        total_parameters = 0\n",
        "        saveString = \"Model Details\" + \"\\n\"\n",
        "        for variable in tf.trainable_variables():\n",
        "            # shape is an array of tf.Dimension\n",
        "            shape = variable.get_shape()\n",
        "            saveString = saveString + \"Shape:\" + \\\n",
        "                str(shape) + \" ,shape length:\" + str(len(shape))\n",
        "            variable_parameters = 1\n",
        "            for dim in shape:\n",
        "                variable_parameters *= dim.value\n",
        "            saveString = saveString + \" , parameters: \" + \\\n",
        "                str(variable_parameters) + \"\\n\"\n",
        "            total_parameters += variable_parameters\n",
        "\n",
        "        saveString = saveString + \"Total Parameters: \" + \\\n",
        "            str(total_parameters) + \"\\n\\n\"\n",
        "\n",
        "        print(saveString)\n",
        "        auditLog(saveString)\n",
        "\n",
        "    def setup5LayersCNN(self):\n",
        "        \"create CNN layers and return output of these layers\"\n",
        "        cnnIn4d = tf.expand_dims(input=self.inputImgs, axis=3)\n",
        "        pool = cnnIn4d  # input to first CNN layer\n",
        "\n",
        "        self.kernel1 = tf.Variable(\n",
        "            tf.truncated_normal([5, 5, 1, 32], stddev=0.1))\n",
        "        self.conv1 = tf.nn.conv2d(\n",
        "            pool, self.kernel1, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        conv_norm = tf.layers.batch_normalization(\n",
        "            self.conv1, training=self.is_train)\n",
        "        self.relu1 = tf.nn.relu(conv_norm)\n",
        "        self.pool1 = tf.nn.max_pool(\n",
        "            self.relu1, (1, 2, 2, 1), (1, 2, 2, 1), 'VALID')\n",
        "\n",
        "        kernel = tf.Variable(tf.truncated_normal([5, 5, 32, 64], stddev=0.1))\n",
        "        self.conv2 = tf.nn.conv2d(\n",
        "            self.pool1, kernel, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        conv_norm = tf.layers.batch_normalization(\n",
        "            self.conv2, training=self.is_train)\n",
        "        relu = tf.nn.relu(conv_norm)\n",
        "        pool = tf.nn.max_pool(relu, (1, 2, 2, 1), (1, 2, 2, 1), 'VALID')\n",
        "\n",
        "        kernel = tf.Variable(tf.truncated_normal([3, 3, 64, 128], stddev=0.1))\n",
        "        self.conv3 = tf.nn.conv2d(\n",
        "            pool, kernel, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        conv_norm = tf.layers.batch_normalization(\n",
        "            self.conv3, training=self.is_train)\n",
        "        relu = tf.nn.relu(conv_norm)\n",
        "        pool = tf.nn.max_pool(relu, (1, 1, 2, 1), (1, 1, 2, 1), 'VALID')\n",
        "\n",
        "        kernel = tf.Variable(tf.truncated_normal([3, 3, 128, 128], stddev=0.1))\n",
        "        self.conv4 = tf.nn.conv2d(\n",
        "            pool, kernel, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        conv_norm = tf.layers.batch_normalization(\n",
        "            self.conv4, training=self.is_train)\n",
        "        relu = tf.nn.relu(conv_norm)\n",
        "        pool = tf.nn.max_pool(relu, (1, 1, 2, 1), (1, 1, 2, 1), 'VALID')\n",
        "\n",
        "        kernel = tf.Variable(tf.truncated_normal([3, 3, 128, 256], stddev=0.1))\n",
        "        self.conv5 = tf.nn.conv2d(\n",
        "            pool, kernel, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        conv_norm = tf.layers.batch_normalization(\n",
        "            self.conv5, training=self.is_train)\n",
        "        relu = tf.nn.relu(conv_norm)\n",
        "        pool = tf.nn.max_pool(relu, (1, 1, 2, 1), (1, 1, 2, 1), 'VALID')\n",
        "\n",
        "        self.cnnOut4d = pool\n",
        "\n",
        "    def setupCNN7Layers(self):\n",
        "        \"create CNN layers and return output of these layers\"\n",
        "        cnnIn4d = tf.expand_dims(input=self.inputImgs, axis=3)\n",
        "\n",
        "        kernel1 = tf.Variable(tf.truncated_normal([3, 3, 1, 64], stddev=0.1))\n",
        "        conv1 = tf.nn.conv2d(\n",
        "            cnnIn4d, kernel1, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        pool1 = tf.nn.max_pool(conv1, (1, 2, 2, 1), (1, 2, 2, 1), 'VALID')\n",
        "\n",
        "        kernel2 = tf.Variable(tf.truncated_normal([3, 3, 64, 128], stddev=0.1))\n",
        "        conv2 = tf.nn.conv2d(\n",
        "            pool1, kernel2, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        pool2 = tf.nn.max_pool(conv2, (1, 2, 2, 1), (1, 2, 2, 1), 'VALID')\n",
        "\n",
        "        kernel3 = tf.Variable(tf.truncated_normal(\n",
        "            [3, 3, 128, 256], stddev=0.1))\n",
        "        conv3 = tf.nn.conv2d(\n",
        "            pool2, kernel3, padding='SAME', strides=(1, 1, 1, 1))\n",
        "\n",
        "        kernel4 = tf.Variable(tf.truncated_normal(\n",
        "            [3, 3, 256, 256], stddev=0.1))\n",
        "        conv4 = tf.nn.conv2d(\n",
        "            conv3, kernel4, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        pool3 = tf.nn.max_pool(conv4, (1, 1, 2, 1), (1, 1, 2, 1), 'VALID')\n",
        "\n",
        "        kernel5 = tf.Variable(tf.truncated_normal(\n",
        "            [3, 3, 256, 512], stddev=0.1))\n",
        "        conv5 = tf.nn.conv2d(\n",
        "            pool3, kernel5, padding='SAME', strides=(1, 1, 1, 1))\n",
        "        batch_norm1 = tf.layers.batch_normalization(\n",
        "            conv4, training=self.is_train)\n",
        "\n",
        "        kernel6 = tf.Variable(tf.truncated_normal(\n",
        "            [3, 3, 512, 512], stddev=0.1))\n",
        "        conv6 = tf.nn.conv2d(batch_norm1, kernel6,\n",
        "                             padding='SAME', strides=(1, 1, 1, 1))\n",
        "        batch_norm2 = tf.layers.batch_normalization(\n",
        "            conv6, training=self.is_train)\n",
        "        pool4 = tf.nn.max_pool(batch_norm2, (1, 1, 2, 1),\n",
        "                               (1, 1, 2, 1), 'VALID')\n",
        "\n",
        "        kernel7 = tf.Variable(tf.truncated_normal(\n",
        "            [2, 2, 512, 512], stddev=0.1))\n",
        "        conv7 = tf.nn.conv2d(batch_norm1, kernel7,\n",
        "                             padding='SAME', strides=(1, 1, 1, 1))\n",
        "\n",
        "        self.cnnOut4d = conv7\n",
        "\n",
        "    def setupRNN(self):\n",
        "        \"create RNN layers and return output of these layers\"\n",
        "        rnnIn3d = tf.squeeze(self.cnnOut4d, axis=[2])\n",
        "\n",
        "        # basic cells which is used to build RNN\n",
        "        numHidden = 256\n",
        "        cells = [tf.contrib.rnn.LSTMCell(\n",
        "            num_units=numHidden, state_is_tuple=True) for _ in range(2)]  # 2 layers\n",
        "\n",
        "        # stack basic cells\n",
        "        stacked = tf.contrib.rnn.MultiRNNCell(cells, state_is_tuple=True)\n",
        "\n",
        "        # bidirectional RNN\n",
        "        # BxTxF -> BxTx2H\n",
        "        ((fw, bw), _) = tf.nn.bidirectional_dynamic_rnn(cell_fw=stacked, cell_bw=stacked, inputs=rnnIn3d,\n",
        "                                                        dtype=rnnIn3d.dtype)\n",
        "\n",
        "        # BxTxH + BxTxH -> BxTx2H -> BxTx1X2H\n",
        "        concat = tf.expand_dims(tf.concat([fw, bw], 2), 2)\n",
        "\n",
        "        # project output to chars (including blank): BxTx1x2H -> BxTx1xC -> BxTxC\n",
        "        kernel = tf.Variable(tf.truncated_normal(\n",
        "            [1, 1, numHidden * 2, len(self.charList) + 1], stddev=0.1))\n",
        "        self.rnnOut3d = tf.squeeze(tf.nn.atrous_conv2d(\n",
        "            value=concat, filters=kernel, rate=1, padding='SAME'), axis=[2])\n",
        "\n",
        "    def setupCTC(self):\n",
        "        \"create CTC loss and decoder and return them\"\n",
        "        # BxTxC -> TxBxC\n",
        "        self.ctcIn3dTBC = tf.transpose(self.rnnOut3d, [1, 0, 2])\n",
        "\n",
        "        # ground truth text as sparse tensor\n",
        "        self.gtTexts = tf.SparseTensor(tf.placeholder(tf.int64, shape=[None, 2]), tf.placeholder(tf.int32, [None]),\n",
        "                                       tf.placeholder(tf.int64, [2]))\n",
        "\n",
        "        # calc loss for batch\n",
        "        self.seqLen = tf.placeholder(tf.int32, [None])\n",
        "        self.loss = tf.reduce_mean(\n",
        "            tf.nn.ctc_loss(labels=self.gtTexts, inputs=self.ctcIn3dTBC, sequence_length=self.seqLen,\n",
        "                           ctc_merge_repeated=True))\n",
        "\n",
        "        # calc loss for each element to compute label probability\n",
        "        self.savedCtcInput = tf.placeholder(\n",
        "            tf.float32, shape=[MAX_TEXT_LENGTH, None, len(self.charList) + 1])\n",
        "        self.lossPerElement = tf.nn.ctc_loss(labels=self.gtTexts, inputs=self.savedCtcInput,\n",
        "                                             sequence_length=self.seqLen, ctc_merge_repeated=True)\n",
        "\n",
        "        # decoder: either best path decoding or beam search decoding\n",
        "        if self.decoderType == DecoderType.BestPath:\n",
        "            self.decoder = tf.nn.ctc_greedy_decoder(\n",
        "                inputs=self.ctcIn3dTBC, sequence_length=self.seqLen)\n",
        "        elif self.decoderType == DecoderType.BeamSearch:\n",
        "            self.decoder = tf.nn.ctc_beam_search_decoder(inputs=self.ctcIn3dTBC, sequence_length=self.seqLen,\n",
        "                                                         beam_width=50, merge_repeated=False)\n",
        "        elif self.decoderType == DecoderType.WordBeamSearch:\n",
        "            # import compiled word beam search operation (see https://github.com/githubharald/CTCWordBeamSearch)\n",
        "            word_beam_search_module = tf.load_op_library('TFWordBeamSearch.so')\n",
        "\n",
        "            # prepare information about language (dictionary, characters in dataset, characters forming words)\n",
        "            chars = str().join(self.charList)\n",
        "            wordChars = open(fnwordCharList).read().splitlines()[0]\n",
        "            corpus = open(fnCorpus).read()\n",
        "\n",
        "            # decode using the \"Words\" mode of word beam search\n",
        "            self.decoder = word_beam_search_module.word_beam_search(tf.nn.softmax(self.ctcIn3dTBC, dim=2), 50, 'Words',\n",
        "                                                                    0.0, corpus.encode(\n",
        "                                                                        'utf8'), chars.encode('utf8'),\n",
        "                                                                    wordChars.encode('utf8'))\n",
        "\n",
        "    def setupTF(self):\n",
        "        \"initialize TF\"\n",
        "        print('Python: ' + sys.version)\n",
        "        print('Tensorflow: ' + tf.__version__)\n",
        "\n",
        "        sess = tf.Session()  # TF session\n",
        "\n",
        "        # saver saves model to file\n",
        "        saver = tf.train.Saver(max_to_keep=MAXIMUM_MODELS_TO_KEEP)\n",
        "        modelDir = MODEL_PATH\n",
        "        latestSnapshot = tf.train.latest_checkpoint(\n",
        "            modelDir)  # is there a saved model?\n",
        "\n",
        "        # if model must be restored (for inference), there must be a snapshot\n",
        "        if self.mustRestore and not latestSnapshot:\n",
        "            raise Exception('No saved model found in: ' + modelDir)\n",
        "\n",
        "        # load saved model if available\n",
        "        if latestSnapshot:\n",
        "            print('Init with stored values from ' + latestSnapshot)\n",
        "            saver.restore(sess, latestSnapshot)\n",
        "        else:\n",
        "            print('Init with new values')\n",
        "            sess.run(tf.global_variables_initializer())\n",
        "\n",
        "        return (sess, saver)\n",
        "\n",
        "    def toSparse(self, texts):\n",
        "        \"put ground truth texts into sparse tensor for ctc_loss\"\n",
        "        indices = []\n",
        "        values = []\n",
        "        shape = [len(texts), 0]  # last entry must be max(labelList[i])\n",
        "\n",
        "        # go over all texts\n",
        "        for (batchElement, text) in enumerate(texts):\n",
        "            # convert to string of label (i.e. class-ids)\n",
        "            CharactersIndexesOflabels = [self.charList.index(c) for c in text]\n",
        "            # sparse tensor must have size of max. label-string\n",
        "            if len(CharactersIndexesOflabels) > shape[1]:\n",
        "                shape[1] = len(CharactersIndexesOflabels)\n",
        "            # put each label into sparse tensor\n",
        "            for (i, label) in enumerate(CharactersIndexesOflabels):\n",
        "                indices.append([batchElement, i])\n",
        "                values.append(label)\n",
        "\n",
        "        return (indices, values, shape)\n",
        "\n",
        "    def decoderOutputToText(self, ctcOutput, batchSize):\n",
        "        \"extract texts from output of CTC decoder\"\n",
        "\n",
        "        # contains string of labels for each batch element\n",
        "        encodedLabelStrs = [[] for i in range(batchSize)]\n",
        "\n",
        "        # word beam search: label strings terminated by blank\n",
        "        if self.decoderType == DecoderType.WordBeamSearch:\n",
        "            blank = len(self.charList)\n",
        "            for b in range(batchSize):\n",
        "                for label in ctcOutput[b]:\n",
        "                    if label == blank:\n",
        "                        break\n",
        "                    encodedLabelStrs[b].append(label)\n",
        "\n",
        "        # TF decoders: label strings are contained in sparse tensor\n",
        "        else:\n",
        "            # ctc returns tuple, first element is SparseTensor\n",
        "            decoded = ctcOutput[0][0]\n",
        "\n",
        "            # go over all indices and save mapping: batch -> values\n",
        "            idxDict = {b: [] for b in range(batchSize)}\n",
        "            for (idx, idx2d) in enumerate(decoded.indices):\n",
        "                label = decoded.values[idx]\n",
        "                batchElement = idx2d[0]  # index according to [b,t]\n",
        "                encodedLabelStrs[batchElement].append(label)\n",
        "\n",
        "        # map labels to chars for all batch elements\n",
        "        return [str().join([self.charList[c] for c in labelStr]) for labelStr in encodedLabelStrs]\n",
        "\n",
        "    def trainBatch(self, batch):\n",
        "        \"feed a batch into the NN to train it\"\n",
        "        numBatchElements = len(batch.imgs)\n",
        "        sparse = self.toSparse(batch.gtTexts)\n",
        "        rate = 0.01 if self.batchesTrained < 10 else (\n",
        "            0.001 if self.batchesTrained < 10000 else 0.0001)  # decay learning rate\n",
        "        evalList = [self.optimizer, self.loss]\n",
        "        feedDict = {self.inputImgs: batch.imgs, self.gtTexts: sparse,\n",
        "                    self.seqLen: [MAX_TEXT_LENGTH] * numBatchElements, self.learningRate: rate,\n",
        "                    self.is_train: True}\n",
        "\n",
        "        (_, lossVal) = self.sess.run(evalList, feedDict)\n",
        "\n",
        "        self.batchesTrained += 1\n",
        "        return lossVal\n",
        "\n",
        "    def dumpNNOutput(self, rnnOutput):\n",
        "        \"dump the output of the NN to CSV file(s)\"\n",
        "        dumpDir = '../dump/'\n",
        "        if not os.path.isdir(dumpDir):\n",
        "            os.mkdir(dumpDir)\n",
        "\n",
        "        # iterate over all batch elements and create a CSV file for each one\n",
        "        maxT, maxB, maxC = rnnOutput.shape\n",
        "        for b in range(maxB):\n",
        "            csv = ''\n",
        "            for t in range(maxT):\n",
        "                for c in range(maxC):\n",
        "                    csv += str(rnnOutput[t, b, c]) + ';'\n",
        "                csv += '\\n'\n",
        "            fn = dumpDir + 'rnnOutput_' + str(b) + '.csv'\n",
        "            print('Write dump of NN to file: ' + fn)\n",
        "            with open(fn, 'w') as f:\n",
        "                f.write(csv)\n",
        "\n",
        "    def inferBatch(self, batch, calcProbability=False, probabilityOfGT=False):\n",
        "        \"feed a batch into the NN to recognize the texts\"\n",
        "\n",
        "        # decode, optionally save RNN output\n",
        "        numBatchElements = len(batch.imgs)\n",
        "        evalRnnOutput = self.dump or calcProbability\n",
        "        evalList = [self.decoder] + \\\n",
        "            ([self.ctcIn3dTBC] if evalRnnOutput else [])\n",
        "        feedDict = {self.inputImgs: batch.imgs, self.seqLen: [MAX_TEXT_LENGTH] * numBatchElements,\n",
        "                    self.is_train: False}\n",
        "\n",
        "        evalRes = self.sess.run(evalList, feedDict)\n",
        "\n",
        "        decoded = evalRes[0]\n",
        "        texts = self.decoderOutputToText(decoded, numBatchElements)\n",
        "\n",
        "        # feed RNN output and recognized text into CTC loss to compute labeling probability\n",
        "        probs = None\n",
        "        if calcProbability:\n",
        "            sparse = self.toSparse(\n",
        "                batch.gtTexts) if probabilityOfGT else self.toSparse(texts)\n",
        "            ctcInput = evalRes[1]\n",
        "            evalList = self.lossPerElement\n",
        "            feedDict = {self.savedCtcInput: ctcInput, self.gtTexts: sparse,\n",
        "                        self.seqLen: [MAX_TEXT_LENGTH] * numBatchElements, self.is_train: False}\n",
        "\n",
        "            lossVals = self.sess.run(evalList, feedDict)\n",
        "\n",
        "            probs = np.exp(-lossVals)\n",
        "\n",
        "        # dump the output of the NN to CSV file(s)\n",
        "        if self.dump:\n",
        "            self.dumpNNOutput(evalRes[1])\n",
        "\n",
        "        return (texts, probs)\n",
        "\n",
        "    def save(self):\n",
        "        \"save model to file\"\n",
        "        self.snapID += 1\n",
        "        self.saver.save(self.sess, MODEL_PATH +\n",
        "                        EXPERIMENT_NAME, global_step=self.snapID)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-i5T0Evj21j4"
      },
      "source": [
        "SamplePreprosessor.py"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "aQEGXWO32yFa"
      },
      "outputs": [],
      "source": [
        "def preprocess(img):\n",
        "    \"scale image into the desired imgSize, transpose it for TF and normalize gray-values\"\n",
        "\n",
        "    # increase dataset size by applying random stretches to the images\n",
        "    if AUGMENT_IMAGE:\n",
        "        stretch = (random.random() - 0.5)  # -0.5 .. +0.5\n",
        "        # random width, but at least 1\n",
        "        wStretched = max(int(img.shape[1] * (1 + stretch)), 1)\n",
        "        # stretch horizontally by factor 0.5 .. 1.5\n",
        "        img = cv2.resize(img, (wStretched, img.shape[0]))\n",
        "\n",
        "    # create target image and copy sample image into it\n",
        "    (h, w) = img.shape\n",
        "    fx = w / IMAGE_WIDTH\n",
        "    fy = h / IMAGE_HEIGHT\n",
        "    f = max(fx, fy)\n",
        "    # scale according to f (result at least 1 and at most wt or ht)\n",
        "    newSize = (max(min(IMAGE_WIDTH, int(w / f)), 1),\n",
        "               max(min(IMAGE_HEIGHT, int(h / f)), 1))\n",
        "    img = cv2.resize(img, newSize)\n",
        "    target = np.ones([IMAGE_HEIGHT, IMAGE_WIDTH]) * 255\n",
        "    target[0:newSize[1], 0:newSize[0]] = img\n",
        "\n",
        "    # transpose for TF\n",
        "    img = cv2.transpose(target)\n",
        "\n",
        "    # normalize\n",
        "    (m, s) = cv2.meanStdDev(img)\n",
        "    m = m[0][0]\n",
        "    s = s[0][0]\n",
        "    img = img - m\n",
        "    img = img / s if s > 0 else img\n",
        "\n",
        "    return img\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AsrcMrBvw9ll"
      },
      "source": [
        "DataGenerator_BinaryFile.py"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "QCnawxzlmHXM"
      },
      "outputs": [],
      "source": [
        "class Sample:\n",
        "    \"a single sample from the dataset\"\n",
        "\n",
        "    def __init__(self, gtText, imageIdx, imageHeight, imageWidth, imageSize, imageStartPosition):\n",
        "        self.gtText = gtText\n",
        "        self.imageIdx = imageIdx\n",
        "        self.imageHeight = imageHeight\n",
        "        self.imageWidth = imageWidth\n",
        "        self.imageSize = imageSize\n",
        "        self.imageStartPosition = imageStartPosition\n",
        "\n",
        "\n",
        "class Batch:\n",
        "    \"batch containing images and ground truth texts\"\n",
        "\n",
        "    def __init__(self, gtTexts, imgs):\n",
        "        self.gtTexts = gtTexts\n",
        "        self.imgs = np.stack(imgs, axis=0)\n",
        "\n",
        "\n",
        "class DataGenerator:\n",
        "\n",
        "    def __init__(self):\n",
        "        self.binaryImageFile = open(BASE_IMAGES_FILE, \"rb\")\n",
        "        self.currIdx = 0\n",
        "        self.samples = []\n",
        "        self.trainSamples = []\n",
        "        self.validationSamples = []\n",
        "        self.testSamples = []\n",
        "\n",
        "    def LoadData(self, operationType):\n",
        "        if not os.path.isfile(TRAINING_LABELS_FILE) \\\n",
        "                or not os.path.isfile(VALIDATION_LABELS_FILE) \\\n",
        "                or not os.path.isfile(TESTING_LABELS_FILE):\n",
        "            self.createDataFiles()\n",
        "\n",
        "        if operationType == OperationType.Training:\n",
        "            self.loadDataFile(OperationType.Training)\n",
        "            self.loadDataFile(OperationType.Validation)\n",
        "        elif operationType == OperationType.Validation:\n",
        "            self.loadDataFile(OperationType.Validation)\n",
        "        elif operationType == OperationType.Testing:\n",
        "            self.loadDataFile(OperationType.Testing)\n",
        "\n",
        "    def createDataFiles(self):\n",
        "        charsSet = set()\n",
        "        wordsSet = set()\n",
        "\n",
        "        f = open(BASE_LABELS_FILE, encoding=\"utf-8\")\n",
        "        for line in f:\n",
        "            # read all samples ==> append line as is\n",
        "            self.samples.append(line)\n",
        "\n",
        "            if REGENERATE_CHARLIST_AND_CORPUS:\n",
        "                # extract unique characters from text\n",
        "                lineSplit = line.split(';')\n",
        "                gtText = lineSplit[8]\n",
        "                gtText = gtText[5:]\n",
        "                wordsSet.add(gtText)\n",
        "                charsSet = charsSet.union(set(list(gtText)))\n",
        "\n",
        "        f.close()\n",
        "\n",
        "        # create a text file that contains all the characters in the dataset\n",
        "        # this list shall used to create the CTC model\n",
        "        # There might be a problem if a previously saved model used larger data, consequently, not all\n",
        "        # the characters in the previous model will be generated and therefore RNN creation will fail\n",
        "        # note that a problem might arise when we try to open a saved model that was saved on a larger dataset\n",
        "        # conseuqnelty some represented characters might be abscent and the new model will fail to load previous one\n",
        "        # a solution for this problem is to use a static character set for the used dataset\n",
        "        # also create the corpus data file for BeamSearch (if required)\n",
        "\n",
        "        # DONT CREATE THEM UNLESS U R USING LARGER DATASET, ALREADY CREATED IN DIRECTORY\n",
        "        if REGENERATE_CHARLIST_AND_CORPUS:\n",
        "            localCharList = sorted(list(charsSet))\n",
        "            open(fnCharList, 'w',\n",
        "                 encoding=\"utf-8\").write(str().join(localCharList))\n",
        "            open(fnCorpus, 'w',\n",
        "                 encoding=\"utf-8\").write(str().join(sorted(list(wordsSet))))\n",
        "\n",
        "        # first of all, make sure to randomly shuffle the main lables file\n",
        "        # random.shuffle(self.samples)\n",
        "\n",
        "        # split into training, validation, testing\n",
        "        lenOfAllSamples = len(self.samples)\n",
        "        lenOfTrainSamples = int(TRAINING_DATASET_SIZE * lenOfAllSamples)\n",
        "        lenOfTrainingAndValidationSamples = lenOfAllSamples - lenOfTrainSamples\n",
        "        lenOfValidationSamples = int(\n",
        "            VALIDATION_DATASET_SPLIT_SIZE * lenOfTrainingAndValidationSamples)\n",
        "\n",
        "        with open(TRAINING_LABELS_FILE, 'w', encoding=\"utf-8\") as f:\n",
        "            for item in self.samples[:lenOfTrainSamples]:\n",
        "                f.write(item)\n",
        "\n",
        "        with open(VALIDATION_LABELS_FILE, 'w', encoding=\"utf-8\") as f:\n",
        "            for item in self.samples[lenOfTrainSamples:lenOfTrainSamples + lenOfValidationSamples]:\n",
        "                f.write(item)\n",
        "\n",
        "        with open(TESTING_LABELS_FILE, 'w', encoding=\"utf-8\") as f:\n",
        "            for item in self.samples[lenOfTrainSamples + lenOfValidationSamples:]:\n",
        "                f.write(item)\n",
        "\n",
        "        self.samples = []\n",
        "\n",
        "    def loadDataFile(self, operationType):\n",
        "        if operationType == OperationType.Training:\n",
        "            fileName = TRAINING_LABELS_FILE\n",
        "        elif operationType == OperationType.Validation:\n",
        "            fileName = VALIDATION_LABELS_FILE\n",
        "        elif operationType == OperationType.Testing:\n",
        "            fileName = TESTING_LABELS_FILE\n",
        "\n",
        "        f = open(fileName, encoding=\"utf-8\")\n",
        "        for line in f:\n",
        "            lineSplit = line.split(';')\n",
        "\n",
        "            imgIdx = lineSplit[0]\n",
        "            imgIdx = imgIdx[10:]\n",
        "\n",
        "            imgStartPosition = lineSplit[1]\n",
        "            imgStartPosition = int(imgStartPosition[15:])\n",
        "\n",
        "            imgHeight = lineSplit[2]\n",
        "            imgHeight = int(imgHeight[13:])\n",
        "            imgWidth = lineSplit[3]\n",
        "            imgWidth = int(imgWidth[12:])\n",
        "            imgSize = imgHeight * imgWidth\n",
        "\n",
        "            gtText = lineSplit[8]\n",
        "            gtText = gtText[5:]\n",
        "            #gtText = self.truncateLabel(' '.join(gtText), MAX_TEXT_LENGTH)\n",
        "\n",
        "            # put sample into list\n",
        "            if operationType == OperationType.Training:\n",
        "                self.trainSamples.append(\n",
        "                    Sample(gtText, imgIdx, imgHeight, imgWidth, imgSize, imgStartPosition))\n",
        "            elif operationType == OperationType.Validation:\n",
        "                self.validationSamples.append(\n",
        "                    Sample(gtText, imgIdx, imgHeight, imgWidth, imgSize, imgStartPosition))\n",
        "            elif operationType == OperationType.Testing:\n",
        "                self.testSamples.append(\n",
        "                    Sample(gtText, imgIdx, imgHeight, imgWidth, imgSize, imgStartPosition))\n",
        "\n",
        "    def truncateLabel(self, text, maxTextLen):\n",
        "        # ctc_loss can't compute loss if it cannot find a mapping between text label and input\n",
        "        # labels. Repeat letters cost double because of the blank symbol needing to be inserted.\n",
        "        # If a too-long label is provided, ctc_loss returns an infinite gradient\n",
        "        cost = 0\n",
        "        for i in range(len(text)):\n",
        "            if i != 0 and text[i] == text[i - 1]:\n",
        "                cost += 2\n",
        "            else:\n",
        "                cost += 1\n",
        "            if cost > maxTextLen:\n",
        "                return text[:i]\n",
        "        return text\n",
        "\n",
        "    def selectTrainingSet(self):\n",
        "        \"switch to randomly chosen subset of training set\"\n",
        "        self.currIdx = 0\n",
        "        random.shuffle(self.trainSamples)\n",
        "        self.samples = self.trainSamples[:TRAINING_SAMPLES_PER_EPOCH]\n",
        "\n",
        "    def selectValidationSet(self):\n",
        "        \"switch to validation set\"\n",
        "        self.currIdx = 0\n",
        "        random.shuffle(self.validationSamples)\n",
        "        self.samples = self.validationSamples[:\n",
        "                                              VALIDATIOIN_SAMPLES_PER_STEP]\n",
        "\n",
        "    def selectTestSet(self):\n",
        "        \"switch to validation set\"\n",
        "        self.currIdx = 0\n",
        "        random.shuffle(self.testSamples)\n",
        "        self.samples = self.testSamples[:VALIDATIOIN_SAMPLES_PER_STEP]\n",
        "\n",
        "    def getIteratorInfo(self):\n",
        "        \"current batch index and overall number of batches\"\n",
        "        return (self.currIdx // BATCH_SIZE + 1, len(self.samples) // BATCH_SIZE)\n",
        "\n",
        "    def hasNext(self):\n",
        "        \"iterator\"\n",
        "        return self.currIdx + BATCH_SIZE <= len(self.samples)\n",
        "\n",
        "    def getNext(self):\n",
        "        \"iterator\"\n",
        "        batchRange = range(self.currIdx, self.currIdx + BATCH_SIZE)\n",
        "        gtTexts = [self.samples[i].gtText for i in batchRange]\n",
        "\n",
        "        imgs = []\n",
        "        for i in batchRange:\n",
        "            try:\n",
        "                self.binaryImageFile.seek(self.samples[i].imageStartPosition)\n",
        "                img = np.frombuffer(self.binaryImageFile.read(\n",
        "                    self.samples[i].imageSize), np.dtype('B'))\n",
        "                img = img.reshape(\n",
        "                    self.samples[i].imageHeight, self.samples[i].imageWidth)\n",
        "                img = preprocess(img)\n",
        "                # img = preprocess(img, IMAGE_WIDTH, IMAGE_HEIGHT, RESIZE_IMAGE,\n",
        "                #                  CONVERT_IMAGE_TO_MONOCHROME, AUGMENT_IMAGE)\n",
        "                imgs.append(img)\n",
        "            except IOError as e:\n",
        "                print(\"I/O error({0}): {1}\".format(e.errno, e.strerror))\n",
        "                pass\n",
        "            except ValueError as e:\n",
        "                print(e)\n",
        "                pass\n",
        "            except Error as e:\n",
        "                print(\"Unexpected error:\", sys.exc_info()[0])\n",
        "                print(\"Value error({0}): {1}\".format(e.errno, e.strerror))\n",
        "\n",
        "                pass\n",
        "\n",
        "        self.currIdx += BATCH_SIZE\n",
        "        return Batch(gtTexts, imgs)\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8GxWYW6hmiMJ"
      },
      "source": [
        "main.py"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iLw1ko6fmIap"
      },
      "outputs": [],
      "source": [
        "\n",
        "startTime = datetime.now()\n",
        "totalProcessingTime = 0\n",
        "\n",
        "\n",
        "# we only need DataGenerator in training, validation, testing inorder to access the related datasets\n",
        "if OPERATION_TYPE != OperationType.Infer:\n",
        "    dataGenerator = DataGenerator()\n",
        "\n",
        "def accumulateProcessingTime(paraTimeSnapshot):\n",
        "    totalProcessingTime = time.time()\n",
        "    #totalProcessingTime = totalProcessingTime + (time.time() - paraTimeSnapshot)\n",
        "\n",
        "def train(paraModel):\n",
        "    \"train NN\"\n",
        "    epoch = 0  # number of training epochs since start\n",
        "    bestCharErrorRate = float('inf')  # best valdiation character error rate\n",
        "    noImprovementSince = 0  # number of epochs no improvement of character error rate occured\n",
        "\n",
        "    auditString = get_initial_status_log()\n",
        "    print(auditString)\n",
        "    auditLog(auditString)\n",
        "\n",
        "    continueLooping = True\n",
        "\n",
        "    while continueLooping:\n",
        "        print(\"Current Time =\", datetime.now())\n",
        "\n",
        "        epoch += 1\n",
        "        print('Epoch:', epoch)\n",
        "\n",
        "        dataGenerator.selectTrainingSet()\n",
        "\n",
        "        while dataGenerator.hasNext():\n",
        "\n",
        "            timeSnapshot = time.time()\n",
        "\n",
        "            iterInfo = dataGenerator.getIteratorInfo()\n",
        "            batch = dataGenerator.getNext()\n",
        "            loss = paraModel.trainBatch(batch)\n",
        "\n",
        "            # #stop execution after reaching a certain threashold\n",
        "            # if (int(loss) == 1):\n",
        "            #     noImprovementSince = MAXIMUM_NONIMPROVED_EPOCHS;\n",
        "\n",
        "            print('Training Batch:', iterInfo[0],\n",
        "                  '/', iterInfo[1], 'Loss:', loss)\n",
        "\n",
        "            accumulateProcessingTime(timeSnapshot)\n",
        "\n",
        "        # validate\n",
        "        charErrorRate, charSuccessRate, wordsSuccessRate = validate(\n",
        "            paraModel, OperationType.Validation)\n",
        "        auditString = \"Epoch Number %d.\" % epoch + \"\\n\"\n",
        "\n",
        "        # if best validation accuracy so far, save model parameters\n",
        "        if charErrorRate < bestCharErrorRate:\n",
        "            auditString = auditString + 'Character error rate improved, saving model'\n",
        "            paraModel.save()\n",
        "            bestCharErrorRate = charErrorRate\n",
        "            noImprovementSince = 0\n",
        "        else:\n",
        "            auditString = auditString + \"Character error rate not improved\\n\"\n",
        "            noImprovementSince += 1\n",
        "\n",
        "        # stop training if no more improvement in the last x epochs\n",
        "        if noImprovementSince >= MAXIMUM_NONIMPROVED_EPOCHS:\n",
        "            auditString = auditString + \\\n",
        "                \"No more improvement since %d epochs.\" % MAXIMUM_NONIMPROVED_EPOCHS + \"\\n\"\n",
        "\n",
        "            # gracefull termination\n",
        "            continueLooping = False\n",
        "\n",
        "        # Model did not finish, print log and save it\n",
        "        auditString = auditString + \\\n",
        "            get_execution_log(charSuccessRate, wordsSuccessRate)\n",
        "        print(auditString)\n",
        "        auditLog(auditString)\n",
        "\n",
        "\n",
        "def validate(paraModel, paraOperationType):\n",
        "    if paraOperationType == OperationType.Validation:\n",
        "        dataGenerator.selectValidationSet()\n",
        "\n",
        "    elif paraOperationType == OperationType.Testing:\n",
        "        dataGenerator.selectTestSet()\n",
        "\n",
        "    numCharErr = 0\n",
        "    numCharTotal = 0\n",
        "    numWordOK = 0\n",
        "    numWordTotal = 0\n",
        "    timeSnapshot = 0.0\n",
        "\n",
        "    while dataGenerator.hasNext():\n",
        "        timeSnapshot = time.time()\n",
        "\n",
        "        iterInfo = dataGenerator.getIteratorInfo()\n",
        "        print('Validating Batch:', iterInfo[0], '/', iterInfo[1])\n",
        "        batch = dataGenerator.getNext()\n",
        "        (recognized, _) = paraModel.inferBatch(batch)\n",
        "\n",
        "        accumulateProcessingTime(timeSnapshot)\n",
        "\n",
        "        # print('Ground truth -> Recognized')\n",
        "        for i in range(len(recognized)):\n",
        "            numWordTotal += 1\n",
        "            numCharTotal += len(batch.gtTexts[i])\n",
        "\n",
        "            numWordOK += 1 if batch.gtTexts[i] == recognized[i] else 0\n",
        "\n",
        "            dist = editdistance.eval(recognized[i], batch.gtTexts[i])\n",
        "            numCharErr += dist\n",
        "\n",
        "            # remove remark to see each success and error values\n",
        "            #print('[OK]' if dist==0 else '[ERR:%d]' % dist,'\"' + batch.gtTexts[i] + '\"', '->', '\"' + recognized[i] + '\"')\n",
        "\n",
        "    # print validation result\n",
        "    charErrorRate = numCharErr / numCharTotal\n",
        "    charSuccessRate = 1 - (numCharErr / numCharTotal)\n",
        "    wordsSuccessRate = numWordOK / numWordTotal\n",
        "\n",
        "    # print and save validation result, this includes post epoch operation as well as when\n",
        "    # running standalone testing or validation processes\n",
        "\n",
        "    return charErrorRate, charSuccessRate, wordsSuccessRate\n",
        "\n",
        "\n",
        "def inferSingleImage(paraModel, paraFnImg):\n",
        "    \"recognize text in image provided by file path\"\n",
        "    img = cv2.imread(paraFnImg, cv2.IMREAD_GRAYSCALE)\n",
        "    img = preprocess(img)\n",
        "    # img = preprocess(img, IMAGE_WIDTH,\n",
        "    #                  IMAGE_HEIGHT, True, False, False)\n",
        "\n",
        "    batch = Batch(None, [img])\n",
        "    #(recognized, probability) = model.inferBatch(batch)\n",
        "    (recognized, probability) = paraModel.inferBatch(batch, True)\n",
        "    print('Recognized:', '\"' + recognized[0] + '\"')\n",
        "    print('Probability:', probability[0])\n",
        "\n",
        "\n",
        "def get_initial_status_log():\n",
        "    auditString = \"____________________________________________________________\" + \"\\n\"\n",
        "    auditString = auditString + \"Experiment Name: \" + EXPERIMENT_NAME + \"\\n\"\n",
        "    auditString = auditString + \"Base File Name: \" + BASE_FILENAME + \"\\n\"\n",
        "    auditString = auditString + 'Start Execution Time :' + \\\n",
        "        startTime.strftime(\"%m/%d/%Y, %H:%M:%S\") + \"\\n\"\n",
        "    auditString = auditString + \"Training set size: \" + \\\n",
        "        str(len(dataGenerator.trainSamples)) + \"\\n\"\n",
        "    auditString = auditString + \"Validation set size: \" + \\\n",
        "        str(len(dataGenerator.validationSamples)) + \"\\n\"\n",
        "    auditString = auditString + \"Training Samples per epoch: \" + \\\n",
        "        str(TRAINING_SAMPLES_PER_EPOCH) + \"\\n\"\n",
        "    auditString = auditString + \"Validation Samples per step: \" + \\\n",
        "        str(VALIDATIOIN_SAMPLES_PER_STEP) + \"\\n\"\n",
        "    auditString = auditString + \"Batch size: \" + str(BATCH_SIZE) + \"\\n\"\n",
        "    auditString = auditString + \"TRAINING_SAMPLES_PER_EPOCH: \" + \\\n",
        "        str(TRAINING_SAMPLES_PER_EPOCH) + \"\\n\"\n",
        "    auditString = auditString + \"BATCH_SIZE: \" + str(BATCH_SIZE) + \"\\n\"\n",
        "    auditString = auditString + \"VALIDATIOIN_SAMPLES_PER_STEP: \" + \\\n",
        "        str(VALIDATIOIN_SAMPLES_PER_STEP) + \"\\n\"\n",
        "    auditString = auditString + \"TRAINING_DATASET_SIZE: \" + \\\n",
        "        str(TRAINING_DATASET_SIZE) + \"\\n\"\n",
        "    auditString = auditString + \"VALIDATION_DATASET_SPLIT_SIZE: \" + \\\n",
        "        str(VALIDATION_DATASET_SPLIT_SIZE) + \"\\n\"\n",
        "    auditString = auditString + \"IMAGE_WIDTH: \" + \\\n",
        "        str(IMAGE_WIDTH) + \"\\n\"\n",
        "    auditString = auditString + \"IMAGE_HEIGHT: \" + \\\n",
        "        str(IMAGE_HEIGHT) + \"\\n\"\n",
        "    auditString = auditString + \"MAX_TEXT_LENGTH: \" + \\\n",
        "        str(MAX_TEXT_LENGTH) + \"\\n\"\n",
        "    auditString = auditString + \"RESIZE_IMAGE: \" + \\\n",
        "        str(RESIZE_IMAGE) + \"\\n\"\n",
        "    auditString = auditString + \"CONVERT_IMAGE_TO_MONOCHROME: \" + \\\n",
        "        str(CONVERT_IMAGE_TO_MONOCHROME) + \"\\n\"\n",
        "    auditString = auditString + \"MONOCHROME_BINARY_THRESHOLD: \" + \\\n",
        "        str(MONOCHROME_BINARY_THRESHOLD) + \"\\n\"\n",
        "    auditString = auditString + \"AUGMENT_IMAGE: \" + \\\n",
        "        str(AUGMENT_IMAGE) + \"\\n\\n\"\n",
        "\n",
        "    return auditString\n",
        "\n",
        "\n",
        "def get_execution_log(paraCharSuccessRate, paraWordsSuccessRate):\n",
        "    auditString = \"Start Execution Time : \" + \\\n",
        "        startTime.strftime(\"%m/%d/%Y, %H:%M:%S\") + \"\\n\"\n",
        "    auditString = auditString + \"End Execution Time  :\" + \\\n",
        "        datetime.now().strftime(\"%m/%d/%Y, %H:%M:%S\") + \"\\n\"\n",
        "    auditString = auditString + \"Accumulated Processing Time : \" + \\\n",
        "        str(totalProcessingTime / 60) + \" minutes\" + \"\\n\"\n",
        "    auditString = auditString + \"Characters Success Rate: \" + \\\n",
        "        str(paraCharSuccessRate * 100.0) + \"%\\n\"\n",
        "    auditString = auditString + \"Words Success Rate: \" + \\\n",
        "        str(paraWordsSuccessRate * 100.0) + \"%\\n\\n\"\n",
        "\n",
        "    return auditString\n",
        "\n",
        "\n",
        "\n",
        "def main():\n",
        "\n",
        "    if OPERATION_TYPE != OperationType.Infer:\n",
        "        dataGenerator.LoadData(OPERATION_TYPE)\n",
        "\n",
        "    if OPERATION_TYPE == OperationType.Training:\n",
        "        auditString = \"EXPERIMENT_NAME: \" + EXPERIMENT_NAME + \"\\n\"\n",
        "        auditString = auditString + \"Training Using Dataset: \" + \\\n",
        "            str(OPERATION_TYPE) + \"\\n\"\n",
        "        print(auditString)\n",
        "        auditLog(auditString)\n",
        "\n",
        "        model = Model(DECODER_TYPE, mustRestore=False, dump=False)\n",
        "        train(model)\n",
        "    elif OPERATION_TYPE == OperationType.Validation or OPERATION_TYPE == OperationType.Testing:\n",
        "        auditString = \"EXPERIMENT_NAME: \" + EXPERIMENT_NAME + \"\\n\"\n",
        "        auditString = auditString + \"Validation/Tesing Using Dataset: \" + \\\n",
        "            str(OPERATION_TYPE) + \"\\n\"\n",
        "        print(auditString)\n",
        "\n",
        "        model = Model(DECODER_TYPE, mustRestore=True, dump=False)\n",
        "        charErrorRate, charSuccessRate, wordsSuccessRate = validate(\n",
        "            model, OPERATION_TYPE)\n",
        "\n",
        "        auditString = auditString + \\\n",
        "            get_execution_log(charSuccessRate, wordsSuccessRate) + \"\\n\"\n",
        "        print(auditString)\n",
        "\n",
        "        auditLog(auditString)\n",
        "\n",
        "    elif OPERATION_TYPE == OperationType.Infer:  # infer text on test image\n",
        "        print(open(fnResult).read())\n",
        "        tf.reset_default_graph()\n",
        "        #model = Model(open(fnCharList, encoding=\"utf-8\").read(), decoderType, mustRestore=True, dump=args.dump)\n",
        "        model = Model(DECODER_TYPE, mustRestore=True, dump=False)\n",
        "        inferSingleImage(model, fnInfer)\n",
        "\n",
        "\n",
        "if __name__ == '__main__':    \n",
        "    main()\n"
      ]
    }
  ],
  "metadata": {
    "accelerator": "TPU",
    "colab": {
      "authorship_tag": "ABX9TyOnJ7wLqPnN/SJ8dA+jTz7W",
      "collapsed_sections": [],
      "include_colab_link": true,
      "mount_file_id": "1Z0iiLkoSxZ_CdRPE5CdzSdNUJq01fxeb",
      "name": "Arabic OCR.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}