MLP.__init__() - 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.init() C

↳ Parent: MLP

Complexity

Conditions

Size

Total Lines

Duplication

Lines	38
Ratio	62.3 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	7
c	1
b	0
f	0
dl	38
loc	61
rs	6.3068

How to fix Long Method

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.__init__() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

MLP.init() C