MLP.predict() - Code Metrics - Inspection of "learning:nn: Add MLP, SDA, LSTM neural network" - TinghuiWang/pyActLearn - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 39f179...514f8f )

by Tinghui

created 2017-01-23 16:43 UTC

MLP.predict() A

↳ Parent: MLP

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	8
rs	9.4285

import logging
import numpy as np
import tensorflow as tf
from .layers import HiddenLayer, SoftmaxLayer
from .injectors import BatchInjector
from .criterion import MonitorBased, ConstIterations

logger = logging.getLogger(__name__)


class MLP:
    """Multi-Layer Perceptron

    Args:
        num_features (:obj:`int`): Number of features.
        num_classes (:obj:`int`): Number of classes.
        layers (:obj:`list` of :obj:`int`): Series of hidden auto-encoder layers.
        activation_fn: activation function used in hidden layer.
        optimizer: Optimizer used for updating weights.

    Attributes:
        num_features (:obj:`int`): Number of features.
        num_classes (:obj:`int`): Number of classes.
        x (:obj:`tensorflow.placeholder`): Input placeholder.
        y_ (:obj:`tensorflow.placeholder`): Output placeholder.
        inner_layers (:obj:`list`): List of inner hidden layers.
        summaries (:obj:`list`): List of tensorflow summaries.
        output_layer: Output softmax layer for multi-class classification, sigmoid for binary classification
        y (:obj:`tensorflow.Tensor`): Softmax/Sigmoid output layer output tensor.
        y_class (:obj:`tensorflow.Tensor`): Tensor to get class label from output layer.
        loss (:obj:`tensorflow.Tensor`): Tensor that represents the cross-entropy loss.
        correct_prediction (:obj:`tensorflow.Tensor`): Tensor that represents the correctness of classification result.
        accuracy (:obj:`tensorflow.Tensor`): Tensor that represents the accuracy of the classifier (exact matching
            ratio in multi-class classification)
        optimizer: Optimizer used for updating weights.
        fit_step (:obj:`tensorflow.Tensor`): Tensor to update weights based on the optimizer algorithm provided.
        sess: Tensorflow session.
        merged: Merged summaries.
    """
    def __init__(self, num_features, num_classes, layers, activation_fn=tf.sigmoid, optimizer=None):
        self.num_features = num_features
        self.num_classes = num_classes
        with tf.name_scope('input'):
            self.x = tf.placeholder(tf.float32, shape=[None, num_features], name='input_x')
            self.y_ = tf.placeholder(tf.float32, shape=[None, num_classes], name='input_y')
        self.inner_layers = []
        self.summaries = []
        # Create Layers
        for i in range(len(layers)):
            if i == 0:

                # First Layer
                self.inner_layers.append(
                    HiddenLayer(num_features, layers[i], x=self.x, name=('Hidden%d' % i), activation_fn=activation_fn)
                )
            else:
                # inner Layer
                self.inner_layers.append(
                    HiddenLayer(layers[i-1], layers[i], x=self.inner_layers[i-1].y,
                                name=('Hidden%d' % i), activation_fn=activation_fn)
                )
            self.summaries += self.inner_layers[i].summaries
        if num_classes == 1:

            # Output Layers
            self.output_layer = HiddenLayer(layers[len(layers) - 1], num_classes, x=self.inner_layers[len(layers)-1].y,
                                            name='Output', activation_fn=tf.sigmoid)
            # Predicted Probability
            self.y = self.output_layer.y
            self.y_class = tf.cast(tf.greater_equal(self.y, 0.5), tf.float32)
            # Loss
            self.loss = tf.reduce_mean(
                tf.nn.sigmoid_cross_entropy_with_logits(self.output_layer.logits, self.y_,
                                                        name='SigmoidCrossEntropyLoss')
            )
            self.correct_prediction = tf.equal(self.y_class, self.y_)
            self.accuracy = tf.reduce_mean(tf.cast(self.correct_prediction, tf.float32))
        else:
            # Output Layers
            self.output_layer = SoftmaxLayer(layers[len(layers) - 1], num_classes, x=self.inner_layers[len(layers)-1].y,
                                             name='OutputLayer')
            # Predicted Probability
            self.y = self.output_layer.y
            self.y_class = tf.argmax(self.y, 1)
            # Loss
            self.loss = tf.reduce_mean(
                tf.nn.softmax_cross_entropy_with_logits(self.output_layer.logits, self.y_,
                                                        name='SoftmaxCrossEntropyLoss')
            )
            self.correct_prediction = tf.equal(self.y_class, tf.argmax(self.y_, 1))
            self.accuracy = tf.reduce_mean(tf.cast(self.correct_prediction, tf.float32))
        self.summaries.append(tf.summary.scalar('cross_entropy', self.loss))
        self.summaries.append(tf.summary.scalar('accuracy', self.accuracy))
        self.summaries += self.output_layer.summaries
        if optimizer is None:
            self.optimizer = tf.train.AdamOptimizer()
        else:
            self.optimizer = optimizer
        with tf.name_scope('train'):
            self.fit_step = self.optimizer.minimize(self.loss)
        self.merged = tf.summary.merge(self.summaries)
        self.sess = None

    def fit(self, x, y, batch_size=100, iter_num=100,
            summaries_dir=None, summary_interval=100,
            test_x=None, test_y=None,
            session=None, criterion='const_iteration'):
        """Fit the model to the dataset

        Args:
            x (:obj:`numpy.ndarray`): Input features of shape (num_samples, num_features).
            y (:obj:`numpy.ndarray`): Corresponding Labels of shape (num_samples) for binary classification,
                or (num_samples, num_classes) for multi-class classification.
            batch_size (:obj:`int`): Batch size used in gradient descent.
            iter_num (:obj:`int`): Number of training iterations for const iterations, step depth for monitor based
                stopping criterion.
            summaries_dir (:obj:`str`): Path of the directory to store summaries and saved values.
            summary_interval (:obj:`int`): The step interval to export variable summaries.
            test_x (:obj:`numpy.ndarray`): Test feature array used for monitoring training progress.
            test_y (:obj:`numpy.ndarray): Test label array used for monitoring training progress.
            session (:obj:`tensorflow.Session`): Session to run training functions.
            criterion (:obj:`str`): Stopping criteria. 'const_iterations' or 'monitor_based'
        """
        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        if summaries_dir is not None:
            train_writer = tf.summary.FileWriter(summaries_dir + '/train')
            test_writer = tf.summary.FileWriter(summaries_dir + '/test')
            valid_writer = tf.summary.FileWriter(summaries_dir + '/valid')
        session.run(tf.global_variables_initializer())
        # Get Stopping Criterion
        if criterion == 'const_iteration':

            criterion = ConstIterations(num_iters=iter_num)
        elif criterion == 'monitor_based':
            num_samples = x.shape[0]
            valid_set_len = int(1/5 * num_samples)
            valid_x = x[num_samples-valid_set_len:num_samples, :]
            valid_y = y[num_samples-valid_set_len:num_samples, :]
            x = x[0:num_samples-valid_set_len, :]
            y = y[0:num_samples-valid_set_len, :]
            _criterion = MonitorBased(n_steps=iter_num,
                                      monitor_fn=self.predict_accuracy, monitor_fn_args=(valid_x, valid_y),
                                      save_fn=tf.train.Saver().save,
                                      save_fn_args=(session, summaries_dir + '/best.ckpt'))
        else:
            logger.error('Wrong criterion %s specified.' % criterion)
            return
        # Setup batch injector
        injector = BatchInjector(data_x=x, data_y=y, batch_size=batch_size)
        i = 0
        train_accuracy = 0
        while _criterion.continue_learning():
            batch_x, batch_y = injector.next_batch()
            if summaries_dir is not None and (i % summary_interval == 0):
                summary, loss, accuracy = session.run([self.merged, self.loss, self.accuracy],
                                                      feed_dict={self.x: x, self.y_: y})
                train_writer.add_summary(summary, i)
                train_accuracy = accuracy
                logger.info('Step %d, train_set accuracy %g, loss %g' % (i, accuracy, loss))
                if (test_x is not None) and (test_y is not None):
                    merged, accuracy = session.run([self.merged, self.accuracy],
                                                   feed_dict={self.x: test_x, self.y_: test_y})
                    test_writer.add_summary(merged, i)
                    logger.info('test_set accuracy %g' % accuracy)
                if criterion == 'monitor_based':
                    merged, accuracy = session.run([self.merged, self.accuracy],
                                                   feed_dict={self.x: valid_x, self.y_: valid_y})
                    valid_writer.add_summary(merged, i)
                    logger.info('valid_set accuracy %g' % accuracy)
            loss, accuracy, _ = session.run([self.loss, self.accuracy, self.fit_step],
                                            feed_dict={self.x: batch_x, self.y_: batch_y})
            #logger.info('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))
            #_ = session.run(self.fit_step, feed_dict={self.x: batch_x, self.y_: batch_y})
            #logger.info('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))
            i += 1
        tf.train.Saver().restore(session, summaries_dir + '/best.ckpt')

    def predict_accuracy(self, x, y, session=None):
        """Get Accuracy given feature array and corresponding labels
        """
        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        return session.run(self.accuracy, feed_dict={self.x: x, self.y_: y})

    def predict_proba(self, x, session=None):
        """Predict probability (Softmax)
        """
        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        return session.run(self.y, feed_dict={self.x: x})

    def predict(self, x, session=None):
        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        return session.run(self.y_class, feed_dict={self.x: x})


1		import logging
2		import numpy as np
3		import tensorflow as tf
4		from .layers import HiddenLayer, SoftmaxLayer
5		from .injectors import BatchInjector
6		from .criterion import MonitorBased, ConstIterations
7
8		logger = logging.getLogger(__name__)
9
10
11		class MLP:
12		"""Multi-Layer Perceptron
13
14		Args:
15		num_features (:obj:`int`): Number of features.
16		num_classes (:obj:`int`): Number of classes.
17		layers (:obj:`list` of :obj:`int`): Series of hidden auto-encoder layers.
18		activation_fn: activation function used in hidden layer.
19		optimizer: Optimizer used for updating weights.
20
21		Attributes:
22		num_features (:obj:`int`): Number of features.
23		num_classes (:obj:`int`): Number of classes.
24		x (:obj:`tensorflow.placeholder`): Input placeholder.
25		y_ (:obj:`tensorflow.placeholder`): Output placeholder.
26		inner_layers (:obj:`list`): List of inner hidden layers.
27		summaries (:obj:`list`): List of tensorflow summaries.
28		output_layer: Output softmax layer for multi-class classification, sigmoid for binary classification
29		y (:obj:`tensorflow.Tensor`): Softmax/Sigmoid output layer output tensor.
30		y_class (:obj:`tensorflow.Tensor`): Tensor to get class label from output layer.
31		loss (:obj:`tensorflow.Tensor`): Tensor that represents the cross-entropy loss.
32		correct_prediction (:obj:`tensorflow.Tensor`): Tensor that represents the correctness of classification result.
33		accuracy (:obj:`tensorflow.Tensor`): Tensor that represents the accuracy of the classifier (exact matching
34		ratio in multi-class classification)
35		optimizer: Optimizer used for updating weights.
36		fit_step (:obj:`tensorflow.Tensor`): Tensor to update weights based on the optimizer algorithm provided.
37		sess: Tensorflow session.
38		merged: Merged summaries.
39		"""
40		def __init__(self, num_features, num_classes, layers, activation_fn=tf.sigmoid, optimizer=None):
41		self.num_features = num_features
42		self.num_classes = num_classes
43		with tf.name_scope('input'):
44		self.x = tf.placeholder(tf.float32, shape=[None, num_features], name='input_x')
45		self.y_ = tf.placeholder(tf.float32, shape=[None, num_classes], name='input_y')
46		self.inner_layers = []
47		self.summaries = []
48		# Create Layers
49		for i in range(len(layers)):
50	View Code Duplication	if i == 0:
		0 ignored issues – show Duplication introduced 2017-01-23 16:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
51		# First Layer
52		self.inner_layers.append(
53		HiddenLayer(num_features, layers[i], x=self.x, name=('Hidden%d' % i), activation_fn=activation_fn)
54		)
55		else:
56		# inner Layer
57		self.inner_layers.append(
58		HiddenLayer(layers[i-1], layers[i], x=self.inner_layers[i-1].y,
59		name=('Hidden%d' % i), activation_fn=activation_fn)
60		)
61		self.summaries += self.inner_layers[i].summaries
62	View Code Duplication	if num_classes == 1:
		0 ignored issues – show Duplication introduced 2017-01-23 16:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
63		# Output Layers
64		self.output_layer = HiddenLayer(layers[len(layers) - 1], num_classes, x=self.inner_layers[len(layers)-1].y,
65		name='Output', activation_fn=tf.sigmoid)
66		# Predicted Probability
67		self.y = self.output_layer.y
68		self.y_class = tf.cast(tf.greater_equal(self.y, 0.5), tf.float32)
69		# Loss
70		self.loss = tf.reduce_mean(
71		tf.nn.sigmoid_cross_entropy_with_logits(self.output_layer.logits, self.y_,
72		name='SigmoidCrossEntropyLoss')
73		)
74		self.correct_prediction = tf.equal(self.y_class, self.y_)
75		self.accuracy = tf.reduce_mean(tf.cast(self.correct_prediction, tf.float32))
76		else:
77		# Output Layers
78		self.output_layer = SoftmaxLayer(layers[len(layers) - 1], num_classes, x=self.inner_layers[len(layers)-1].y,
79		name='OutputLayer')
80		# Predicted Probability
81		self.y = self.output_layer.y
82		self.y_class = tf.argmax(self.y, 1)
83		# Loss
84		self.loss = tf.reduce_mean(
85		tf.nn.softmax_cross_entropy_with_logits(self.output_layer.logits, self.y_,
86		name='SoftmaxCrossEntropyLoss')
87		)
88		self.correct_prediction = tf.equal(self.y_class, tf.argmax(self.y_, 1))
89		self.accuracy = tf.reduce_mean(tf.cast(self.correct_prediction, tf.float32))
90		self.summaries.append(tf.summary.scalar('cross_entropy', self.loss))
91		self.summaries.append(tf.summary.scalar('accuracy', self.accuracy))
92		self.summaries += self.output_layer.summaries
93		if optimizer is None:
94		self.optimizer = tf.train.AdamOptimizer()
95		else:
96		self.optimizer = optimizer
97		with tf.name_scope('train'):
98		self.fit_step = self.optimizer.minimize(self.loss)
99		self.merged = tf.summary.merge(self.summaries)
100		self.sess = None
101
102		def fit(self, x, y, batch_size=100, iter_num=100,
103		summaries_dir=None, summary_interval=100,
104		test_x=None, test_y=None,
105		session=None, criterion='const_iteration'):
106		"""Fit the model to the dataset
107
108		Args:
109		x (:obj:`numpy.ndarray`): Input features of shape (num_samples, num_features).
110		y (:obj:`numpy.ndarray`): Corresponding Labels of shape (num_samples) for binary classification,
111		or (num_samples, num_classes) for multi-class classification.
112		batch_size (:obj:`int`): Batch size used in gradient descent.
113		iter_num (:obj:`int`): Number of training iterations for const iterations, step depth for monitor based
114		stopping criterion.
115		summaries_dir (:obj:`str`): Path of the directory to store summaries and saved values.
116		summary_interval (:obj:`int`): The step interval to export variable summaries.
117		test_x (:obj:`numpy.ndarray`): Test feature array used for monitoring training progress.
118		test_y (:obj:`numpy.ndarray): Test label array used for monitoring training progress.
119		session (:obj:`tensorflow.Session`): Session to run training functions.
120		criterion (:obj:`str`): Stopping criteria. 'const_iterations' or 'monitor_based'
121		"""
122		if session is None:
123		if self.sess is None:
124		session = tf.Session()
125		self.sess = session
126		else:
127		session = self.sess
128		if summaries_dir is not None:
129		train_writer = tf.summary.FileWriter(summaries_dir + '/train')
130		test_writer = tf.summary.FileWriter(summaries_dir + '/test')
131		valid_writer = tf.summary.FileWriter(summaries_dir + '/valid')
132		session.run(tf.global_variables_initializer())
133		# Get Stopping Criterion
134	View Code Duplication	if criterion == 'const_iteration':
		0 ignored issues – show Duplication introduced 2017-01-23 16:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
135		criterion = ConstIterations(num_iters=iter_num)
136		elif criterion == 'monitor_based':
137		num_samples = x.shape[0]
138		valid_set_len = int(1/5 * num_samples)
139		valid_x = x[num_samples-valid_set_len:num_samples, :]
140		valid_y = y[num_samples-valid_set_len:num_samples, :]
141		x = x[0:num_samples-valid_set_len, :]
142		y = y[0:num_samples-valid_set_len, :]
143		_criterion = MonitorBased(n_steps=iter_num,
144		monitor_fn=self.predict_accuracy, monitor_fn_args=(valid_x, valid_y),
145		save_fn=tf.train.Saver().save,
146		save_fn_args=(session, summaries_dir + '/best.ckpt'))
147		else:
148		logger.error('Wrong criterion %s specified.' % criterion)
149		return
150		# Setup batch injector
151		injector = BatchInjector(data_x=x, data_y=y, batch_size=batch_size)
152		i = 0
153		train_accuracy = 0
154		while _criterion.continue_learning():
155		batch_x, batch_y = injector.next_batch()
156		if summaries_dir is not None and (i % summary_interval == 0):
157		summary, loss, accuracy = session.run([self.merged, self.loss, self.accuracy],
158		feed_dict={self.x: x, self.y_: y})
159		train_writer.add_summary(summary, i)
160		train_accuracy = accuracy
161		logger.info('Step %d, train_set accuracy %g, loss %g' % (i, accuracy, loss))
162		if (test_x is not None) and (test_y is not None):
163		merged, accuracy = session.run([self.merged, self.accuracy],
164		feed_dict={self.x: test_x, self.y_: test_y})
165		test_writer.add_summary(merged, i)
166		logger.info('test_set accuracy %g' % accuracy)
167		if criterion == 'monitor_based':
168		merged, accuracy = session.run([self.merged, self.accuracy],
169		feed_dict={self.x: valid_x, self.y_: valid_y})
170		valid_writer.add_summary(merged, i)
171		logger.info('valid_set accuracy %g' % accuracy)
172		loss, accuracy, _ = session.run([self.loss, self.accuracy, self.fit_step],
173		feed_dict={self.x: batch_x, self.y_: batch_y})
174		#logger.info('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))
175		#_ = session.run(self.fit_step, feed_dict={self.x: batch_x, self.y_: batch_y})
176		#logger.info('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))
177		i += 1
178		tf.train.Saver().restore(session, summaries_dir + '/best.ckpt')
179
180		def predict_accuracy(self, x, y, session=None):
181		"""Get Accuracy given feature array and corresponding labels
182		"""
183		if session is None:
184		if self.sess is None:
185		session = tf.Session()
186		self.sess = session
187		else:
188		session = self.sess
189		return session.run(self.accuracy, feed_dict={self.x: x, self.y_: y})
190
191		def predict_proba(self, x, session=None):
192		"""Predict probability (Softmax)
193		"""
194		if session is None:
195		if self.sess is None:
196		session = tf.Session()
197		self.sess = session
198		else:
199		session = self.sess
200		return session.run(self.y, feed_dict={self.x: x})
201
202		def predict(self, x, session=None):
203		if session is None:
204		if self.sess is None:
205		session = tf.Session()
206		self.sess = session
207		else:
208		session = self.sess
209		return session.run(self.y_class, feed_dict={self.x: x})
210

TinghuiWang / pyActLearn

Push — master ( 39f179...514f8f )

MLP.predict() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like