I just made my machine learning code work a few days ago and I would like to know if there's a way to improve my code.

Before I get to the implementation of the tasks at hand, I would like to introduce the design I have in mind.

I first created interfaces containing methods that are the tasks needed to be performed in general. These are the interfaces relevant to the code I'm refactoring.

Network Interface

from abc import ABC, abstractmethod

class Network(ABC):
    def __init__(self, network_config):
        self.network_config = network_config
        super(Network, self).__init__()

    def train(self, train_config, train_features, validation_features, train_labels, validation_labels):
        raise NotImplementedError

    def test(self, test_features, test_labels):
        raise NotImplementedError

    def predict(self, data_to_be_predicted):
        raise NotImplementedError

These are the tasks I think that are performed in all Machine Learning problems. The reason for using a network_config as the initialization parameter and train_config as parameters is that the parameters (e.g. number of layers for the network, learning rate, etc.) Machine Learning problems use differ from each other. Some require paremeters for the use of momentum and whatnot. Some don't.

My reason for this design is flexibility such that people can switch from one implementation of this interface to another any time.

ImagePreprocessor Interface

from abc import ABC, abstractmethod

class ImagePreprocessor(ABC):
    def __init__(self):
        super(ImagePreprocessor, self).__init__()

    def resize(images, shape):
        raise NotImplementedError

    def read(image_paths):
        raise NotImplementedError

Since the tasks involves the use of images, I also created an interface that will read and preprocess the images.

DatasetPreparator Interface

class DatasetPreparator(ABC):
    def get_dataset_from(dataset_config):
        raise NotImplementedError

    def split_into_train_validation_and_test_sets(features, validation_size, test_size, labels=None):
        raise NotImplementedError

Apparently, it's best to split the dataset into training, validation, and testing. Machine learning problems also require getting the dataset.

Here are the implementations for said interfaces:


def load_charset(charset_file):
    return ''.join([line.rstrip('\n') for line in open(charset_file)])

class TensorflowNetwork(Network):
    def __init__(self, network_config):
        self.graph = tf.Graph()
        with self.graph.as_default():
            self.inputs = tf.placeholder(tf.float32, [None, None, network_config.num_features], name="input")
            self.labels = tf.sparse_placeholder(tf.int32, name="label")
            self.seq_len = tf.placeholder(tf.int32, [None], name="seq_len_input")

            logits = self._bidirectional_lstm_layers(

            self.global_step = tf.Variable(0, trainable=False)
            self.loss = tf.nn.ctc_loss(labels=self.labels, inputs=logits, sequence_length=self.seq_len)
            self.cost = tf.reduce_mean(self.loss)

            self.optimizer = tf.train.AdamOptimizer(network_config.learning_rate).minimize(self.cost)
            self.decoded, _ = tf.nn.ctc_beam_search_decoder(inputs=logits, sequence_length=self.seq_len, merge_repeated=False)
            self.dense_decoded = tf.sparse_to_dense(tf.to_int32(self.decoded[0].indices),
            self.label_error_rate = tf.reduce_mean(tf.edit_distance(tf.cast(self.decoded[0], tf.int32), self.labels))

            tf.summary.scalar('cost', self.cost)
            self.merged_summary = tf.summary.merge_all()
        super(TensorflowNetwork, self).__init__(network_config)

    def train(self, train_config, train_features, validation_features, train_labels=None, validation_labels=None):
        logger = logging.getLogger('Training for ocr using BI_LSTM_CTC')

        encoder_decoder = EncoderDecoder()

        encoded_train_labels = []
        encoded_val_labels = []
        for train_label in train_labels:
        for val_label in validation_labels:
        train_labels = encoded_train_labels
        validation_labels = encoded_val_labels

        print('loading train data, please wait---------------------', end=' ')
        train_feeder = DataIterator(train_features, train_labels, train_config.batch_size)
        print('number of training images: ', train_feeder.get_number_of_examples())

        print('loading validation data, please wait---------------------', end=' ')
        val_feeder = DataIterator(validation_features, validation_labels, train_config.batch_size)
        print('number of validation images: ', val_feeder.get_number_of_examples())

        num_train_samples = train_feeder.get_number_of_examples()
        num_batches_per_epoch = int(num_train_samples/train_config.batch_size)

        config = tf.ConfigProto(log_device_placement=False, allow_soft_placement=False)
        with tf.Session(graph=self.graph, config=config) as sess:

            log_dir = train_config.log_dir + 'train'

            saver = tf.train.Saver(max_to_keep=5)

            train_writer = tf.summary.FileWriter(log_dir, sess.graph)
            if train_config.is_restore:
                ckpt = tf.train.latest_checkpoint(train_config.checkpoint_dir)
                if ckpt:
                    saver.restore(sess, ckpt)
                    print('restore from the checkpoint{0}'.format(ckpt))

            print('=============================begin training=============================')
            val_inputs, val_seq_len, val_labels = val_feeder.get_whole_data()
            val_feed = {
                self.inputs: val_inputs,
                self.labels: val_labels,
                self.seq_len: val_seq_len

            for current_epoch in range(train_config.num_epochs):
                shuffle_index = np.random.permutation(num_train_samples)
                train_cost = 0
                start = time.time()

                for current_batch_number in range(num_batches_per_epoch):
                    train_feed = self._get_batch_feed(current_batch_number, shuffle_index, train_feeder)

                    summary_str, batch_cost, step, _ = sess.run([self.merged_summary, self.cost, self.global_step, self.optimizer], train_feed)
                    train_cost += batch_cost * train_config.batch_size
                    train_writer.add_summary(summary_str, step)

                    if not os.path.isdir(train_config.checkpoint_dir):
                    logger.info('save the checkpoint of {}'.format(step))

                    tf.train.write_graph(sess.graph_def, log_dir, 'bi_lstm_ctc_ocr.pbtxt')
                    saver.save(sess, os.path.join(train_config.checkpoint_dir, 'ocr-model-{}.ckpt'.format(current_epoch)), global_step=step)

                    if step % train_config.validation_steps == 0:
                        dense_decoded, last_batch_err = sess.run([self.dense_decoded, self.label_error_rate], val_feed)
                        avg_train_cost = train_cost/((current_batch_number + 1) * train_config.batch_size)
                        print("Batch {}/{}, Epoch {}/{}, avg_train_cost: {:.3f}, last_batch_err: {:.3f}, time: {:.3f}"
                              .format(current_batch_number, num_batches_per_epoch, current_epoch + 1, train_config.num_epochs, avg_train_cost, last_batch_err, time.time() - start))

    def _get_batch_feed(self, current_batch_number, shuffle_index, train_feeder):
        batch_inputs, batch_seq_len, batch_labels = train_feeder.get_next_batch(current_batch_number,
        feed = {
            self.inputs: batch_inputs,
            self.labels: batch_labels,
            self.seq_len: batch_seq_len
        return feed

    def test(self, test_features, test_labels=None):
        print("I'm testing!")

    def predict(self, data_to_be_predicted):
        print("I'm predicting!")

    def _bidirectional_lstm_layers(self, num_hidden, num_layers, num_classes):
        lstm_fw_cells = [tf.contrib.rnn.BasicLSTMCell(num_hidden, forget_bias=1.0) for _ in range(num_layers)]
        lstm_bw_cells = [tf.contrib.rnn.BasicLSTMCell(num_hidden, forget_bias=1.0) for _ in range(num_layers)]

        outputs, _, _ = tf.contrib.rnn.stack_bidirectional_dynamic_rnn(lstm_fw_cells, lstm_bw_cells,
                                                                           self.inputs, dtype=tf.float32)

        batch_size = tf.shape(self.inputs)[0]

        outputs = tf.reshape(outputs, [-1, num_hidden])

        W = tf.Variable(tf.truncated_normal([num_hidden, num_classes], stddev=0.1, dtype=tf.float32))
        b = tf.Variable(tf.constant(0., shape=[num_classes], dtype=tf.float32))

        logits = tf.matmul(outputs, W) + b
        logits = tf.reshape(logits, [batch_size, -1, num_classes])
        logits = tf.transpose(logits, (1, 0, 2))

        return logits

I haven't implemented the testing and predicting codes yet. For now, what I'm trying to refactor is the training part and the initialization of the network. As you can see, I used Tensorflow since that's the library I like using.


class CV2ImagePreprocessor(ImagePreprocessor):
    def resize(images, shape):
        resized_images = []
        for image in images:
            resized_images.append(cv2.resize(image, shape).swapaxes(0,1))
        return resized_images

    def read(image_paths):
        images = []
        for image_path in image_paths:
            images.append(cv2.imread(image_path, 0).astype(np.float32))
        return images

As you can see, I used CV2 to read and resize the images.


class IAMDatasetPreparator(DatasetPreparator):
    def get_dataset_from(iam_dataset_config):
        image_paths = []
        labels = []
        with open(iam_dataset_config.labels_file) as f:
            labeled_data = f.readlines()
        labeled_data = [x.strip() for x in labeled_data]
        for example in labeled_data:
            example_data = example.split()
            if example_data[1] == "ok":
                image_path = iam_dataset_config.data_dir + example_data[0] + ".png"
                label = example_data[-1]
        return image_paths, labels

    def split_into_train_validation_and_test_sets(images, validation_size, test_size, labels=None):
        train_images, test_images, train_labels, test_labels = train_test_split(
        validation_images, test_images, validation_labels, test_labels = train_test_split(
        return train_images, train_labels, validation_images, validation_labels, test_images, test_labels

The dataset I am working with here is the IAM Offline Handwriting Dataset. I used sklearn to split the dataset.

Here's the script I have for training a model.

network_config = NetworkConfig()
network = TensorflowNetwork(network_config)
dataset_config = DatasetConfig()
image_paths, labels = IAMDatasetPreparator.get_image_paths_and_labels_from(dataset_config)
train_data, train_labels, val_data, val_labels, _, _ = \
            IAMDatasetPreparator.split_into_train_validation_and_test_sets(image_paths, 0.5, 0.5, labels)
train_config = TrainConfig()
network.train(train_config, train_data, val_data, train_labels, val_labels)

If you want to try it out for yourself, clone this repository. The working implementation uses Tensroflow so Tensorflow is required to get it running. Simply run dummy_train.py to train a model for 1 epoch. Do note that I'm not trying to improve the accuracy for this question. I'm just trying to know if there's something I can do to improve my code.

  • \$\begingroup\$ i'm trying to run your program, gives me line.txt file not found, could please tell the contents of that file. thanks \$\endgroup\$ – hicham taybi Jul 16 '18 at 14:26
  • \$\begingroup\$ The lines.txt is part of the IAM handwriting dataset you choose (which would be a dataset containing lines of text). \$\endgroup\$ – Rocket Pingu Feb 21 at 21:51

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.