NeuralLayer.init() - Code Metrics - Inspection of "fix GRU" - zomux/deepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( dccd0d...37cade )

by Raphael

created 2016-07-09 23:02 UTC

NeuralLayer.init() A

↳ Parent: NeuralLayer

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

#!/usr/bin/env python
# -*- coding: utf-8 -*-


import logging as loggers

import numpy as np
import theano

from deepy.utils import FLOATX, UniformInitializer, neural_computation, neural_computation_prefer_tensor
from deepy.utils import convert_to_neural_var, convert_to_theano_var, build_activation

logging = loggers.getLogger(__name__)

class NeuralLayer(object):

    def __init__(self, name=None):
        """
        Create a neural layer.
        """
        self.name = name if name else self.__class__.__name__
        self.input_dim = 0
        self.input_dims = [0]
        self.output_dim = 0
        self.output_dims= [0]

        self._linked_block = None

        self.initialized = False
        self.updates = []
        self.training_updates = []
        self.free_parameters = []
        self.parameters = []
        self.training_monitors = []
        self.testing_monitors = []
        self._registered_monitors = set()
        self._registered_updates = set()
        self._registered_training_updates = set()
        self.external_inputs = []
        self.external_targets = []
        self.parameter_count = 0
        self.epoch_callbacks = []
        self.training_callbacks = []
        self.testing_callbacks = []

    def init(self, input_dim=0, input_dims=None, no_prepare=False):
        """
        A short version for initialize function.
        """
        return self.initialize(input_dim, input_dims, no_prepare)

    def initialize(self, input_dim=0, input_dims=None, no_prepare=False):
        """
        Initialize the layer.
        :param no_prepare: avoid calling preparation function
        """
        if self.initialized:
            return
        # configure input dimensions
        if input_dims:
            self.input_dims = input_dims
            self.input_dim = input_dims[0]
        else:
            self.input_dim = input_dim
            self.input_dims = [input_dims]
        # set default output dimension
        if self.output_dim == 0:
            self.output_dim = self.input_dim
        self.initialized = True
        # call prepare
        if not no_prepare:
            self.prepare()
        return self

    def compute(self, *inputs, **kwargs):
        """
        Compute based on NeuralVariable.
        :type inputs:  list of NeuralVariable
        :return: NeuralVariable
        """
        from var import NeuralVariable
        if type(inputs[0]) != NeuralVariable:
            raise SystemError("The input of `compute` must be NeuralVar")

        dims = [t.dim() for t in inputs]
        if len(inputs) == 1:
            self.initialize(input_dim=dims[0])
        else:
            self.initialize(input_dims=dims)
        # convert kwargs
        train_kwargs, test_kwargs, _, _ = convert_to_theano_var(kwargs)

        output = self.compute_tensor(*[t.tensor for t in inputs], **train_kwargs)
        test_output = self.compute_test_tesnor(*[t.test_tensor for t in inputs], **test_kwargs)

        if type(output) != list:
            return NeuralVariable(output, test_output, self.output_dim)
        else:
            return [NeuralVariable(*item) for item in zip(self.output_dims, output, test_output)]

    def prepare(self):
        """
        Prepare function will be called after connected.
        """
        return self.setup()

    def setup(self):
        """
        !!! DEPRECATED !!!
        Setup function will be called after connected.
        """
        pass

    @neural_computation_prefer_tensor
    def compute_tensor(self, *args, **kwargs):
        """
        Compute with tensors in Theano.
        """
        raise NotImplementedError("output function of '%s' is not implemented" % self.name)

    @neural_computation_prefer_tensor
    def compute_test_tesnor(self, *args, **kwargs):
        """
        Compute with tensors in Theano in test time.
        """
        return self.compute_tensor(*args, **kwargs)

    def compute_flexible_tensor(self, x, test=False):
        """
        Deprecated.
        Compute with tensors in Theano, with a parameter to switch test or not.
        """
        if test:
            return self.compute_test_tesnor(x)
        else:
            return self.compute_tensor(x)

    def belongs_to(self, block):
        """
        Let the given block or network manage the parameters of this layer.
        :param block: Block or NeuralNetwork
        :return: NeuralLayer
        """
        if self._linked_block:
            raise SystemError("One layer can not belong to two blocks")
        self._linked_block = block
        block.register_layer(self)
        return self

    def register(self, *layers):
        """
        Register inner layers.
        """
        self.register_inner_layers(*layers)

    def register_inner_layers(self, *layers):
        for layer in layers:
            self.register_parameters(*layer.parameters)
            self.register_updates(*layer.updates)
            self.register_training_updates(*layer.training_updates)

    def register_parameters(self, *parameters):
        """
        Register parameters.
        """
        for param in parameters:
            self.parameter_count += np.prod(param.get_value().shape)
        self.parameters.extend(parameters)

    def register_free_parameters(self, *free_parameters):
        """
        Register free parameters, which means their value will not be learned by trainer.
        """
        return self.free_parameters.extend(free_parameters)

    def register_updates(self, *updates):
        """
        Register updates that will be executed in each iteration.
        """
        for key, node in updates:
            if key not in self._registered_updates:
                self.updates.append((key, node))
                self._registered_updates.add(key)

    def register_training_updates(self, *updates):
        """
        Register updates that will only be executed in training phase.
        """
        for key, node in updates:
            if key not in self._registered_training_updates:
                self.training_updates.append((key, node))
                self._registered_training_updates.add(key)

    def register_monitors(self, *monitors):
        """
        Register monitors they should be tuple of name and Theano variable.
        """
        for key, node in monitors:
            if key not in self._registered_monitors:
                self.training_monitors.append((key, node))
                self.testing_monitors.append((key, node))
                self._registered_monitors.add(key)

    def register_external_inputs(self, *variables):
        """
        Register external input variables.
        """
        self.external_inputs.extend(variables)

    def register_external_targets(self, *variables):
        """
        Register extenal target variables.
        """
        self.external_targets.extend(variables)

    def register_training_callbacks(self, *callbacks):
        """
        Register callback for each iteration in the training.
        """
        self.training_callbacks.extend(callbacks)

    def register_testing_callbacks(self, *callbacks):
        """
        Register callback for each iteration in the testing.
        """
        self.testing_callbacks.extend(callbacks)

    def register_epoch_callbacks(self, *callbacks):
        """
        Register callback which will be called after epoch finished.
        """
        self.epoch_callbacks.extend(callbacks)

    def create_weight(self, input_n=1, output_n=1, suffix="", initializer=None, shape=None):
        if not shape:
            shape = (input_n, output_n)

        if not initializer:
            initializer = UniformInitializer()

        weight = theano.shared(initializer.sample(shape).astype(FLOATX), name='W_{}'.format(suffix))

        logging.info('create weight W_%s: %s', suffix, str(shape))
        return weight

    def create_bias(self, output_n=1, suffix="", value=0., shape=None):
        if not shape:
            shape = (output_n, )
        bs =  np.ones(shape)
        bs *= value
        bias = theano.shared(bs.astype(FLOATX), name='B_{}'.format(suffix))
        logging.info('create bias B_%s: %s', suffix, str(shape))
        return bias

    def create_vector(self, n, name, dtype=FLOATX):
        bs =  np.zeros(n)
        v = theano.shared(bs.astype(dtype), name='{}'.format(name))

        logging.info('create vector %s: %d', name, n)
        return v

    def create_matrix(self, m, n, name):

        matrix = theano.shared(np.zeros((m, n)).astype(FLOATX), name=name)

        logging.info('create matrix %s: %d x %d', name, m, n)
        return matrix

    def activation(self, name):
        return build_activation(name)

    def callback_forward_propagation(self):
        pass



1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4
5			import logging as loggers
6
7			import numpy as np
8			import theano
9
10			from deepy.utils import FLOATX, UniformInitializer, neural_computation, neural_computation_prefer_tensor
11			from deepy.utils import convert_to_neural_var, convert_to_theano_var, build_activation
12
13			logging = loggers.getLogger(__name__)
14
15			class NeuralLayer(object):
16
17			def __init__(self, name=None):
18			"""
19			Create a neural layer.
20			"""
21			self.name = name if name else self.__class__.__name__
22			self.input_dim = 0
23			self.input_dims = [0]
24			self.output_dim = 0
25			self.output_dims= [0]
26
27			self._linked_block = None
28
29			self.initialized = False
30			self.updates = []
31			self.training_updates = []
32			self.free_parameters = []
33			self.parameters = []
34			self.training_monitors = []
35			self.testing_monitors = []
36			self._registered_monitors = set()
37			self._registered_updates = set()
38			self._registered_training_updates = set()
39			self.external_inputs = []
40			self.external_targets = []
41			self.parameter_count = 0
42			self.epoch_callbacks = []
43			self.training_callbacks = []
44			self.testing_callbacks = []
45
46			def init(self, input_dim=0, input_dims=None, no_prepare=False):
47			"""
48			A short version for initialize function.
49			"""
50			return self.initialize(input_dim, input_dims, no_prepare)
51
52			def initialize(self, input_dim=0, input_dims=None, no_prepare=False):
53			"""
54			Initialize the layer.
55			:param no_prepare: avoid calling preparation function
56			"""
57			if self.initialized:
58			return
59			# configure input dimensions
60			if input_dims:
61			self.input_dims = input_dims
62			self.input_dim = input_dims[0]
63			else:
64			self.input_dim = input_dim
65			self.input_dims = [input_dims]
66			# set default output dimension
67			if self.output_dim == 0:
68			self.output_dim = self.input_dim
69			self.initialized = True
70			# call prepare
71			if not no_prepare:
72			self.prepare()
73			return self
74
75			def compute(self, inputs, *kwargs):
76			"""
77			Compute based on NeuralVariable.
78			:type inputs: list of NeuralVariable
79			:return: NeuralVariable
80			"""
81			from var import NeuralVariable
82			if type(inputs[0]) != NeuralVariable:
83			raise SystemError("The input of `compute` must be NeuralVar")
84
85			dims = [t.dim() for t in inputs]
86			if len(inputs) == 1:
87			self.initialize(input_dim=dims[0])
88			else:
89			self.initialize(input_dims=dims)
90			# convert kwargs
91			train_kwargs, test_kwargs, _, _ = convert_to_theano_var(kwargs)
92
93			output = self.compute_tensor([t.tensor for t in inputs], *train_kwargs)
94			test_output = self.compute_test_tesnor([t.test_tensor for t in inputs], *test_kwargs)
95
96			if type(output) != list:
97			return NeuralVariable(output, test_output, self.output_dim)
98			else:
99			return [NeuralVariable(*item) for item in zip(self.output_dims, output, test_output)]
100
101			def prepare(self):
102			"""
103			Prepare function will be called after connected.
104			"""
105			return self.setup()
106
107			def setup(self):
108			"""
109			!!! DEPRECATED !!!
110			Setup function will be called after connected.
111			"""
112			pass
113
114			@neural_computation_prefer_tensor
115			def compute_tensor(self, args, *kwargs):
116			"""
117			Compute with tensors in Theano.
118			"""
119			raise NotImplementedError("output function of '%s' is not implemented" % self.name)
120
121			@neural_computation_prefer_tensor
122			def compute_test_tesnor(self, args, *kwargs):
123			"""
124			Compute with tensors in Theano in test time.
125			"""
126			return self.compute_tensor(args, *kwargs)
127
128			def compute_flexible_tensor(self, x, test=False):
129			"""
130			Deprecated.
131			Compute with tensors in Theano, with a parameter to switch test or not.
132			"""
133			if test:
134			return self.compute_test_tesnor(x)
135			else:
136			return self.compute_tensor(x)
137
138			def belongs_to(self, block):
139			"""
140			Let the given block or network manage the parameters of this layer.
141			:param block: Block or NeuralNetwork
142			:return: NeuralLayer
143			"""
144			if self._linked_block:
145			raise SystemError("One layer can not belong to two blocks")
146			self._linked_block = block
147			block.register_layer(self)
148			return self
149
150			def register(self, *layers):
151			"""
152			Register inner layers.
153			"""
154			self.register_inner_layers(*layers)
155
156			def register_inner_layers(self, *layers):
157			for layer in layers:
158			self.register_parameters(*layer.parameters)
159			self.register_updates(*layer.updates)
160			self.register_training_updates(*layer.training_updates)
161
162			def register_parameters(self, *parameters):
163			"""
164			Register parameters.
165			"""
166			for param in parameters:
167			self.parameter_count += np.prod(param.get_value().shape)
168			self.parameters.extend(parameters)
169
170			def register_free_parameters(self, *free_parameters):
171			"""
172			Register free parameters, which means their value will not be learned by trainer.
173			"""
174			return self.free_parameters.extend(free_parameters)
175
176			def register_updates(self, *updates):
177			"""
178			Register updates that will be executed in each iteration.
179			"""
180			for key, node in updates:
181			if key not in self._registered_updates:
182			self.updates.append((key, node))
183			self._registered_updates.add(key)
184
185			def register_training_updates(self, *updates):
186			"""
187			Register updates that will only be executed in training phase.
188			"""
189			for key, node in updates:
190			if key not in self._registered_training_updates:
191			self.training_updates.append((key, node))
192			self._registered_training_updates.add(key)
193
194			def register_monitors(self, *monitors):
195			"""
196			Register monitors they should be tuple of name and Theano variable.
197			"""
198			for key, node in monitors:
199			if key not in self._registered_monitors:
200			self.training_monitors.append((key, node))
201			self.testing_monitors.append((key, node))
202			self._registered_monitors.add(key)
203
204			def register_external_inputs(self, *variables):
205			"""
206			Register external input variables.
207			"""
208			self.external_inputs.extend(variables)
209
210			def register_external_targets(self, *variables):
211			"""
212			Register extenal target variables.
213			"""
214			self.external_targets.extend(variables)
215
216			def register_training_callbacks(self, *callbacks):
217			"""
218			Register callback for each iteration in the training.
219			"""
220			self.training_callbacks.extend(callbacks)
221
222			def register_testing_callbacks(self, *callbacks):
223			"""
224			Register callback for each iteration in the testing.
225			"""
226			self.testing_callbacks.extend(callbacks)
227
228			def register_epoch_callbacks(self, *callbacks):
229			"""
230			Register callback which will be called after epoch finished.
231			"""
232			self.epoch_callbacks.extend(callbacks)
233
234			def create_weight(self, input_n=1, output_n=1, suffix="", initializer=None, shape=None):
235			if not shape:
236			shape = (input_n, output_n)
237
238			if not initializer:
239			initializer = UniformInitializer()
240
241			weight = theano.shared(initializer.sample(shape).astype(FLOATX), name='W_{}'.format(suffix))
242
243			logging.info('create weight W_%s: %s', suffix, str(shape))
244			return weight
245
246			def create_bias(self, output_n=1, suffix="", value=0., shape=None):
247			if not shape:
248			shape = (output_n, )
249			bs = np.ones(shape)
250			bs *= value
251			bias = theano.shared(bs.astype(FLOATX), name='B_{}'.format(suffix))
252			logging.info('create bias B_%s: %s', suffix, str(shape))
253			return bias
254
255			def create_vector(self, n, name, dtype=FLOATX):
256			bs = np.zeros(n)
257			v = theano.shared(bs.astype(dtype), name='{}'.format(name))
258
259			logging.info('create vector %s: %d', name, n)
260			return v
261
262			def create_matrix(self, m, n, name):
263
264			matrix = theano.shared(np.zeros((m, n)).astype(FLOATX), name=name)
265
266			logging.info('create matrix %s: %d x %d', name, m, n)
267			return matrix
268
269			def activation(self, name):
270			return build_activation(name)
271
272			def callback_forward_propagation(self):
273			pass
274
275

zomux / deepy

Push — master ( dccd0d...37cade )

NeuralLayer.init() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like