https://github.com/GPflow/GPflow
Revision ef29b2eb4fcb825cae0630d8a233469ae74ad6e2 authored by Artem Artemev on 20 February 2019, 18:32:20 UTC, committed by Artem Artemev on 20 February 2019, 18:32:20 UTC
1 parent 74434d5
Raw File
Tip revision: ef29b2eb4fcb825cae0630d8a233469ae74ad6e2 authored by Artem Artemev on 20 February 2019, 18:32:20 UTC
Move to TF2.0 preview version.
Tip revision: ef29b2e
base.py
from typing import Optional, Union, List, Tuple
import numpy as np
import tensorflow as tf
import tensorflow_probability as tfp
from .util import default_float


DType = Union[np.dtype, tf.DType]
VariableData = Union[List, Tuple, np.ndarray, int, float]
Transform = tfp.bijectors.Bijector
Prior = tfp.distributions.Distribution


positive = tfp.bijectors.Softplus
triangular = tfp.bijectors.FillTriangular


class Parameter(tf.Variable):
    def __init__(self,
                 value, *,
                 transform: Optional[Transform] = None,
                 prior: Optional[Prior] = None,
                 trainable: bool = True,
                 dtype: DType = None,
                 name: str =None):
        """
        Unconstrained parameter representation.
        According to standart terminology `y` is always transformed representation or,
        in other words, it is constrained version of the parameter. Normally, it is hard
        to operate with unconstrained parameters. For e.g. `variance` cannot be negative,
        therefore we need positive constraint and it is natural to use constrained values.
        """
        if isinstance(value, tf.Variable):
            self._unconstrained = value
        else:
            value = _to_unconstrained(value, transform)
            self._unconstrained = tf.Variable(value, dtype=dtype, name=name, trainable=trainable)

        self.prior = prior
        self._transform = transform

    def log_prior(self):
        x = self.read_value()
        y = self._unconstrained
        dtype = x.dtype

        log_prob = tf.convert_to_tensor(0., dtype=dtype)
        log_det_jacobian = tf.convert_to_tensor(0., dtype=dtype)

        bijector = self.transform
        if self.prior is not None:
            log_prob = self.prior.log_prob(x)
        if self.transform is not None:
            log_det_jacobian = bijector.forward_log_det_jacobian(y, y.shape.ndims)
        return log_prob + log_det_jacobian

    @property
    def handle(self):
        return self._unconstrained.handle

    def value(self):
        return _to_constrained(self._unconstrained.value(), self.transform)

    def read_value(self):
        return _to_constrained(self._unconstrained.read_value(), self.transform)

    @property
    def transform(self):
        return self._transform

    @property
    def trainable(self):
        return self._unconstrained.trainable

    @trainable.setter
    def trainable(self, flag: Union[bool, int]):
        self._unconstrained._trainable = bool(flag)

    @property
    def initial_value(self):
        return self._unconstrained.initial_value

    def assign(self, value, use_locking=False, name=None, read_value=True):
        unconstrained_value = _to_unconstrained_data(value, self.transform)
        self._unconstrained.assign(unconstrained_value,
            read_value=read_value,
            use_locking=use_locking)

    @property
    def name(self):
        return self._unconstrained.name

    @property
    def initializer(self):
        return self._unconstrained.initializer

    @property
    def device(self):
        return self._unconstrained.device

    @property
    def dtype(self):
        return self._unconstrained.dtype

    @property
    def graph(self):
        return self._unconstrained.graph

    @property
    def op(self):
        return self._unconstrained.op

    @property
    def shape(self):
        return self._unconstrained.shape

    def get_shape(self):
        return self.shape

    def _should_act_as_resource_variable(self):
        pass

    def __repr__(self):
        return self.read_value().__repr__()

    def __ilshift__(self, value: VariableData) -> 'Parameter':
        self.assign(value)
        return self


Parameter._OverloadAllOperators()
tf.register_tensor_conversion_function(Parameter, lambda x, *args, **kwds: x.read_value())


def _to_unconstrained_data(data: VariableData, dtype: Optional[DType] = None) -> np.ndarray:
    if isinstance(data, (tf.Tensor, tf.Variable)):
        if dtype is not None and data.dtype != dtype:
            return tf.cast(data, dtype)
        return data
    if dtype is not None and isinstance(dtype, tf.DType):
        dtype = dtype.as_numpy_dtype
    if dtype is None and not isinstance(data, np.ndarray):
        dtype = default_float()
    return np.array(data, dtype=dtype)


def _to_constrained(value: VariableData, transform: Transform) -> tf.Tensor:
    if transform is not None:
        return transform.forward(value)
    return value


def _to_unconstrained(value: VariableData, transform: Transform) -> tf.Tensor:
    if transform is not None:
        return transform.inverse(value)
    return value
back to top