|
- #! /usr/bin/python
- # -*- coding: utf-8 -*-
-
- import tensorflow as tf
- from tensorflow.python.ops.rnn_cell import LSTMStateTuple
-
- import tensorlayer as tl
- from tensorlayer import logging
- from tensorlayer.decorators import deprecated, deprecated_alias
- from tensorlayer.backend.ops.load_backend import BACKEND
-
- __all__ = [
- 'cabs',
- 'compute_alpha',
- 'flatten_reshape',
- 'get_collection_trainable',
- 'get_layers_with_name',
- 'get_variables_with_name',
- 'initialize_global_variables',
- 'initialize_rnn_state',
- 'list_remove_repeat',
- 'merge_networks',
- 'print_all_variables',
- 'quantize',
- 'quantize_active',
- 'quantize_weight',
- 'quantize_active_overflow',
- 'quantize_weight_overflow',
- 'set_name_reuse',
- 'ternary_operation',
- ]
-
- ########## Module Public Functions ##########
-
-
- def cabs(x):
- return tf.minimum(1.0, tf.abs(x), name='cabs')
-
-
- def compute_alpha(x):
- """Computing the scale parameter."""
- threshold = _compute_threshold(x)
- alpha1_temp1 = tf.where(tf.greater(x, threshold), x, tf.zeros_like(x, tf.float32))
- alpha1_temp2 = tf.where(tf.less(x, -threshold), x, tf.zeros_like(x, tf.float32))
- alpha_array = tf.add(alpha1_temp1, alpha1_temp2, name=None)
- alpha_array_abs = tf.abs(alpha_array)
- alpha_array_abs1 = tf.where(
- tf.greater(alpha_array_abs, 0), tf.ones_like(alpha_array_abs, tf.float32),
- tf.zeros_like(alpha_array_abs, tf.float32)
- )
- alpha_sum = tf.reduce_sum(input_tensor=alpha_array_abs)
- n = tf.reduce_sum(input_tensor=alpha_array_abs1)
- # alpha = tf.compat.v1.div(alpha_sum, n)
- alpha = tf.math.divide(alpha_sum, n)
- return alpha
-
-
- def flatten_reshape(variable, name='flatten'):
- """Reshapes a high-dimension vector input.
-
- [batch_size, mask_row, mask_col, n_mask] ---> [batch_size, mask_row x mask_col x n_mask]
-
- Parameters
- ----------
- variable : TensorFlow variable or tensor
- The variable or tensor to be flatten.
- name : str
- A unique layer name.
-
- Returns
- -------
- Tensor
- Flatten Tensor
-
- """
- dim = 1
- for d in tl.get_tensor_shape(variable)[1:]: # variable.get_shape()[1:].as_list():
- dim *= d
- return tl.reshape(variable, shape=[-1, dim])
-
-
- def get_collection_trainable(name=''):
- variables = []
- for p in tf.compat.v1.trainable_variables():
- # print(p.name.rpartition('/')[0], self.name)
- if p.name.rpartition('/')[0] == name:
- variables.append(p)
- return variables
-
-
- @deprecated_alias(printable='verbose', end_support_version=1.9) # TODO remove this line for the 1.9 release
- def get_layers_with_name(net, name="", verbose=False):
- """Get a list of layers' output in a network by a given name scope.
-
- Parameters
- -----------
- net : :class:`Layer`
- The last layer of the network.
- name : str
- Get the layers' output that contain this name.
- verbose : boolean
- If True, print information of all the layers' output
-
- Returns
- --------
- list of Tensor
- A list of layers' output (TensorFlow tensor)
-
- Examples
- ---------
- >>> import tensorlayer as tl
- >>> layers = tl.layers.get_layers_with_name(net, "CNN", True)
-
- """
- logging.info(" [*] geting layers with %s" % name)
-
- layers = []
- i = 0
-
- for layer in net.all_layers:
- # logging.info(type(layer.name))
- if name in layer.name:
- layers.append(layer)
-
- if verbose:
- logging.info(" got {:3}: {:15} {}".format(i, layer.name, str(layer.get_shape())))
- i = i + 1
-
- return layers
-
-
- def get_variable_with_initializer(scope_name, var_name, shape, init=tl.initializers.random_normal(), trainable=True):
- # FIXME: documentation needed
- var_name = scope_name + "/" + var_name
- # FIXME: not sure whether this is correct?
- # TODO mindspore weights shape : [out_channel, in_channel, kernel_h, kernel_w]
- if BACKEND == 'mindspore':
- if len(shape) == 2:
- pass
- else:
- shape = shape[::-1]
-
- initial_value = init(shape=shape)
-
- if BACKEND == 'dragon':
- return initial_value
-
- var = tl.Variable(initial_value=initial_value, name=var_name, trainable=trainable)
- return var
-
-
- @deprecated_alias(printable='verbose', end_support_version=1.9) # TODO remove this line for the 1.9 release
- def get_variables_with_name(name=None, train_only=True, verbose=False):
- """Get a list of TensorFlow variables by a given name scope.
-
- Parameters
- ----------
- name : str
- Get the variables that contain this name.
- train_only : boolean
- If Ture, only get the trainable variables.
- verbose : boolean
- If True, print the information of all variables.
-
- Returns
- -------
- list of Tensor
- A list of TensorFlow variables
-
- Examples
- --------
- >>> import tensorlayer as tl
- >>> dense_vars = tl.layers.get_variables_with_name('dense', True, True)
-
- """
- if name is None:
- raise Exception("please input a name")
-
- logging.info(" [*] geting variables with %s" % name)
-
- # tvar = tf.trainable_variables() if train_only else tf.all_variables()
- if train_only:
- t_vars = tf.compat.v1.trainable_variables()
-
- else:
- t_vars = tf.compat.v1.global_variables()
-
- d_vars = [var for var in t_vars if name in var.name]
-
- if verbose:
- for idx, v in enumerate(d_vars):
- logging.info(" got {:3}: {:15} {}".format(idx, v.name, str(v.get_shape())))
-
- return d_vars
-
-
- @deprecated(
- date="2018-09-30", instructions="This API is deprecated in favor of `sess.run(tf.global_variables_initializer())`"
- )
- def initialize_global_variables(sess):
- """Initialize the global variables of TensorFlow.
-
- Run ``sess.run(tf.global_variables_initializer())`` for TF 0.12+ or
- ``sess.run(tf.initialize_all_variables())`` for TF 0.11.
-
- Parameters
- ----------
- sess : Session
- TensorFlow session.
-
- """
- if sess is None:
- raise AssertionError('The session must be defined')
-
- sess.run(tf.compat.v1.global_variables_initializer())
-
-
- def initialize_rnn_state(state, feed_dict=None):
- """Returns the initialized RNN state.
- The inputs are `LSTMStateTuple` or `State` of `RNNCells`, and an optional `feed_dict`.
-
- Parameters
- ----------
- state : RNN state.
- The TensorFlow's RNN state.
- feed_dict : dictionary
- Initial RNN state; if None, returns zero state.
-
- Returns
- -------
- RNN state
- The TensorFlow's RNN state.
-
- """
- if isinstance(state, LSTMStateTuple):
- c = state.c.eval(feed_dict=feed_dict)
- h = state.h.eval(feed_dict=feed_dict)
- return c, h
- else:
- new_state = state.eval(feed_dict=feed_dict)
- return new_state
-
-
- def list_remove_repeat(x):
- """Remove the repeated items in a list, and return the processed list.
- You may need it to create merged layer like Concat, Elementwise and etc.
-
- Parameters
- ----------
- x : list
- Input
-
- Returns
- -------
- list
- A list that after removing it's repeated items
-
- Examples
- -------
- >>> l = [2, 3, 4, 2, 3]
- >>> l = list_remove_repeat(l)
- [2, 3, 4]
-
- """
- y = []
- for i in x:
- if i not in y:
- y.append(i)
-
- return y
-
-
- def merge_networks(layers=None):
- """Merge all parameters, layers and dropout probabilities to a :class:`Layer`.
- The output of return network is the first network in the list.
-
- Parameters
- ----------
- layers : list of :class:`Layer`
- Merge all parameters, layers and dropout probabilities to the first layer in the list.
-
- Returns
- --------
- :class:`Layer`
- The network after merging all parameters, layers and dropout probabilities to the first network in the list.
-
- Examples
- ---------
- >>> import tensorlayer as tl
- >>> n1 = ...
- >>> n2 = ...
- >>> n1 = tl.layers.merge_networks([n1, n2])
-
- """
- if layers is None:
- raise Exception("layers should be a list of TensorLayer's Layers.")
- layer = layers[0]
-
- all_params = []
- all_layers = []
- all_drop = {}
-
- for l in layers:
- all_params.extend(l.all_params)
- all_layers.extend(l.all_layers)
- all_drop.update(l.all_drop)
-
- layer.all_params = list(all_params)
- layer.all_layers = list(all_layers)
- layer.all_drop = dict(all_drop)
-
- layer.all_layers = list_remove_repeat(layer.all_layers)
- layer.all_params = list_remove_repeat(layer.all_params)
-
- return layer
-
-
- def print_all_variables(train_only=False):
- """Print information of trainable or all variables,
- without ``tl.layers.initialize_global_variables(sess)``.
-
- Parameters
- ----------
- train_only : boolean
- Whether print trainable variables only.
- - If True, print the trainable variables.
- - If False, print all variables.
-
- """
- # tvar = tf.trainable_variables() if train_only else tf.all_variables()
- if train_only:
- t_vars = tf.compat.v1.trainable_variables()
- logging.info(" [*] printing trainable variables")
-
- else:
- t_vars = tf.compat.v1.global_variables()
- logging.info(" [*] printing global variables")
-
- for idx, v in enumerate(t_vars):
- logging.info(" var {:3}: {:15} {}".format(idx, str(v.get_shape()), v.name))
-
-
- def quantize(x):
- # ref: https://github.com/AngusG/tensorflow-xnor-bnn/blob/master/models/binary_net.py#L70
- # https://github.com/itayhubara/BinaryNet.tf/blob/master/nnUtils.py
- with tf.compat.v1.get_default_graph().gradient_override_map({"Sign": "TL_Sign_QuantizeGrad"}):
- return tf.sign(x)
-
-
- def quantize_active(x, bitA):
- if bitA == 32:
- return x
- return _quantize_dorefa(x, bitA)
-
-
- def quantize_weight(x, bitW, force_quantization=False):
- G = tf.compat.v1.get_default_graph()
- if bitW == 32 and not force_quantization:
- return x
- if bitW == 1: # BWN
- with G.gradient_override_map({"Sign": "Identity"}):
- E = tf.stop_gradient(tf.reduce_mean(input_tensor=tf.abs(x)))
- return tf.sign(x / E) * E
- x = tf.clip_by_value(x * 0.5 + 0.5, 0.0, 1.0) # it seems as though most weights are within -1 to 1 region anyways
- return 2 * _quantize_dorefa(x, bitW) - 1
-
-
- def quantize_active_overflow(x, bitA):
- if bitA == 32:
- return x
- return _quantize_overflow(x, bitA)
-
-
- def quantize_weight_overflow(x, bitW):
- if bitW == 32:
- return x
- return _quantize_overflow(x, bitW)
-
-
- @deprecated(date="2018-06-30", instructions="TensorLayer relies on TensorFlow to check name reusing")
- def set_name_reuse(enable=True):
- logging.warning('this method is DEPRECATED and has no effect, please remove it from your code.')
-
-
- def ternary_operation(x):
- """Ternary operation use threshold computed with weights."""
- g = tf.compat.v1.get_default_graph()
- with g.gradient_override_map({"Sign": "Identity"}):
- threshold = _compute_threshold(x)
- x = tf.sign(tf.add(tf.sign(tf.add(x, threshold)), tf.sign(tf.add(x, -threshold))))
- return x
-
-
- ########## Module Private Functions ##########
-
-
- @tf.RegisterGradient("TL_Sign_QuantizeGrad")
- def _quantize_grad(op, grad):
- """Clip and binarize tensor using the straight through estimator (STE) for the gradient."""
- return tf.clip_by_value(grad, -1, 1)
-
-
- def _quantize_dorefa(x, k):
- G = tf.compat.v1.get_default_graph()
- n = float(2**k - 1)
- with G.gradient_override_map({"Round": "Identity"}):
- return tf.round(x * n) / n
-
-
- def _quantize_overflow(x, k):
- G = tf.compat.v1.get_default_graph()
- n = float(2**k - 1)
- max_value = tf.reduce_max(input_tensor=x)
- min_value = tf.reduce_min(input_tensor=x)
- with G.gradient_override_map({"Round": "Identity"}):
- step = tf.stop_gradient((max_value - min_value) / n)
- return tf.round((tf.maximum(tf.minimum(x, max_value), min_value) - min_value) / step) * step + min_value
-
-
- def _compute_threshold(x):
- """
- ref: https://github.com/XJTUWYD/TWN
- Computing the threshold.
- """
- x_sum = tf.reduce_sum(input_tensor=tf.abs(x), axis=None, keepdims=False, name=None)
- # threshold = tf.compat.v1.div(x_sum, tf.cast(tf.size(input=x), tf.float32), name=None)
- threshold = tf.math.divide(x_sum, tf.cast(tf.size(input=x), tf.float32), name=None)
- threshold = tf.multiply(0.7, threshold, name=None)
- return threshold
-
-
- def mean_var_with_update(update_moving_mean, update_moving_variance, mean, variance):
- with tf.control_dependencies([update_moving_mean, update_moving_variance]):
- return tf.identity(mean), tf.identity(variance)
-
-
- def w_fold(w, gama, var, epsilon):
- return tf.compat.v1.div(tf.multiply(gama, w), tf.sqrt(var + epsilon))
-
-
- def bias_fold(beta, gama, mean, var, epsilon):
- return tf.subtract(beta, tf.compat.v1.div(tf.multiply(gama, mean), tf.sqrt(var + epsilon)))
|