LSTM.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

LSTM.predict() B

↳ Parent: LSTM

Complexity

Conditions

Size

Total Lines

Duplication

Lines	20
Ratio	100 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	20
loc	20
rs	8.5454

import logging
import numpy as np
import tensorflow as tf
from .layers import HiddenLayer, SoftmaxLayer
from .injectors import BatchSequenceInjector

logger = logging.getLogger(__name__)


class LSTM:
    """Basic Single Layer Long-Short-Term Memory
    """
    def __init__(self, num_features, num_classes, num_units, num_steps, optimizer=None):
        self.num_features = num_features
        self.num_classes = num_classes
        self.num_steps = num_steps
        self.num_units = num_units
        self.summaries = []
        with tf.name_scope('input'):
            self.x = tf.placeholder(tf.float32, shape=[None, num_steps, num_features], name='input_x')
            self.init_state = tf.placeholder(tf.float32, shape=[None, 2 * num_units], name='init_state')
            self.y_ = tf.placeholder(tf.float32, shape=[None, num_classes], name='input_y')
        # Input Hidden Layer - Need to unroll num_steps and apply W/b
        hidden_x = tf.reshape(tf.transpose(self.x, [1, 0, 2]), [-1, num_features])
        self.hidden_layer = HiddenLayer(num_features, num_units, 'Hidden', x=hidden_x)
        # Output of the hidden layer needs to be split to be used with RNN
        hidden_y = tf.split(0, num_steps, self.hidden_layer.y)
        # Apply RNN
        self.cell = tf.nn.rnn_cell.BasicLSTMCell(num_units=num_units, state_is_tuple=False)
        outputs, states = tf.nn.rnn(self.cell, hidden_y, initial_state=self.init_state)
        self.last_state = states[-1]
        # Output Softmax Layer
        self.output_layer = SoftmaxLayer(num_units, num_classes, 'SoftmaxLayer', x=outputs[-1])
        # Predicted Probability
        self.y = self.output_layer.y
        self.y_class = tf.argmax(self.y, 1)
        # Softmax Cross-Entropy Loss
        self.loss = tf.reduce_mean(
            tf.nn.softmax_cross_entropy_with_logits(self.output_layer.logits, self.y_,
                                                    name='SoftmaxCrossEntropy')
        )
        # Setup Optimizer
        if optimizer is None:
            self.optimizer = tf.train.AdamOptimizer()
        else:
            self.optimizer = optimizer
        # Evaluation
        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))
        # Fit Step
        with tf.name_scope('train'):
            self.fit_step = self.optimizer.minimize(self.loss)
        # Setup Summaries
        self.summaries += self.hidden_layer.summaries
        self.summaries += self.output_layer.summaries
        self.summaries.append(tf.summary.scalar('cross_entropy', self.loss))
        self.summaries.append(tf.summary.scalar('accuracy', self.accuracy))
        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=10,
            test_x=None, test_y=None, session=None, criterion=None):
        """Fit the model to the dataset
        """
        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')
        session.run(tf.global_variables_initializer())
        # Setup batch injector
        injector = BatchSequenceInjector(data_x=x, data_y=y, batch_size=batch_size, seq_len=self.num_steps)
        # Test sequences
        if (test_x is not None) and (test_y is not None):
            test_seq_x, test_seq_y = BatchSequenceInjector.to_sequence(
                self.num_steps, test_x, test_y, start=0, end=2000
            )
        train_seq_x, train_seq_y = BatchSequenceInjector.to_sequence(
            self.num_steps, x, y, start=0, end=2000
        )
        for i in range(iter_num):
            batch_x, batch_y = injector.next_batch()
            if summaries_dir is not None and (i % summary_interval == 0):
                summary = session.run(
                    self.merged,
                    feed_dict={self.x: train_seq_x, self.y_: train_seq_y,
                               self.init_state: np.zeros((train_seq_x.shape[0], 2 * self.num_units))}
                )
                train_writer.add_summary(summary, i)
                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_seq_x, self.y_: test_seq_y,
                                   self.init_state: np.zeros((test_seq_x.shape[0], 2*self.num_units))})
                    test_writer.add_summary(merged, i)
                    print('test accuracy %g' % accuracy)
            loss, accuracy, _ = session.run(
                [self.loss, self.accuracy, self.fit_step],
                feed_dict={self.x: batch_x, self.y_: batch_y,
                           self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
            print('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))

    def predict_proba(self, x, session=None, batch_size=500):

        """Predict probability (Softmax)
        """
        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        injector = BatchSequenceInjector(batch_size=batch_size, data_x=x, seq_len=self.num_steps)
        injector.reset()
        result = None
        while injector.num_epochs == 0:
            batch_x = injector.next_batch()
            batch_y = session.run(self.y,
                                  feed_dict={self.x: batch_x,
                                             self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
            if result is None:
                result = batch_y
            else:
                result = np.concatenate((result, batch_y), axis=0)
        return result

    def predict(self, x, session=None, batch_size=500):

        if session is None:
            if self.sess is None:
                session = tf.Session()
                self.sess = session
            else:
                session = self.sess
        injector = BatchSequenceInjector(batch_size=batch_size, data_x=x, seq_len=self.num_steps)
        injector.reset()
        result = None
        while injector.num_epochs == 0:
            batch_x = injector.next_batch()
            batch_y = session.run(self.y_class,
                                  feed_dict={self.x: batch_x,
                                             self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
            if result is None:
                result = batch_y
            else:
                result = np.concatenate((result, batch_y), axis=0)
        return result


1		import logging
2		import numpy as np
3		import tensorflow as tf
4		from .layers import HiddenLayer, SoftmaxLayer
5		from .injectors import BatchSequenceInjector
6
7		logger = logging.getLogger(__name__)
8
9
10		class LSTM:
11		"""Basic Single Layer Long-Short-Term Memory
12		"""
13		def __init__(self, num_features, num_classes, num_units, num_steps, optimizer=None):
14		self.num_features = num_features
15		self.num_classes = num_classes
16		self.num_steps = num_steps
17		self.num_units = num_units
18		self.summaries = []
19		with tf.name_scope('input'):
20		self.x = tf.placeholder(tf.float32, shape=[None, num_steps, num_features], name='input_x')
21		self.init_state = tf.placeholder(tf.float32, shape=[None, 2 * num_units], name='init_state')
22		self.y_ = tf.placeholder(tf.float32, shape=[None, num_classes], name='input_y')
23		# Input Hidden Layer - Need to unroll num_steps and apply W/b
24		hidden_x = tf.reshape(tf.transpose(self.x, [1, 0, 2]), [-1, num_features])
25		self.hidden_layer = HiddenLayer(num_features, num_units, 'Hidden', x=hidden_x)
26		# Output of the hidden layer needs to be split to be used with RNN
27		hidden_y = tf.split(0, num_steps, self.hidden_layer.y)
28		# Apply RNN
29		self.cell = tf.nn.rnn_cell.BasicLSTMCell(num_units=num_units, state_is_tuple=False)
30		outputs, states = tf.nn.rnn(self.cell, hidden_y, initial_state=self.init_state)
31		self.last_state = states[-1]
32		# Output Softmax Layer
33		self.output_layer = SoftmaxLayer(num_units, num_classes, 'SoftmaxLayer', x=outputs[-1])
34		# Predicted Probability
35		self.y = self.output_layer.y
36		self.y_class = tf.argmax(self.y, 1)
37		# Softmax Cross-Entropy Loss
38		self.loss = tf.reduce_mean(
39		tf.nn.softmax_cross_entropy_with_logits(self.output_layer.logits, self.y_,
40		name='SoftmaxCrossEntropy')
41		)
42		# Setup Optimizer
43		if optimizer is None:
44		self.optimizer = tf.train.AdamOptimizer()
45		else:
46		self.optimizer = optimizer
47		# Evaluation
48		self.correct_prediction = tf.equal(self.y_class, tf.argmax(self.y_, 1))
49		self.accuracy = tf.reduce_mean(tf.cast(self.correct_prediction, tf.float32))
50		# Fit Step
51		with tf.name_scope('train'):
52		self.fit_step = self.optimizer.minimize(self.loss)
53		# Setup Summaries
54		self.summaries += self.hidden_layer.summaries
55		self.summaries += self.output_layer.summaries
56		self.summaries.append(tf.summary.scalar('cross_entropy', self.loss))
57		self.summaries.append(tf.summary.scalar('accuracy', self.accuracy))
58		self.merged = tf.summary.merge(self.summaries)
59		self.sess = None
60
61		def fit(self, x, y, batch_size=100, iter_num=100, summaries_dir=None, summary_interval=10,
62		test_x=None, test_y=None, session=None, criterion=None):
63		"""Fit the model to the dataset
64		"""
65		if session is None:
66		if self.sess is None:
67		session = tf.Session()
68		self.sess = session
69		else:
70		session = self.sess
71		if summaries_dir is not None:
72		train_writer = tf.summary.FileWriter(summaries_dir + '/train')
73		test_writer = tf.summary.FileWriter(summaries_dir + '/test')
74		session.run(tf.global_variables_initializer())
75		# Setup batch injector
76		injector = BatchSequenceInjector(data_x=x, data_y=y, batch_size=batch_size, seq_len=self.num_steps)
77		# Test sequences
78		if (test_x is not None) and (test_y is not None):
79		test_seq_x, test_seq_y = BatchSequenceInjector.to_sequence(
80		self.num_steps, test_x, test_y, start=0, end=2000
81		)
82		train_seq_x, train_seq_y = BatchSequenceInjector.to_sequence(
83		self.num_steps, x, y, start=0, end=2000
84		)
85		for i in range(iter_num):
86		batch_x, batch_y = injector.next_batch()
87		if summaries_dir is not None and (i % summary_interval == 0):
88		summary = session.run(
89		self.merged,
90		feed_dict={self.x: train_seq_x, self.y_: train_seq_y,
91		self.init_state: np.zeros((train_seq_x.shape[0], 2 * self.num_units))}
92		)
93		train_writer.add_summary(summary, i)
94		if (test_x is not None) and (test_y is not None):
95		merged, accuracy = session.run(
96		[self.merged, self.accuracy],
97		feed_dict={self.x: test_seq_x, self.y_: test_seq_y,
98		self.init_state: np.zeros((test_seq_x.shape[0], 2*self.num_units))})
99		test_writer.add_summary(merged, i)
100		print('test accuracy %g' % accuracy)
101		loss, accuracy, _ = session.run(
102		[self.loss, self.accuracy, self.fit_step],
103		feed_dict={self.x: batch_x, self.y_: batch_y,
104		self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
105		print('Step %d, training accuracy %g, loss %g' % (i, accuracy, loss))
106
107	View Code Duplication	def predict_proba(self, x, session=None, batch_size=500):
		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...
108		"""Predict probability (Softmax)
109		"""
110		if session is None:
111		if self.sess is None:
112		session = tf.Session()
113		self.sess = session
114		else:
115		session = self.sess
116		injector = BatchSequenceInjector(batch_size=batch_size, data_x=x, seq_len=self.num_steps)
117		injector.reset()
118		result = None
119		while injector.num_epochs == 0:
120		batch_x = injector.next_batch()
121		batch_y = session.run(self.y,
122		feed_dict={self.x: batch_x,
123		self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
124		if result is None:
125		result = batch_y
126		else:
127		result = np.concatenate((result, batch_y), axis=0)
128		return result
129
130	View Code Duplication	def predict(self, x, session=None, batch_size=500):
		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...
131		if session is None:
132		if self.sess is None:
133		session = tf.Session()
134		self.sess = session
135		else:
136		session = self.sess
137		injector = BatchSequenceInjector(batch_size=batch_size, data_x=x, seq_len=self.num_steps)
138		injector.reset()
139		result = None
140		while injector.num_epochs == 0:
141		batch_x = injector.next_batch()
142		batch_y = session.run(self.y_class,
143		feed_dict={self.x: batch_x,
144		self.init_state: np.zeros((batch_x.shape[0], 2 * self.num_units))})
145		if result is None:
146		result = batch_y
147		else:
148		result = np.concatenate((result, batch_y), axis=0)
149		return result
150

TinghuiWang / pyActLearn

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

LSTM.predict() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like