text:
import torch
import os
import numpy as np
import argparse
from utils import relative_to_abs, convert_pixel_to_world, convert_world_to_pixel

parser = argparse.ArgumentParser()
parser.add_argument('--save_dir', default='../output', type=str)
parser.add_argument('--model_dir', default='saved_models', type=str)
parser.add_argument('--homo_dir', default='../data/homography', type=str)
parser.add_argument('--train_dataset', default='trajnet_train', type=str)
parser.add_argument('--test_dataset', default='little3', type=str)
parser.add_argument('--obs_len', default=4, type=int)
parser.add_argument('--pred_len', default=12, type=int)
parser.add_argument('--fps', default=2.5, type=float)
args = parser.parse_args()

def bbox_predict(vlstm_gt, all_obs, args):
    obs_len = args.obs_len
    pred_len = args.pred_len
    seq_len = args.obs_len + args.pred_len

    all_pred = np.zeros((pred_len, all_obs.shape[1], all_obs.shape[2]))
    obs_center = all_obs[:,:,2:4]
    obs_traj = torch.from_numpy(obs_center).type(torch.float)
    obs_traj_rel = torch.zeros(obs_traj.shape, dtype=torch.float)
    obs_traj_rel[1:,:,:] = obs_traj[1:,:,:] - obs_traj[:-1:,:,:]

    pred_traj_rel = vlstm_gt(obs_traj, obs_traj_rel)
    pred_traj = relative_to_abs(pred_traj_rel, obs_traj[-1])
    pred_center = pred_traj.detach().numpy()

    wh = np.repeat(np.mean(all_obs[:,:,4:], axis=0, keepdims=True), pred_len, axis=0)
    all_pred[:, :, 0] = all_obs[[-1], :, 0] + np.expand_dims(np.arange(pred_len), axis=1) + 1
    all_pred[:, :, 1] = np.repeat(all_obs[[0],:,1], pred_len, axis=0)
    all_pred[:, :, 2:4] = pred_center
    all_pred[:, :, 4:6] = wh

    return all_pred

def main(args):
    if not os.path.exists(args.save_dir):
        os.makedirs(args.save_dir)

    vlstm = torch.load(os.path.join(args.model_dir,'{}_vlstm_o{}_p{}.pt'.format(
        args.train_dataset, args.obs_len, args.pred_len)))

    H_path = os.path.join(args.homo_dir, args.test_dataset + '.txt')
    H_mat = np.loadtxt(H_path, delimiter=',')

    ## Load original tracking data
    data_dir = '../output'
    data_path = os.path.join(data_dir, '{}_sort_fps{}.txt'.format(
        args.test_dataset, args.fps))
    save_file_path = os.path.join(args.save_dir, '{}_sort_fps{}_predict.txt'.format(
        args.test_dataset, args.fps))

    track_all = np.loadtxt(data_path, delimiter=',')
    track_center_img = track_all[:,2:4] + 0.5 * track_all[:,4:6]
    track_center_wld = convert_pixel_to_world(track_center_img, H_mat)
    track_all[:, 0] /= (30/args.fps)
    track_all[:, 2:4] = track_center_wld # [frame, id, world_x, world_y, w, h]

    all_frame = np.sort(np.unique(track_all[:, 0])).tolist()
    all_peds = np.unique(track_all[:, 1])
    all_observ = []

    for i in range(len(all_peds)):
        traj_i = track_all[track_all[:,1]==all_peds[i], :]

        # Trajectory is not long enough for prediction
        if traj_i.shape[0] < args.obs_len:
            continue
        num_obs = traj_i.shape[0] - args.obs_len + 1
        obs_i = np.zeros((args.obs_len, num_obs, traj_i.shape[1]))

        for j in range(args.obs_len):
            obs_i[j, :, :] = traj_i[j:num_obs+j, :]
        all_observ.append(obs_i)

    all_obs = np.concatenate(all_observ, axis=1)
    vlstm.eval()
    all_pred = bbox_predict(vlstm, all_obs, args)

    # Swapping axes is for saving the trajectory in order
    pred_all = np.reshape(np.swapaxes(all_pred,0,1), (-1, all_pred.shape[2])) # [frame, id, world_x, world_y, w, h]
    pred_center_wld = pred_all[:,2:4]
    pred_center_img = convert_world_to_pixel(pred_center_wld, H_mat)
    pred_all[:,2:4] = pred_center_img - 0.5 * pred_all[:,4:6] # [frame, id, x1, y1, w, h]
    pred_all[:, 0] *= (30/args.fps)

    np.savetxt(save_file_path, pred_all, fmt='%1.3f', delimiter=',')

if __name__ == '__main__':
    main(args)

max_stars_repo_name: rui-yu-public/retracking-by-prediction

max_issues_repo_name: rui-yu-public/retracking-by-prediction

max_forks_repo_name: rui-yu-public/retracking-by-prediction


text:
import torch
import torchvision
import numpy as np
import PIL
import collections
import random
import cv2
import os

__all__ = ['VideoFilePathToTensor', 'VideoFolderPathToTensor', 'VideoResize', 'VideoRandomCrop', 'VideoCenterCrop', 'VideoRandomHorizontalFlip',
            'VideoRandomVerticalFlip', 'VideoGrayscale']

class VideoFilePathToTensor(object):
    """ load video at given file path to torch.Tensor (C x L x H x W, C = 3)
        It can be composed with torchvision.transforms.Compose().

    Args:
        max_len (int): Maximum output time depth (L <= max_len). Default is None.
            If it is set to None, it will output all frames.
        fps (int): sample frame per seconds. It must lower than or equal the origin video fps.
            Default is None.
        padding_mode (str): Type of padding. Default to None. Only available when max_len is not None.
            - None: won't padding, video length is variable.
            - 'zero': padding the rest empty frames to zeros.
            - 'last': padding the rest empty frames to the last frame.
    """

    def __init__(self, max_len=None, fps=None, padding_mode=None):
        self.max_len = max_len
        self.fps = fps
        assert padding_mode in (None, 'zero', 'last')
        self.padding_mode = padding_mode
        self.channels = 3  # only available to read 3 channels video

    def __call__(self, path):
        """
        Args:
            path (str): path of video file.

        Returns:
            torch.Tensor: Video Tensor (C x L x H x W)
        """

        # open video file
        cap = cv2.VideoCapture(path)
        assert(cap.isOpened())

        # calculate sample_factor to reset fps
        sample_factor = 1
        if self.fps:
            old_fps = cap.get(cv2.CAP_PROP_FPS)  # fps of video
            sample_factor = int(old_fps / self.fps)
            assert(sample_factor >= 1)

        # init empty output frames (C x L x H x W)
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        num_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

        time_len = None
        if self.max_len:
            # time length has upper bound
            if self.padding_mode:
                # padding all video to the same time length
                time_len = self.max_len
            else:
                # video have variable time length
                time_len = min(int(num_frames / sample_factor), self.max_len)
        else:
            # time length is unlimited
            time_len = int(num_frames / sample_factor)

        frames = torch.FloatTensor(self.channels, time_len, height, width)

        for index in range(time_len):
            frame_index = sample_factor * index

            # read frame
            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
            ret, frame = cap.read()
            if ret:
                # successfully read frame
                # BGR to RGB
                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                frame = torch.from_numpy(frame)
                # (H x W x C) to (C x H x W)
                frame = frame.permute(2, 0, 1)
                frames[:, index, :, :] = frame.float()
            else:
                # reach the end of the video
                if self.padding_mode == 'zero':
                    # fill the rest frames with 0.0
                    frames[:, index:, :, :] = 0
                elif self.padding_mode == 'last':
                    # fill the rest frames with the last frame
                    assert(index > 0)
                    frames[:, index:, :, :] = frames[:, index-1, :, :].view(self.channels, 1, height, width)
                break

        frames /= 255
        cap.release()
        return frames

class VideoFolderPathToTensor(object):
    """ load video at given folder path to torch.Tensor (C x L x H x W)
        It can be composed with torchvision.transforms.Compose().

    Args:
        max_len (int): Maximum output time depth (L <= max_len). Default is None.
            If it is set to None, it will output all frames.
        padding_mode (str): Type of padding. Default to None. Only available when max_len is not None.
            - None: won't padding, video length is variable.
            - 'zero': padding the rest empty frames to zeros.
            - 'last': padding the rest empty frames to the last frame.
    """

    def __init__(self, max_len=None, padding_mode=None):
        self.max_len = max_len
        assert padding_mode in (None, 'zero', 'last')
        self.padding_mode = padding_mode

    def __call__(self, path):
        """
        Args:
            path (str): path of video folder.

        Returns:
            torch.Tensor: Video Tensor (C x L x H x W)
        """

        # get video properity
        frames_path = sorted([os.path.join(path,f) for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))])
        frame = cv2.imread(frames_path[0])
        height, width, channels = frame.shape
        num_frames = len(frames_path)

        # init empty output frames (C x L x H x W)
        time_len = None
        if self.max_len:
            # time length has upper bound
            if self.padding_mode:
                # padding all video to the same time length
                time_len = self.max_len
            else:
                # video have variable time length
                time_len = min(num_frames, self.max_len)
        else:
            # time length is unlimited
            time_len = num_frames

        frames = torch.FloatTensor(channels, time_len, height, width)

        # load the video to tensor
        for index in range(time_len):
            if index < num_frames:
                # frame exists
                # read frame
                frame = cv2.imread(frames_path[index])
                # BGR to RGB
                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                frame = torch.from_numpy(frame)
                # (H x W x C) to (C x H x W)
                frame = frame.permute(2, 0, 1)
                frames[:, index, :, :] = frame.float()
            else:
                # reach the end of the video
                if self.padding_mode == 'zero':
                    # fill the rest frames with 0.0
                    frames[:, index:, :, :] = 0
                elif self.padding_mode == 'last':
                    # fill the rest frames with the last frame
                    assert(index > 0)
                    frames[:, index:, :, :] = frames[:, index-1, :, :].view(channels, 1, height, width)
                break

        frames /= 255
        return frames

class VideoResize(object):
    """ resize video tensor (C x L x H x W) to (C x L x h x w)

    Args:
        size (sequence): Desired output size. size is a sequence like (H, W),
            output size will matched to this.
        interpolation (int, optional): Desired interpolation. Default is 'PIL.Image.BILINEAR'
    """

    def __init__(self, size, interpolation=PIL.Image.BILINEAR):
        assert isinstance(size, collections.Iterable) and len(size) == 2
        self.size = size
        self.interpolation = interpolation

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video to be scaled (C x L x H x W)

        Returns:
            torch.Tensor: Rescaled video (C x L x h x w)
        """

        h, w = self.size
        C, L, H, W = video.size()
        rescaled_video = torch.FloatTensor(C, L, h, w)

        # use torchvision implemention to resize video frames
        transform = torchvision.transforms.Compose([
            torchvision.transforms.ToPILImage(),
            torchvision.transforms.Resize(self.size, self.interpolation),
            torchvision.transforms.ToTensor(),
        ])

        for l in range(L):
            frame = video[:, l, :, :]
            frame = transform(frame)
            rescaled_video[:, l, :, :] = frame

        return rescaled_video

    def __repr__(self):
        return self.__class__.__name__ + '(size={0})'.format(self.size)

class VideoRandomCrop(object):
    """ Crop the given Video Tensor (C x L x H x W) at a random location.

    Args:
        size (sequence): Desired output size like (h, w).
    """

    def __init__(self, size):
        assert len(size) == 2
        self.size = size

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video (C x L x H x W) to be cropped.

        Returns:
            torch.Tensor: Cropped video (C x L x h x w).
        """

        H, W = video.size()[2:]
        h, w = self.size
        assert H >= h and W >= w

        top = np.random.randint(0, H - h)
        left = np.random.randint(0, W - w)

        video = video[:, :, top : top + h, left : left + w]

        return video

class VideoCenterCrop(object):
    """ Crops the given video tensor (C x L x H x W) at the center.

    Args:
        size (sequence): Desired output size of the crop like (h, w).
    """

    def __init__(self, size):
        self.size = size

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video (C x L x H x W) to be cropped.

        Returns:
            torch.Tensor: Cropped Video (C x L x h x w).
        """

        H, W = video.size()[2:]
        h, w = self.size
        assert H >= h and W >= w

        top = int((H - h) / 2)
        left = int((W - w) / 2)

        video = video[:, :, top : top + h, left : left + w]

        return video

class VideoRandomHorizontalFlip(object):
    """ Horizontal flip the given video tensor (C x L x H x W) randomly with a given probability.

    Args:
        p (float): probability of the video being flipped. Default value is 0.5.
    """

    def __init__(self, p=0.5):
        self.p = p

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video to flipped.

        Returns:
            torch.Tensor: Randomly flipped video.
        """

        if random.random() < self.p:
            # horizontal flip the video
            video = video.flip([3])

        return video

class VideoRandomVerticalFlip(object):
    """ Vertical flip the given video tensor (C x L x H x W) randomly with a given probability.

    Args:
        p (float): probability of the video being flipped. Default value is 0.5.
    """

    def __init__(self, p=0.5):
        self.p = p

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video to flipped.

        Returns:
            torch.Tensor: Randomly flipped video.
        """

        if random.random() < self.p:
            # horizontal flip the video
            video = video.flip([2])

        return video

class VideoGrayscale(object):
    """ Convert video (C x L x H x W) to grayscale (C' x L x H x W, C' = 1 or 3)

    Args:
        num_output_channels (int): (1 or 3) number of channels desired for output video
    """

    def __init__(self, num_output_channels=1):
        assert num_output_channels in (1, 3)
        self.num_output_channels = num_output_channels

    def __call__(self, video):
        """
        Args:
            video (torch.Tensor): Video (3 x L x H x W) to be converted to grayscale.

        Returns:
            torch.Tensor: Grayscaled video (1 x L x H x W  or  3 x L x H x W)
        """

        C, L, H, W = video.size()
        grayscaled_video = torch.FloatTensor(self.num_output_channels, L, H, W)

        # use torchvision implemention to convert video frames to gray scale
        transform = torchvision.transforms.Compose([
            torchvision.transforms.ToPILImage(),
            torchvision.transforms.Grayscale(self.num_output_channels),
            torchvision.transforms.ToTensor(),
        ])

        for l in range(L):
            frame = video[:, l, :, :]
            frame = transform(frame)
            grayscaled_video[:, l, :, :] = frame

        return grayscaled_video

ext: py

max_stars_repo_name: aakgun/pytorch-VideoDataset

max_issues_repo_name: mbencherif/pytorch-VideoDataset

max_forks_repo_name: mbencherif/pytorch-VideoDataset

reason: import numpy


text:
from __future__ import division

import argparse
from datetime import datetime
import json
import os

import librosa
import numpy as np
import tensorflow as tf

from wavenet import WaveNet
from wavenet_ops import mu_law_decode, mu_law_encode

SAMPLES = 16000
LOGDIR = './logdir'
WINDOW = 8000
WAVENET_PARAMS = './wavenet_params.json'
SAVE_EVERY = None

def get_arguments():
    parser = argparse.ArgumentParser(description='WaveNet generation script')
    parser.add_argument(
        'checkpoint', type=str, help='Which model checkpoint to generate from')
    parser.add_argument(
        '--samples',
        type=int,
        default=SAMPLES,
        help='How many waveform samples to generate')
    parser.add_argument(
        '--logdir',
        type=str,
        default=LOGDIR,
        help='Directory in which to store the logging '
        'information for TensorBoard.')
    parser.add_argument(
        '--window',
        type=int,
        default=WINDOW,
        help='The number of past samples to take into '
        'account at each step')
    parser.add_argument(
        '--wavenet_params',
        type=str,
        default=WAVENET_PARAMS,
        help='JSON file with the network parameters')
    parser.add_argument(
        '--wav_out_path',
        type=str,
        default=None,
        help='Path to output wav file')
    parser.add_argument(
        '--save_every',
        type=int,
        default=SAVE_EVERY,
        help='How many samples before saving in-progress wav')
    parser.add_argument(
        '--fast_generation',
        type=bool,
        default=True,
        help='Use fast generation')
    parser.add_argument(
        '--wav_seed',
        type=str,
        default=None,
        help='The wav file to start generation from')
    return parser.parse_args()

def write_wav(waveform, sample_rate, filename):
    y = np.array(waveform)
    librosa.output.write_wav(filename, y, sample_rate)
    print('Updated wav file at {}'.format(filename))

def create_seed(filename, sample_rate, quantization_channels, window_size=WINDOW):
    audio, _ = librosa.load(filename, sr=sample_rate, mono=True)

    quantized = mu_law_encode(audio, quantization_channels)
    cut_index = tf.size(quantized) + tf.constant(window_size) - tf.constant(1)

    return quantized[:cut_index]

def main():
    args = get_arguments()
    logdir = os.path.join(args.logdir, 'train', str(datetime.now()))
    with open(args.wavenet_params, 'r') as config_file:
        wavenet_params = json.load(config_file)

    sess = tf.Session()

    net = WaveNet(
        batch_size=1,
        dilations=wavenet_params['dilations'],
        filter_width=wavenet_params['filter_width'],
        residual_channels=wavenet_params['residual_channels'],
        dilation_channels=wavenet_params['dilation_channels'],
        quantization_channels=wavenet_params['quantization_channels'],
        skip_channels=wavenet_params['skip_channels'],
        use_biases=wavenet_params['use_biases'],
        fast_generation=args.fast_generation)

    samples = tf.placeholder(tf.int32)

    next_sample = net.predict_proba(samples)

    if args.fast_generation:
        sess.run(tf.initialize_all_variables())
        sess.run(net.init_ops)

    variables_to_restore = {var.name[:-2]: var for var in tf.all_variables(
    ) if not ('state_buffer' in var.name or 'pointer' in var.name)}
    saver = tf.train.Saver(variables_to_restore)

    print('Restoring model from {}'.format(args.checkpoint))
    saver.restore(sess, args.checkpoint)

    decode = mu_law_decode(samples, wavenet_params['quantization_channels'])

    quantization_channels = wavenet_params['quantization_channels']
    if args.wav_seed:
        seed = create_seed(args.wav_seed,
                    wavenet_params['sample_rate'],
                    quantization_channels)
        waveform = sess.run(seed).tolist()
    else:
        waveform = np.random.randint(quantization_channels, size=(1,)).tolist()

    for step in range(args.samples):
        if args.fast_generation:
            window = waveform[-1]
            outputs = [next_sample]
            outputs.extend(net.push_ops)
        else:
            if len(waveform) > args.window:
                window = waveform[-args.window:]
            else:
                window = waveform
            outputs = [next_sample]

        prediction = sess.run(outputs, feed_dict={samples: window})[0]
        sample = np.random.choice(
            np.arange(quantization_channels), p=prediction)
        waveform.append(sample)
        print('Sample {:3<d}/{:3<d}: {}'
              .format(step + 1, args.samples, sample))

        if (args.wav_out_path and args.save_every and
                (step + 1) % args.save_every == 0):

            out = sess.run(decode, feed_dict={samples: waveform})
            write_wav(out, wavenet_params['sample_rate'], args.wav_out_path)

    datestring = str(datetime.now()).replace(' ', 'T')
    writer = tf.train.SummaryWriter(
        os.path.join(logdir, 'generation', datestring))
    tf.audio_summary('generated', decode, wavenet_params['sample_rate'])
    summaries = tf.merge_all_summaries()

    summary_out = sess.run(summaries,
                           feed_dict={
                               samples: np.reshape(waveform, [-1, 1])
                           })
    writer.add_summary(summary_out)

    if args.wav_out_path:
        out = sess.run(decode, feed_dict={samples: waveform})
        write_wav(out, wavenet_params['sample_rate'], args.wav_out_path)

    print('Finished generating. The result can be viewed in TensorBoard.')

if __name__ == '__main__':
    main()

max_stars_repo_name: amikey/tensorflow-wavenet

max_issues_repo_name: amikey/tensorflow-wavenet

max_forks_repo_name: amikey/tensorflow-wavenet


text:
from math import sqrt

from numpy import (array, unravel_index, nditer, linalg, random, subtract,
                   power, exp, pi, zeros, arange, outer, meshgrid, dot)
from collections import defaultdict
from warnings import warn

"""
    minimalistic implementation of the self organizing maps (som).

    giuseppe vettigli 2013.
"""

def fast_norm(x):
    """returns norm-2 of a 1-d numpy array.

    * faster than linalg.norm in case of 1-d arrays (numpy 1.9.2rc1).
    """
    return sqrt(dot(x, x.T))

class MiniSom:
    def __init__(self, x, y, input_len, sigma=1.0,
                 learning_rate=0.5, random_seed=None):
        """
            initializes a self organizing maps.
            x,y - dimensions of the som
            input_len - number of the elements of the vectors in input
            sigma - spread of the neighborhood function (gaussian), needs to be adequate to the dimensions of the map.
            (at the iteration t we have sigma(t) = sigma / (1 + t/t) where t is #num_iteration/2)
            learning_rate - initial learning rate
            (at the iteration t we have learning_rate(t) = learning_rate / (1 + t/t) where t is #num_iteration/2)
            random_seed, random seed to use.
        """
        if sigma >= x / 2.0 or sigma >= y / 2.0:
            warn('warning: sigma is too high for the dimension of the map.')
        random.seed(seed=random_seed)
        self.random_generator = random
        self.learning_rate = learning_rate
        self.sigma = sigma
        # random initialization
        self.weights = self.random_generator.rand(x, y, input_len) * 2 - 1
        # normalization
        self.weights = array([v / linalg.norm(v) for v in self.weights])
        self.activation_map = zeros((x, y))
        self.neigx = arange(x)
        self.neigy = arange(y)  # used to evaluate the neighborhood function
        self.neighborhood = self.gaussian

    def _activate(self, x):
        """ updates matrix activation_map, in this matrix the element i,j is the response of the neuron i,j to x """
        s = subtract(x, self.weights)  # x - w
        it = nditer(self.activation_map, flags=['multi_index'])
        while not it.finished:
            self.activation_map[it.multi_index] = fast_norm(s[it.multi_index])  # || x - w ||
            it.iternext()

    def activate(self, x):
        """ returns the activation map to x """
        self._activate(x)
        return self.activation_map

    def gaussian(self, c, sigma):
        """ returns a gaussian centered in c """
        d = 2 * pi * sigma * sigma
        ax = exp(-power(self.neigx - c[0], 2) / d)
        ay = exp(-power(self.neigy - c[1], 2) / d)
        return outer(ax, ay)  # the external product gives a matrix

    def diff_gaussian(self, c, sigma):
        """ mexican hat centered in c (unused) """
        xx, yy = meshgrid(self.neigx, self.neigy)
        p = power(xx - c[0], 2) + power(yy - c[1], 2)
        d = 2 * pi * sigma * sigma
        return exp(-(p) / d) * (1 - 2 / d * p)

    def winner(self, x):
        """ computes the coordinates of the winning neuron for the sample x """
        self._activate(x)
        return unravel_index(self.activation_map.argmin(), self.activation_map.shape)

    def update(self, x, win, t):
        """
            updates the weights of the neurons.
            x - current pattern to learn
            win - position of the winning neuron for x (array or tuple).
            t - iteration index
        """
        # eta(t) = eta(0) / (1 + t/t)
        # keeps the learning rate nearly constant for the first t iterations and then adjusts it
        eta = self.learning_rate / (1 + t / self.t)
        sig = self.sigma/(1 + t / self.t)  # sigma and learning rate decrease with the same rule
        g = self.neighborhood(win, sig) * eta  # improves the performances
        it = nditer(g, flags=['multi_index'])
        while not it.finished:
            # eta * neighborhood_function * (x-w)
            self.weights[it.multi_index] += g[it.multi_index] * (x-self.weights[it.multi_index])
            # normalization
            self.weights[it.multi_index] = self.weights[it.multi_index] / fast_norm(self.weights[it.multi_index])
            it.iternext()

    def quantization(self, data):
        """ assigns a code book (weights vector of the winning neuron) to each sample in data. """
        q = zeros(data.shape)
        for i, x in enumerate(data):
            q[i] = self.weights[self.winner(x)]
        return q

    def random_weights_init(self, data):
        """ initializes the weights of the som picking random samples from data """
        it = nditer(self.activation_map, flags=['multi_index'])
        while not it.finished:
            self.weights[it.multi_index] = data[int(self.random_generator.rand() * len(data) - 1)]
            self.weights[it.multi_index] = self.weights[it.multi_index] / fast_norm(self.weights[it.multi_index])
            it.iternext()

    def train_random(self, data, num_iteration):
        """ trains the som picking samples at random from data """
        self._init_t(num_iteration)
        for iteration in range(num_iteration):
            rand_i = int(round(self.random_generator.rand() * len(data) - 1))  # pick a random sample
            self.update(data[rand_i], self.winner(data[rand_i]), iteration)

    def train_batch(self, data, num_iteration):
        """ trains using all the vectors in data sequentially """
        self._init_t(len(data) * num_iteration)
        iteration = 0
        while iteration < num_iteration:
            idx = iteration % (len(data) - 1)
            self.update(data[idx], self.winner(data[idx]), iteration)
            iteration += 1

    def _init_t(self, num_iteration):
        """ initializes the parameter t needed to adjust the learning rate """
        self.t = num_iteration / 2  # keeps the learning rate nearly constant for the first half of the iterations

    def distance_map(self):
        """ returns the average distance map of the weights.
            (each mean is normalized in order to sum up to 1) """
        um = zeros((self.weights.shape[0], self.weights.shape[1]))
        it = nditer(um, flags=['multi_index'])
        while not it.finished:
            for ii in range(it.multi_index[0] - 1, it.multi_index[0] + 2):
                for jj in range(it.multi_index[1] - 1, it.multi_index[1] + 2):
                    if ii >= 0 and ii < self.weights.shape[0] and jj >= 0 and jj < self.weights.shape[1]:
                        um[it.multi_index] += fast_norm(self.weights[ii, jj, :] - self.weights[it.multi_index])
            it.iternext()
        um = um / um.max()
        return um

    def activation_response(self, data):
        """
            returns a matrix where the element i,j is the number of times
            that the neuron i,j have been winner.
        """
        a = zeros((self.weights.shape[0], self.weights.shape[1]))
        for x in data:
            a[self.winner(x)] += 1
        return a

    def quantization_error(self, data):
        """
            returns the quantization error computed as the average distance between
            each input sample and its best matching unit.
        """
        error = 0
        for x in data:
            error += fast_norm(x - self.weights[self.winner(x)])
        return error / len(data)

    def win_map(self, data):
        """
            returns a dictionary wm where wm[(i,j)] is a list with all the patterns
            that have been mapped in the position i,j.
        """
        winmap = defaultdict(list)
        for x in data:
            winmap[self.winner(x)].append(x)
        return winmap

# unit tests
from numpy.testing import assert_almost_equal, assert_array_almost_equal, assert_array_equal

class TestMinisom:
    def setup_method(self, method):
        self.som = MiniSom(5, 5, 1)
        for w in self.som.weights:  # checking weights normalization
            assert_almost_equal(1.0, linalg.norm(w))
        self.som.weights = zeros((5, 5))  # fake weights
        self.som.weights[2, 3] = 5.0
        self.som.weights[1, 1] = 2.0

    def test_fast_norm(self):
        assert fast_norm(array([1, 3])) == sqrt(1 + 9)

    def test_gaussian(self):
        bell = self.som.gaussian((2, 2), 1)
        assert bell.max() == 1.0
        assert bell.argmax() == 12  # unravel(12) = (2,2)

    def test_win_map(self):
        winners = self.som.win_map([5.0, 2.0])
        assert winners[(2, 3)][0] == 5.0
        assert winners[(1, 1)][0] == 2.0

    def test_activation_reponse(self):
        response = self.som.activation_response([5.0, 2.0])
        assert response[2, 3] == 1
        assert response[1, 1] == 1

    def test_activate(self):
        assert self.som.activate(5.0).argmin() == 13.0  # unravel(13) = (2,3)

    def test_quantization_error(self):
        self.som.quantization_error([5, 2]) == 0.0
        self.som.quantization_error([4, 1]) == 0.5

    def test_quantization(self):
        q = self.som.quantization(array([4, 2]))
        assert q[0] == 5.0
        assert q[1] == 2.0

    def test_random_seed(self):
        som1 = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        som2 = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        assert_array_almost_equal(som1.weights, som2.weights)  # same initialization
        data = random.rand(100, 2)
        som1 = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        som1.train_random(data, 10)
        som2 = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        som2.train_random(data, 10)
        assert_array_almost_equal(som1.weights, som2.weights)  # same state after training

    def test_train_batch(self):
        som = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        data = array([[4, 2], [3, 1]])
        q1 = som.quantization_error(data)
        som.train_batch(data, 10)
        assert q1 > som.quantization_error(data)

    def test_train_random(self):
        som = MiniSom(5, 5, 2, sigma=1.0, learning_rate=0.5, random_seed=1)
        data = array([[4, 2], [3, 1]])
        q1 = som.quantization_error(data)
        som.train_random(data, 10)
        assert q1 > som.quantization_error(data)

    def test_random_weights_init(self):
        som = MiniSom(2, 2, 2, random_seed=1)
        som.random_weights_init(array([[1.0, .0]]))
        for w in som.weights:
            assert_array_equal(w[0], array([1.0, .0]))

max_stars_repo_name: xiwenc/cbir-invenio

max_issues_repo_name: xiwenc/cbir-invenio

max_forks_repo_name: xiwenc/cbir-invenio

reason: from numpy


text:
#!/usr/bin/env/python
"""
Usage:
    make_dataset.py [options]

Options:
    -h --help           Show this screen.
    --dataset NAME      QM9 or ZINC
"""

import json
import os
import sys

import numpy as np
from docopt import docopt
from rdkit import Chem
from rdkit.Chem import QED

import utils

sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))

# get current directory in order to work with full path and not dynamic
current_dir = os.path.dirname(os.path.realpath(__file__))

def readStr_qm9():
    f = open(current_dir + '/qm9.smi', 'r')
    L = []
    for line in f:
        line = line.strip()
        L.append(line)
    f.close()
    np.random.seed(1)
    np.random.shuffle(L)
    return L

def read_zinc():
    f = open(current_dir + '/zinc.smi', 'r')
    L = []
    for line in f:
        line = line.strip()
        L.append(line)
    f.close()
    return L

def train_valid_split(dataset):
    n_mol_out = 0
    n_test = 5000
    test_idx = np.arange(0, n_test)
    valid_idx = np.random.randint(n_test, high=len(dataset), size=round(len(dataset) * 0.1))

    # save the train, valid dataset.
    raw_data = {'train': [], 'valid': [], 'test': []}
    file_count = 0
    for i, smiles in enumerate(dataset):
        val = QED.qed(Chem.MolFromSmiles(smiles))
        hist = make_hist(smiles)
        if hist is not None:
            if i in valid_idx:
                raw_data['valid'].append({'smiles': smiles, 'QED': val, 'hist': hist.tolist()})
            elif i in test_idx:
                raw_data['test'].append({'smiles': smiles, 'QED': val, 'hist': hist.tolist()})
            else:
                raw_data['train'].append({'smiles': smiles, 'QED': val, 'hist': hist.tolist()})
            file_count += 1
            if file_count % 1000 == 0:
                print('Finished reading: %d' % file_count, end='\r')
        else:
            n_mol_out += 1

    print("Number of molecules left out: ", n_mol_out)
    return raw_data

def make_hist(smiles):
    mol = Chem.MolFromSmiles(smiles)
    atoms = mol.GetAtoms()
    hist = np.zeros(utils.dataset_info(dataset)['hist_dim'])
    for atom in atoms:
        if dataset == 'qm9':
            atom_str = atom.GetSymbol()
        else:
            # zinc dataset # transform using "<atom_symbol><valence>(<charge>)"  notation
            symbol = atom.GetSymbol()
            valence = atom.GetTotalValence()
            charge = atom.GetFormalCharge()
            atom_str = "%s%i(%i)" % (symbol, valence, charge)

            if atom_str not in utils.dataset_info(dataset)['atom_types']:
                print('Unrecognized atom type %s' % atom_str)
                return None

        ind = utils.dataset_info(dataset)['atom_types'].index(atom_str)
        val = utils.dataset_info(dataset)['maximum_valence'][ind]
        hist[val - 1] += 1  # in the array the valence number start from 1, instead the array start from 0
    return hist

def preprocess(raw_data, dataset):
    print('Parsing smiles as graphs...')
    processed_data = {'train': [], 'valid': [], 'test': []}

    file_count = 0
    for section in ['train', 'valid', 'test']:
        all_smiles = []  # record all smiles in training dataset
        for i, (smiles, QED, hist) in enumerate([(mol['smiles'], mol['QED'], mol['hist'])
                                                 for mol in raw_data[section]]):
            nodes, edges = utils.to_graph(smiles, dataset)
            if len(edges) <= 0:
                print('Error. Molecule with len(edges) <= 0')
                continue
            processed_data[section].append({
                'targets': [[QED]],
                'graph': edges,
                'node_features': nodes,
                'smiles': smiles,
                'hist': hist
            })
            all_smiles.append(smiles)
            if file_count % 1000 == 0:
                print('Finished processing: %d' % file_count, end='\r')
            file_count += 1
        print('%s: 100 %%                   ' % (section))
        with open('molecules_%s_%s.json' % (section, dataset), 'w') as f:
            json.dump(processed_data[section], f)

    print("Train molecules = " + str(len(processed_data['train'])))
    print("Valid molecules = " + str(len(processed_data['valid'])))
    print("Test molecules = " + str(len(processed_data['test'])))

if __name__ == "__main__":
    args = docopt(__doc__)
    dataset = args.get('--dataset')

    print('Reading dataset: ' + str(dataset))
    data = []
    if dataset == 'qm9':
        data = readStr_qm9()
    elif dataset == 'zinc':
        data = read_zinc()
    else:
        print('Error. The database doesn\'t exist')
        exit(1)

    raw_data = train_valid_split(data)
    preprocess(raw_data, dataset)

max_stars_repo_name: drigoni/ConditionalCGVAE

max_issues_repo_name: drigoni/ConditionalCGVAE

max_forks_repo_name: drigoni/ConditionalCGVAE


text:
# Copyright 2016 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Tests for Keras metrics functions."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import json

from absl.testing import parameterized
import numpy as np

from tensorflow.python.framework import constant_op
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import test_util
from tensorflow.python.keras import metrics
from tensorflow.python.keras.utils import metrics_utils
from tensorflow.python.ops import math_ops
from tensorflow.python.ops import random_ops
from tensorflow.python.ops import variables
from tensorflow.python.platform import test

@test_util.run_all_in_graph_and_eager_modes
class FalsePositivesTest(test.TestCase):

  def test_config(self):
    fp_obj = metrics.FalsePositives(name='my_fp', thresholds=[0.4, 0.9])
    self.assertEqual(fp_obj.name, 'my_fp')
    self.assertEqual(len(fp_obj.variables), 1)
    self.assertEqual(fp_obj.thresholds, [0.4, 0.9])

    # Check save and restore config
    fp_obj2 = metrics.FalsePositives.from_config(fp_obj.get_config())
    self.assertEqual(fp_obj2.name, 'my_fp')
    self.assertEqual(len(fp_obj2.variables), 1)
    self.assertEqual(fp_obj2.thresholds, [0.4, 0.9])

  def test_unweighted(self):
    fp_obj = metrics.FalsePositives()
    self.evaluate(variables.variables_initializer(fp_obj.variables))

    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))

    update_op = fp_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = fp_obj.result()
    self.assertAllClose(7., result)

  def test_weighted(self):
    fp_obj = metrics.FalsePositives()
    self.evaluate(variables.variables_initializer(fp_obj.variables))
    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))
    sample_weight = constant_op.constant((1., 1.5, 2., 2.5))
    result = fp_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose(14., self.evaluate(result))

  def test_unweighted_with_thresholds(self):
    fp_obj = metrics.FalsePositives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(fp_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))

    update_op = fp_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = fp_obj.result()
    self.assertAllClose([7., 4., 2.], result)

  def test_weighted_with_thresholds(self):
    fp_obj = metrics.FalsePositives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(fp_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))
    sample_weight = ((1.0, 2.0, 3.0, 5.0), (7.0, 11.0, 13.0, 17.0),
                     (19.0, 23.0, 29.0, 31.0), (5.0, 15.0, 10.0, 0))

    result = fp_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose([125., 42., 12.], self.evaluate(result))

  def test_threshold_limit(self):
    with self.assertRaisesRegexp(
        ValueError,
        r'Threshold values must be in \[0, 1\]. Invalid values: \[-1, 2\]'):
      metrics.FalsePositives(thresholds=[-1, 0.5, 2])

    with self.assertRaisesRegexp(
        ValueError,
        r'Threshold values must be in \[0, 1\]. Invalid values: \[None\]'):
      metrics.FalsePositives(thresholds=[None])

@test_util.run_all_in_graph_and_eager_modes
class FalseNegativesTest(test.TestCase):

  def test_config(self):
    fn_obj = metrics.FalseNegatives(name='my_fn', thresholds=[0.4, 0.9])
    self.assertEqual(fn_obj.name, 'my_fn')
    self.assertEqual(len(fn_obj.variables), 1)
    self.assertEqual(fn_obj.thresholds, [0.4, 0.9])

    # Check save and restore config
    fn_obj2 = metrics.FalseNegatives.from_config(fn_obj.get_config())
    self.assertEqual(fn_obj2.name, 'my_fn')
    self.assertEqual(len(fn_obj2.variables), 1)
    self.assertEqual(fn_obj2.thresholds, [0.4, 0.9])

  def test_unweighted(self):
    fn_obj = metrics.FalseNegatives()
    self.evaluate(variables.variables_initializer(fn_obj.variables))

    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))

    update_op = fn_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = fn_obj.result()
    self.assertAllClose(3., result)

  def test_weighted(self):
    fn_obj = metrics.FalseNegatives()
    self.evaluate(variables.variables_initializer(fn_obj.variables))
    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))
    sample_weight = constant_op.constant((1., 1.5, 2., 2.5))
    result = fn_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose(5., self.evaluate(result))

  def test_unweighted_with_thresholds(self):
    fn_obj = metrics.FalseNegatives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(fn_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))

    update_op = fn_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = fn_obj.result()
    self.assertAllClose([1., 4., 6.], result)

  def test_weighted_with_thresholds(self):
    fn_obj = metrics.FalseNegatives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(fn_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))
    sample_weight = ((3.0,), (5.0,), (7.0,), (4.0,))

    result = fn_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose([4., 16., 23.], self.evaluate(result))

@test_util.run_all_in_graph_and_eager_modes
class TrueNegativesTest(test.TestCase):

  def test_config(self):
    tn_obj = metrics.TrueNegatives(name='my_tn', thresholds=[0.4, 0.9])
    self.assertEqual(tn_obj.name, 'my_tn')
    self.assertEqual(len(tn_obj.variables), 1)
    self.assertEqual(tn_obj.thresholds, [0.4, 0.9])

    # Check save and restore config
    tn_obj2 = metrics.TrueNegatives.from_config(tn_obj.get_config())
    self.assertEqual(tn_obj2.name, 'my_tn')
    self.assertEqual(len(tn_obj2.variables), 1)
    self.assertEqual(tn_obj2.thresholds, [0.4, 0.9])

  def test_unweighted(self):
    tn_obj = metrics.TrueNegatives()
    self.evaluate(variables.variables_initializer(tn_obj.variables))

    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))

    update_op = tn_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = tn_obj.result()
    self.assertAllClose(3., result)

  def test_weighted(self):
    tn_obj = metrics.TrueNegatives()
    self.evaluate(variables.variables_initializer(tn_obj.variables))
    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))
    sample_weight = constant_op.constant((1., 1.5, 2., 2.5))
    result = tn_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose(4., self.evaluate(result))

  def test_unweighted_with_thresholds(self):
    tn_obj = metrics.TrueNegatives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(tn_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))

    update_op = tn_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = tn_obj.result()
    self.assertAllClose([2., 5., 7.], result)

  def test_weighted_with_thresholds(self):
    tn_obj = metrics.TrueNegatives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(tn_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))
    sample_weight = ((0.0, 2.0, 3.0, 5.0),)

    result = tn_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose([5., 15., 23.], self.evaluate(result))

@test_util.run_all_in_graph_and_eager_modes
class TruePositivesTest(test.TestCase):

  def test_config(self):
    tp_obj = metrics.TruePositives(name='my_tp', thresholds=[0.4, 0.9])
    self.assertEqual(tp_obj.name, 'my_tp')
    self.assertEqual(len(tp_obj.variables), 1)
    self.assertEqual(tp_obj.thresholds, [0.4, 0.9])

    # Check save and restore config
    tp_obj2 = metrics.TruePositives.from_config(tp_obj.get_config())
    self.assertEqual(tp_obj2.name, 'my_tp')
    self.assertEqual(len(tp_obj2.variables), 1)
    self.assertEqual(tp_obj2.thresholds, [0.4, 0.9])

  def test_unweighted(self):
    tp_obj = metrics.TruePositives()
    self.evaluate(variables.variables_initializer(tp_obj.variables))

    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))

    update_op = tp_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = tp_obj.result()
    self.assertAllClose(7., result)

  def test_weighted(self):
    tp_obj = metrics.TruePositives()
    self.evaluate(variables.variables_initializer(tp_obj.variables))
    y_true = constant_op.constant(((0, 1, 0, 1, 0), (0, 0, 1, 1, 1),
                                   (1, 1, 1, 1, 0), (0, 0, 0, 0, 1)))
    y_pred = constant_op.constant(((0, 0, 1, 1, 0), (1, 1, 1, 1, 1),
                                   (0, 1, 0, 1, 0), (1, 1, 1, 1, 1)))
    sample_weight = constant_op.constant((1., 1.5, 2., 2.5))
    result = tp_obj(y_true, y_pred, sample_weight=sample_weight)
    self.assertAllClose(12., self.evaluate(result))

  def test_unweighted_with_thresholds(self):
    tp_obj = metrics.TruePositives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(tp_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))

    update_op = tp_obj.update_state(y_true, y_pred)
    self.evaluate(update_op)
    result = tp_obj.result()
    self.assertAllClose([6., 3., 1.], result)

  def test_weighted_with_thresholds(self):
    tp_obj = metrics.TruePositives(thresholds=[0.15, 0.5, 0.85])
    self.evaluate(variables.variables_initializer(tp_obj.variables))

    y_pred = constant_op.constant(((0.9, 0.2, 0.8, 0.1), (0.2, 0.9, 0.7, 0.6),
                                   (0.1, 0.2, 0.4, 0.3), (0, 1, 0.7, 0.3)))
    y_true = constant_op.constant(((0, 1, 1, 0), (1, 0, 0, 0), (0, 0, 0, 0),
                                   (1, 1, 1, 1)))

    result = tp_obj(y_true, y_pred, sample_weight=37.)
    self.assertAllClose([222., 111., 37.], self.evaluate(result))

@test_util.run_all_in_graph_and_eager_modes
class PrecisionTest(test.TestCase):

  def test_config(self):
    p_obj = metrics.Precision(
        name='my_precision', thresholds=[0.4, 0.9], top_k=15, class_id=12)
    self.assertEqual(p_obj.name, 'my_precision')
    self.assertEqual(len(p_obj.variables), 2)
    self.assertEqual([v.name for v in p_obj.variables],
                     ['true_positives:0', 'false_positives:0'])
    self.assertEqual(p_obj.thresholds, [0.4, 0.9])
    self.assertEqual(p_obj.top_k, 15)
    self.assertEqual(p_obj.class_id, 12)

    # Check save and restore config
    p_obj2 = metrics.Precision.from_config(p_obj.get_config())
    self.assertEqual(p_obj2.name, 'my_precision')
    self.assertEqual(len(p_obj2.variables), 2)
    self.assertEqual(p_obj2.thresholds, [0.4, 0.9])
    self.assertEqual(p_obj2.top_k, 15)
    self.assertEqual(p_obj2.class_id, 12)

  def test_value_is_idempotent(self):
    p_obj = metrics.Precision(thresholds=[0.3, 0.72])
    y_pred = random_ops.random_uniform(shape=(10, 3))
    y_true = random_ops.random_uniform(shape=(10, 3))
    update_op = p_obj.update_state(y_true, y_pred)
    self.evaluate(variables.variables_initializer(p_obj.variables))

    # Run several updates.
    for _ in range(10):
      self.evaluate(update_op)

    # Then verify idempotency.
    initial_precision = self.evaluate(p_obj.result())
    for _ in range(10):
      self.assertArrayNear(initial_precision, self.evaluate(p_obj.result()),
                           1e-3)

  def test_unweighted(self):
    p_obj = metrics.Precision()
    y_pred = constant_op.constant([1, 0, 1, 0], shape=(1, 4))
    y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))

  def test_unweighted_all_incorrect(self):
    p_obj = metrics.Precision(thresholds=[0.5])
    inputs = np.random.randint(0, 2, size=(100, 1))
    y_pred = constant_op.constant(inputs)
    y_true = constant_op.constant(1 - inputs)
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(0, self.evaluate(result))

  def test_weighted(self):
    p_obj = metrics.Precision()
    y_pred = constant_op.constant([[1, 0, 1, 0], [1, 0, 1, 0]])
    y_true = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(
        y_true,
        y_pred,
        sample_weight=constant_op.constant([[1, 2, 3, 4], [4, 3, 2, 1]]))
    weighted_tp = 3.0 + 4.0
    weighted_positives = (1.0 + 3.0) + (4.0 + 2.0)
    expected_precision = weighted_tp / weighted_positives
    self.assertAlmostEqual(expected_precision, self.evaluate(result))

  def test_div_by_zero(self):
    p_obj = metrics.Precision()
    y_pred = constant_op.constant([0, 0, 0, 0])
    y_true = constant_op.constant([0, 0, 0, 0])
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred)
    self.assertEqual(0, self.evaluate(result))

  def test_unweighted_with_threshold(self):
    p_obj = metrics.Precision(thresholds=[0.5, 0.7])
    y_pred = constant_op.constant([1, 0, 0.6, 0], shape=(1, 4))
    y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred)
    self.assertArrayNear([0.5, 0.], self.evaluate(result), 0)

  def test_weighted_with_threshold(self):
    p_obj = metrics.Precision(thresholds=[0.5, 1.])
    y_true = constant_op.constant([[0, 1], [1, 0]], shape=(2, 2))
    y_pred = constant_op.constant([[1, 0], [0.6, 0]],
                                  shape=(2, 2),
                                  dtype=dtypes.float32)
    weights = constant_op.constant([[4, 0], [3, 1]],
                                   shape=(2, 2),
                                   dtype=dtypes.float32)
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred, sample_weight=weights)
    weighted_tp = 0 + 3.
    weighted_positives = (0 + 3.) + (4. + 0.)
    expected_precision = weighted_tp / weighted_positives
    self.assertArrayNear([expected_precision, 0], self.evaluate(result), 1e-3)

  def test_multiple_updates(self):
    p_obj = metrics.Precision(thresholds=[0.5, 1.])
    y_true = constant_op.constant([[0, 1], [1, 0]], shape=(2, 2))
    y_pred = constant_op.constant([[1, 0], [0.6, 0]],
                                  shape=(2, 2),
                                  dtype=dtypes.float32)
    weights = constant_op.constant([[4, 0], [3, 1]],
                                   shape=(2, 2),
                                   dtype=dtypes.float32)
    self.evaluate(variables.variables_initializer(p_obj.variables))
    update_op = p_obj.update_state(y_true, y_pred, sample_weight=weights)
    for _ in range(2):
      self.evaluate(update_op)

    weighted_tp = (0 + 3.) + (0 + 3.)
    weighted_positives = ((0 + 3.) + (4. + 0.)) + ((0 + 3.) + (4. + 0.))
    expected_precision = weighted_tp / weighted_positives
    self.assertArrayNear([expected_precision, 0], self.evaluate(p_obj.result()),
                         1e-3)

  def test_unweighted_top_k(self):
    p_obj = metrics.Precision(top_k=3)
    y_pred = constant_op.constant([0.2, 0.1, 0.5, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    self.evaluate(variables.variables_initializer(p_obj.variables))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1. / 3, self.evaluate(result))

  def test_weighted_top_k(self):
    p_obj = metrics.Precision(top_k=3)
    y_pred1 = constant_op.constant([0.2, 0.1, 0.4, 0, 0.2], shape=(1, 5))
    y_true1 = constant_op.constant([0, 1, 1, 0, 1], shape=(1, 5))
    self.evaluate(variables.variables_initializer(p_obj.variables))
    self.evaluate(
        p_obj(
            y_true1,
            y_pred1,
            sample_weight=constant_op.constant([[1, 4, 2, 3, 5]])))

    y_pred2 = constant_op.constant([0.2, 0.6, 0.4, 0.2, 0.2], shape=(1, 5))
    y_true2 = constant_op.constant([1, 0, 1, 1, 1], shape=(1, 5))
    result = p_obj(y_true2, y_pred2, sample_weight=constant_op.constant(3))

    tp = (2 + 5) + (3 + 3)
    predicted_positives = (1 + 2 + 5) + (3 + 3 + 3)
    expected_precision = tp / predicted_positives
    self.assertAlmostEqual(expected_precision, self.evaluate(result))

  def test_unweighted_class_id(self):
    p_obj = metrics.Precision(class_id=2)
    self.evaluate(variables.variables_initializer(p_obj.variables))

    y_pred = constant_op.constant([0.2, 0.1, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(p_obj.false_positives))

    y_pred = constant_op.constant([0.2, 0.1, 0, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(p_obj.false_positives))

    y_pred = constant_op.constant([0.2, 0.1, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 0, 0, 0], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(1, self.evaluate(p_obj.false_positives))

  def test_unweighted_top_k_and_class_id(self):
    p_obj = metrics.Precision(class_id=2, top_k=2)
    self.evaluate(variables.variables_initializer(p_obj.variables))

    y_pred = constant_op.constant([0.2, 0.6, 0.3, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(p_obj.false_positives))

    y_pred = constant_op.constant([1, 1, 0.9, 1, 1], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(p_obj.false_positives))

  def test_unweighted_top_k_and_threshold(self):
    p_obj = metrics.Precision(thresholds=.7, top_k=2)
    self.evaluate(variables.variables_initializer(p_obj.variables))

    y_pred = constant_op.constant([0.2, 0.8, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 1], shape=(1, 5))
    result = p_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(p_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(p_obj.false_positives))

@test_util.run_all_in_graph_and_eager_modes
class RecallTest(test.TestCase):

  def test_config(self):
    r_obj = metrics.Recall(
        name='my_recall', thresholds=[0.4, 0.9], top_k=15, class_id=12)
    self.assertEqual(r_obj.name, 'my_recall')
    self.assertEqual(len(r_obj.variables), 2)
    self.assertEqual([v.name for v in r_obj.variables],
                     ['true_positives:0', 'false_negatives:0'])
    self.assertEqual(r_obj.thresholds, [0.4, 0.9])
    self.assertEqual(r_obj.top_k, 15)
    self.assertEqual(r_obj.class_id, 12)

    # Check save and restore config
    r_obj2 = metrics.Recall.from_config(r_obj.get_config())
    self.assertEqual(r_obj2.name, 'my_recall')
    self.assertEqual(len(r_obj2.variables), 2)
    self.assertEqual(r_obj2.thresholds, [0.4, 0.9])
    self.assertEqual(r_obj2.top_k, 15)
    self.assertEqual(r_obj2.class_id, 12)

  def test_value_is_idempotent(self):
    r_obj = metrics.Recall(thresholds=[0.3, 0.72])
    y_pred = random_ops.random_uniform(shape=(10, 3))
    y_true = random_ops.random_uniform(shape=(10, 3))
    update_op = r_obj.update_state(y_true, y_pred)
    self.evaluate(variables.variables_initializer(r_obj.variables))

    # Run several updates.
    for _ in range(10):
      self.evaluate(update_op)

    # Then verify idempotency.
    initial_recall = self.evaluate(r_obj.result())
    for _ in range(10):
      self.assertArrayNear(initial_recall, self.evaluate(r_obj.result()), 1e-3)

  def test_unweighted(self):
    r_obj = metrics.Recall()
    y_pred = constant_op.constant([1, 0, 1, 0], shape=(1, 4))
    y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))

  def test_unweighted_all_incorrect(self):
    r_obj = metrics.Recall(thresholds=[0.5])
    inputs = np.random.randint(0, 2, size=(100, 1))
    y_pred = constant_op.constant(inputs)
    y_true = constant_op.constant(1 - inputs)
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0, self.evaluate(result))

  def test_weighted(self):
    r_obj = metrics.Recall()
    y_pred = constant_op.constant([[1, 0, 1, 0], [0, 1, 0, 1]])
    y_true = constant_op.constant([[0, 1, 1, 0], [1, 0, 0, 1]])
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(
        y_true,
        y_pred,
        sample_weight=constant_op.constant([[1, 2, 3, 4], [4, 3, 2, 1]]))
    weighted_tp = 3.0 + 1.0
    weighted_t = (2.0 + 3.0) + (4.0 + 1.0)
    expected_recall = weighted_tp / weighted_t
    self.assertAlmostEqual(expected_recall, self.evaluate(result))

  def test_div_by_zero(self):
    r_obj = metrics.Recall()
    y_pred = constant_op.constant([0, 0, 0, 0])
    y_true = constant_op.constant([0, 0, 0, 0])
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred)
    self.assertEqual(0, self.evaluate(result))

  def test_unweighted_with_threshold(self):
    r_obj = metrics.Recall(thresholds=[0.5, 0.7])
    y_pred = constant_op.constant([1, 0, 0.6, 0], shape=(1, 4))
    y_true = constant_op.constant([0, 1, 1, 0], shape=(1, 4))
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred)
    self.assertArrayNear([0.5, 0.], self.evaluate(result), 0)

  def test_weighted_with_threshold(self):
    r_obj = metrics.Recall(thresholds=[0.5, 1.])
    y_true = constant_op.constant([[0, 1], [1, 0]], shape=(2, 2))
    y_pred = constant_op.constant([[1, 0], [0.6, 0]],
                                  shape=(2, 2),
                                  dtype=dtypes.float32)
    weights = constant_op.constant([[1, 4], [3, 2]],
                                   shape=(2, 2),
                                   dtype=dtypes.float32)
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred, sample_weight=weights)
    weighted_tp = 0 + 3.
    weighted_positives = (0 + 3.) + (4. + 0.)
    expected_recall = weighted_tp / weighted_positives
    self.assertArrayNear([expected_recall, 0], self.evaluate(result), 1e-3)

  def test_multiple_updates(self):
    r_obj = metrics.Recall(thresholds=[0.5, 1.])
    y_true = constant_op.constant([[0, 1], [1, 0]], shape=(2, 2))
    y_pred = constant_op.constant([[1, 0], [0.6, 0]],
                                  shape=(2, 2),
                                  dtype=dtypes.float32)
    weights = constant_op.constant([[1, 4], [3, 2]],
                                   shape=(2, 2),
                                   dtype=dtypes.float32)
    self.evaluate(variables.variables_initializer(r_obj.variables))
    update_op = r_obj.update_state(y_true, y_pred, sample_weight=weights)
    for _ in range(2):
      self.evaluate(update_op)

    weighted_tp = (0 + 3.) + (0 + 3.)
    weighted_positives = ((0 + 3.) + (4. + 0.)) + ((0 + 3.) + (4. + 0.))
    expected_recall = weighted_tp / weighted_positives
    self.assertArrayNear([expected_recall, 0], self.evaluate(r_obj.result()),
                         1e-3)

  def test_unweighted_top_k(self):
    r_obj = metrics.Recall(top_k=3)
    y_pred = constant_op.constant([0.2, 0.1, 0.5, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    self.evaluate(variables.variables_initializer(r_obj.variables))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))

  def test_weighted_top_k(self):
    r_obj = metrics.Recall(top_k=3)
    y_pred1 = constant_op.constant([0.2, 0.1, 0.4, 0, 0.2], shape=(1, 5))
    y_true1 = constant_op.constant([0, 1, 1, 0, 1], shape=(1, 5))
    self.evaluate(variables.variables_initializer(r_obj.variables))
    self.evaluate(
        r_obj(
            y_true1,
            y_pred1,
            sample_weight=constant_op.constant([[1, 4, 2, 3, 5]])))

    y_pred2 = constant_op.constant([0.2, 0.6, 0.4, 0.2, 0.2], shape=(1, 5))
    y_true2 = constant_op.constant([1, 0, 1, 1, 1], shape=(1, 5))
    result = r_obj(y_true2, y_pred2, sample_weight=constant_op.constant(3))

    tp = (2 + 5) + (3 + 3)
    positives = (4 + 2 + 5) + (3 + 3 + 3 + 3)
    expected_recall = tp / positives
    self.assertAlmostEqual(expected_recall, self.evaluate(result))

  def test_unweighted_class_id(self):
    r_obj = metrics.Recall(class_id=2)
    self.evaluate(variables.variables_initializer(r_obj.variables))

    y_pred = constant_op.constant([0.2, 0.1, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(r_obj.false_negatives))

    y_pred = constant_op.constant([0.2, 0.1, 0, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(1, self.evaluate(r_obj.false_negatives))

    y_pred = constant_op.constant([0.2, 0.1, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 0, 0, 0], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(1, self.evaluate(r_obj.false_negatives))

  def test_unweighted_top_k_and_class_id(self):
    r_obj = metrics.Recall(class_id=2, top_k=2)
    self.evaluate(variables.variables_initializer(r_obj.variables))

    y_pred = constant_op.constant([0.2, 0.6, 0.3, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(0, self.evaluate(r_obj.false_negatives))

    y_pred = constant_op.constant([1, 1, 0.9, 1, 1], shape=(1, 5))
    y_true = constant_op.constant([0, 1, 1, 0, 0], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.5, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(1, self.evaluate(r_obj.false_negatives))

  def test_unweighted_top_k_and_threshold(self):
    r_obj = metrics.Recall(thresholds=.7, top_k=2)
    self.evaluate(variables.variables_initializer(r_obj.variables))

    y_pred = constant_op.constant([0.2, 0.8, 0.6, 0, 0.2], shape=(1, 5))
    y_true = constant_op.constant([1, 1, 1, 0, 1], shape=(1, 5))
    result = r_obj(y_true, y_pred)
    self.assertAlmostEqual(0.25, self.evaluate(result))
    self.assertAlmostEqual(1, self.evaluate(r_obj.true_positives))
    self.assertAlmostEqual(3, self.evaluate(r_obj.false_negatives))

@test_util.run_all_in_graph_and_eager_modes
class SensitivityAtSpecificityTest(test.TestCase, parameterized.TestCase):

  def test_config(self):
    s_obj = metrics.SensitivityAtSpecificity(
        0.4, num_thresholds=100, name='sensitivity_at_specificity_1')
    self.assertEqual(s_obj.name, 'sensitivity_at_specificity_1')
    self.assertLen(s_obj.variables, 4)
    self.assertEqual(s_obj.specificity, 0.4)
    self.assertEqual(s_obj.num_thresholds, 100)

    # Check save and restore config
    s_obj2 = metrics.SensitivityAtSpecificity.from_config(s_obj.get_config())
    self.assertEqual(s_obj2.name, 'sensitivity_at_specificity_1')
    self.assertLen(s_obj2.variables, 4)
    self.assertEqual(s_obj2.specificity, 0.4)
    self.assertEqual(s_obj2.num_thresholds, 100)

  def test_value_is_idempotent(self):
    s_obj = metrics.SensitivityAtSpecificity(0.7)
    y_pred = random_ops.random_uniform((10, 3),
                                       maxval=1,
                                       dtype=dtypes.float32,
                                       seed=1)
    y_true = random_ops.random_uniform((10, 3),
                                       maxval=2,
                                       dtype=dtypes.int64,
                                       seed=1)
    update_op = s_obj.update_state(y_true, y_pred)
    self.evaluate(variables.variables_initializer(s_obj.variables))

    # Run several updates.
    for _ in range(10):
      self.evaluate(update_op)

    # Then verify idempotency.
    initial_sensitivity = self.evaluate(s_obj.result())
    for _ in range(10):
      self.assertAlmostEqual(initial_sensitivity, self.evaluate(s_obj.result()),
                             1e-3)

  def test_unweighted_all_correct(self):
    s_obj = metrics.SensitivityAtSpecificity(0.7)
    inputs = np.random.randint(0, 2, size=(100, 1))
    y_pred = constant_op.constant(inputs, dtype=dtypes.float32)
    y_true = constant_op.constant(inputs)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))

  def test_unweighted_high_specificity(self):
    s_obj = metrics.SensitivityAtSpecificity(0.8)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.1, 0.45, 0.5, 0.8, 0.9]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = constant_op.constant(label_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(0.8, self.evaluate(result))

  def test_unweighted_low_specificity(self):
    s_obj = metrics.SensitivityAtSpecificity(0.4)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.01, 0.02, 0.25, 0.26, 0.26]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = constant_op.constant(label_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(0.6, self.evaluate(result))

  @parameterized.parameters([dtypes.bool, dtypes.int32, dtypes.float32])
  def test_weighted(self, label_dtype):
    s_obj = metrics.SensitivityAtSpecificity(0.4)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.01, 0.02, 0.25, 0.26, 0.26]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]
    weight_values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = math_ops.cast(label_values, dtype=label_dtype)
    weights = constant_op.constant(weight_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred, sample_weight=weights)
    self.assertAlmostEqual(0.675, self.evaluate(result))

  def test_invalid_specificity(self):
    with self.assertRaisesRegexp(
        ValueError, r'`specificity` must be in the range \[0, 1\].'):
      metrics.SensitivityAtSpecificity(-1)

  def test_invalid_num_thresholds(self):
    with self.assertRaisesRegexp(ValueError, '`num_thresholds` must be > 0.'):
      metrics.SensitivityAtSpecificity(0.4, num_thresholds=-1)

@test_util.run_all_in_graph_and_eager_modes
class SpecificityAtSensitivityTest(test.TestCase, parameterized.TestCase):

  def test_config(self):
    s_obj = metrics.SpecificityAtSensitivity(
        0.4, num_thresholds=100, name='specificity_at_sensitivity_1')
    self.assertEqual(s_obj.name, 'specificity_at_sensitivity_1')
    self.assertLen(s_obj.variables, 4)
    self.assertEqual(s_obj.sensitivity, 0.4)
    self.assertEqual(s_obj.num_thresholds, 100)

    # Check save and restore config
    s_obj2 = metrics.SpecificityAtSensitivity.from_config(s_obj.get_config())
    self.assertEqual(s_obj2.name, 'specificity_at_sensitivity_1')
    self.assertLen(s_obj2.variables, 4)
    self.assertEqual(s_obj2.sensitivity, 0.4)
    self.assertEqual(s_obj2.num_thresholds, 100)

  def test_value_is_idempotent(self):
    s_obj = metrics.SpecificityAtSensitivity(0.7)
    y_pred = random_ops.random_uniform((10, 3),
                                       maxval=1,
                                       dtype=dtypes.float32,
                                       seed=1)
    y_true = random_ops.random_uniform((10, 3),
                                       maxval=2,
                                       dtype=dtypes.int64,
                                       seed=1)
    update_op = s_obj.update_state(y_true, y_pred)
    self.evaluate(variables.variables_initializer(s_obj.variables))

    # Run several updates.
    for _ in range(10):
      self.evaluate(update_op)

    # Then verify idempotency.
    initial_specificity = self.evaluate(s_obj.result())
    for _ in range(10):
      self.assertAlmostEqual(initial_specificity, self.evaluate(s_obj.result()),
                             1e-3)

  def test_unweighted_all_correct(self):
    s_obj = metrics.SpecificityAtSensitivity(0.7)
    inputs = np.random.randint(0, 2, size=(100, 1))
    y_pred = constant_op.constant(inputs, dtype=dtypes.float32)
    y_true = constant_op.constant(inputs)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(1, self.evaluate(result))

  def test_unweighted_high_sensitivity(self):
    s_obj = metrics.SpecificityAtSensitivity(0.8)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.1, 0.45, 0.5, 0.8, 0.9]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = constant_op.constant(label_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(0.4, self.evaluate(result))

  def test_unweighted_low_sensitivity(self):
    s_obj = metrics.SpecificityAtSensitivity(0.4)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.01, 0.02, 0.25, 0.26, 0.26]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = constant_op.constant(label_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred)
    self.assertAlmostEqual(0.6, self.evaluate(result))

  @parameterized.parameters([dtypes.bool, dtypes.int32, dtypes.float32])
  def test_weighted(self, label_dtype):
    s_obj = metrics.SpecificityAtSensitivity(0.4)
    pred_values = [0.0, 0.1, 0.2, 0.3, 0.4, 0.01, 0.02, 0.25, 0.26, 0.26]
    label_values = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]
    weight_values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

    y_pred = constant_op.constant(pred_values, dtype=dtypes.float32)
    y_true = math_ops.cast(label_values, dtype=label_dtype)
    weights = constant_op.constant(weight_values)
    self.evaluate(variables.variables_initializer(s_obj.variables))
    result = s_obj(y_true, y_pred, sample_weight=weights)
    self.assertAlmostEqual(0.4, self.evaluate(result))

  def test_invalid_sensitivity(self):
    with self.assertRaisesRegexp(
        ValueError, r'`sensitivity` must be in the range \[0, 1\].'):
      metrics.SpecificityAtSensitivity(-1)

  def test_invalid_num_thresholds(self):
    with self.assertRaisesRegexp(ValueError, '`num_thresholds` must be > 0.'):
      metrics.SpecificityAtSensitivity(0.4, num_thresholds=-1)

@test_util.run_all_in_graph_and_eager_modes
class AUCTest(test.TestCase):

  def setup(self):
    self.num_thresholds = 3
    self.y_pred = constant_op.constant([0, 0.5, 0.3, 0.9], dtype=dtypes.float32)
    self.y_true = constant_op.constant([0, 0, 1, 1])
    self.sample_weight = [1, 2, 3, 4]

    # threshold values are [0 - 1e-7, 0.5, 1 + 1e-7]
    # y_pred when threshold = 0 - 1e-7  : [1, 1, 1, 1]
    # y_pred when threshold = 0.5       : [0, 0, 0, 1]
    # y_pred when threshold = 1 + 1e-7  : [0, 0, 0, 0]

    # without sample_weight:
    # tp = np.sum([[0, 0, 1, 1], [0, 0, 0, 1], [0, 0, 0, 0]], axis=1)
    # fp = np.sum([[1, 1, 0, 0], [0, 0, 0, 0], [0, 0, 0, 0]], axis=1)
    # fn = np.sum([[0, 0, 0, 0], [0, 0, 1, 0], [0, 0, 1, 1]], axis=1)
    # tn = np.sum([[0, 0, 0, 0], [1, 1, 0, 0], [1, 1, 0, 0]], axis=1)

    # tp = [2, 1, 0], fp = [2, 0, 0], fn = [0, 1, 2], tn = [0, 2, 2]

    # with sample_weight:
    # tp = np.sum([[0, 0, 3, 4], [0, 0, 0, 4], [0, 0, 0, 0]], axis=1)
    # fp = np.sum([[1, 2, 0, 0], [0, 0, 0, 0], [0, 0, 0, 0]], axis=1)
    # fn = np.sum([[0, 0, 0, 0], [0, 0, 3, 0], [0, 0, 3, 4]], axis=1)
    # tn = np.sum([[0, 0, 0, 0], [1, 2, 0, 0], [1, 2, 0, 0]], axis=1)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]

  def test_config(self):
    auc_obj = metrics.AUC(
        num_thresholds=100,
        curve='PR',
        summation_method='majoring',
        name='auc_1')
    self.assertEqual(auc_obj.name, 'auc_1')
    self.assertEqual(len(auc_obj.variables), 4)
    self.assertEqual(auc_obj.num_thresholds, 100)
    self.assertEqual(auc_obj.curve, metrics_utils.AUCCurve.PR)
    self.assertEqual(auc_obj.summation_method,
                     metrics_utils.AUCSummationMethod.MAJORING)
    old_config = auc_obj.get_config()
    self.assertDictEqual(old_config, json.loads(json.dumps(old_config)))

    # Check save and restore config.
    auc_obj2 = metrics.AUC.from_config(auc_obj.get_config())
    self.assertEqual(auc_obj2.name, 'auc_1')
    self.assertEqual(len(auc_obj2.variables), 4)
    self.assertEqual(auc_obj2.num_thresholds, 100)
    self.assertEqual(auc_obj2.curve, metrics_utils.AUCCurve.PR)
    self.assertEqual(auc_obj2.summation_method,
                     metrics_utils.AUCSummationMethod.MAJORING)
    new_config = auc_obj2.get_config()
    self.assertDictEqual(old_config, new_config)
    self.assertAllClose(auc_obj.thresholds, auc_obj2.thresholds)

  def test_config_manual_thresholds(self):
    auc_obj = metrics.AUC(
        num_thresholds=None,
        curve='PR',
        summation_method='majoring',
        name='auc_1',
        thresholds=[0.3, 0.5])
    self.assertEqual(auc_obj.name, 'auc_1')
    self.assertEqual(len(auc_obj.variables), 4)
    self.assertEqual(auc_obj.num_thresholds, 4)
    self.assertAllClose(auc_obj.thresholds, [0.0, 0.3, 0.5, 1.0])
    self.assertEqual(auc_obj.curve, metrics_utils.AUCCurve.PR)
    self.assertEqual(auc_obj.summation_method,
                     metrics_utils.AUCSummationMethod.MAJORING)
    old_config = auc_obj.get_config()
    self.assertDictEqual(old_config, json.loads(json.dumps(old_config)))

    # Check save and restore config.
    auc_obj2 = metrics.AUC.from_config(auc_obj.get_config())
    self.assertEqual(auc_obj2.name, 'auc_1')
    self.assertEqual(len(auc_obj2.variables), 4)
    self.assertEqual(auc_obj2.num_thresholds, 4)
    self.assertEqual(auc_obj2.curve, metrics_utils.AUCCurve.PR)
    self.assertEqual(auc_obj2.summation_method,
                     metrics_utils.AUCSummationMethod.MAJORING)
    new_config = auc_obj2.get_config()
    self.assertDictEqual(old_config, new_config)
    self.assertAllClose(auc_obj.thresholds, auc_obj2.thresholds)

  def test_value_is_idempotent(self):
    self.setup()
    auc_obj = metrics.AUC(num_thresholds=3)
    self.evaluate(variables.variables_initializer(auc_obj.variables))

    # Run several updates.
    update_op = auc_obj.update_state(self.y_true, self.y_pred)
    for _ in range(10):
      self.evaluate(update_op)

    # Then verify idempotency.
    initial_auc = self.evaluate(auc_obj.result())
    for _ in range(10):
      self.assertAllClose(initial_auc, self.evaluate(auc_obj.result()), 1e-3)

  def test_unweighted_all_correct(self):
    self.setup()
    auc_obj = metrics.AUC()
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_true)
    self.assertEqual(self.evaluate(result), 1)

  def test_unweighted(self):
    self.setup()
    auc_obj = metrics.AUC(num_thresholds=self.num_thresholds)
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred)

    # tp = [2, 1, 0], fp = [2, 0, 0], fn = [0, 1, 2], tn = [0, 2, 2]
    # recall = [2/2, 1/(1+1), 0] = [1, 0.5, 0]
    # fp_rate = [2/2, 0, 0] = [1, 0, 0]
    # heights = [(1 + 0.5)/2, (0.5 + 0)/2] = [0.75, 0.25]
    # widths = [(1 - 0), (0 - 0)] = [1, 0]
    expected_result = (0.75 * 1 + 0.25 * 0)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_manual_thresholds(self):
    self.setup()
    # Verify that when specified, thresholds are used instead of num_thresholds.
    auc_obj = metrics.AUC(num_thresholds=2, thresholds=[0.5])
    self.assertEqual(auc_obj.num_thresholds, 3)
    self.assertAllClose(auc_obj.thresholds, [0.0, 0.5, 1.0])
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred)

    # tp = [2, 1, 0], fp = [2, 0, 0], fn = [0, 1, 2], tn = [0, 2, 2]
    # recall = [2/2, 1/(1+1), 0] = [1, 0.5, 0]
    # fp_rate = [2/2, 0, 0] = [1, 0, 0]
    # heights = [(1 + 0.5)/2, (0.5 + 0)/2] = [0.75, 0.25]
    # widths = [(1 - 0), (0 - 0)] = [1, 0]
    expected_result = (0.75 * 1 + 0.25 * 0)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_roc_interpolation(self):
    self.setup()
    auc_obj = metrics.AUC(num_thresholds=self.num_thresholds)
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # recall = [7/7, 4/(4+3), 0] = [1, 0.571, 0]
    # fp_rate = [3/3, 0, 0] = [1, 0, 0]
    # heights = [(1 + 0.571)/2, (0.571 + 0)/2] = [0.7855, 0.2855]
    # widths = [(1 - 0), (0 - 0)] = [1, 0]
    expected_result = (0.7855 * 1 + 0.2855 * 0)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_roc_majoring(self):
    self.setup()
    auc_obj = metrics.AUC(
        num_thresholds=self.num_thresholds, summation_method='majoring')
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # recall = [7/7, 4/(4+3), 0] = [1, 0.571, 0]
    # fp_rate = [3/3, 0, 0] = [1, 0, 0]
    # heights = [max(1, 0.571), max(0.571, 0)] = [1, 0.571]
    # widths = [(1 - 0), (0 - 0)] = [1, 0]
    expected_result = (1 * 1 + 0.571 * 0)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_roc_minoring(self):
    self.setup()
    auc_obj = metrics.AUC(
        num_thresholds=self.num_thresholds, summation_method='minoring')
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # recall = [7/7, 4/(4+3), 0] = [1, 0.571, 0]
    # fp_rate = [3/3, 0, 0] = [1, 0, 0]
    # heights = [min(1, 0.571), min(0.571, 0)] = [0.571, 0]
    # widths = [(1 - 0), (0 - 0)] = [1, 0]
    expected_result = (0.571 * 1 + 0 * 0)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_pr_majoring(self):
    self.setup()
    auc_obj = metrics.AUC(
        num_thresholds=self.num_thresholds,
        curve='PR',
        summation_method='majoring')
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # precision = [7/(7+3), 4/4, 0] = [0.7, 1, 0]
    # recall = [7/7, 4/(4+3), 0] = [1, 0.571, 0]
    # heights = [max(0.7, 1), max(1, 0)] = [1, 1]
    # widths = [(1 - 0.571), (0.571 - 0)] = [0.429, 0.571]
    expected_result = (1 * 0.429 + 1 * 0.571)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_pr_minoring(self):
    self.setup()
    auc_obj = metrics.AUC(
        num_thresholds=self.num_thresholds,
        curve='PR',
        summation_method='minoring')
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # precision = [7/(7+3), 4/4, 0] = [0.7, 1, 0]
    # recall = [7/7, 4/(4+3), 0] = [1, 0.571, 0]
    # heights = [min(0.7, 1), min(1, 0)] = [0.7, 0]
    # widths = [(1 - 0.571), (0.571 - 0)] = [0.429, 0.571]
    expected_result = (0.7 * 0.429 + 0 * 0.571)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_weighted_pr_interpolation(self):
    self.setup()
    auc_obj = metrics.AUC(num_thresholds=self.num_thresholds, curve='PR')
    self.evaluate(variables.variables_initializer(auc_obj.variables))
    result = auc_obj(self.y_true, self.y_pred, sample_weight=self.sample_weight)

    # auc = (slope / Total Pos) * [dTP - intercept * log(Pb/Pa)]

    # tp = [7, 4, 0], fp = [3, 0, 0], fn = [0, 3, 7], tn = [0, 3, 3]
    # P = tp + fp = [10, 4, 0]
    # dTP = [7-4, 4-0] = [3, 4]
    # dP = [10-4, 4-0] = [6, 4]
    # slope = dTP/dP = [0.5, 1]
    # intercept = (TPa+(slope*Pa) = [(4 - 0.5*4), (0 - 1*0)] = [2, 0]
    # (Pb/Pa) = (Pb/Pa) if Pb > 0 AND Pa > 0 else 1 = [10/4, 4/0] = [2.5, 1]
    # auc * TotalPos = [(0.5 * (3 + 2 * log(2.5))), (1 * (4 + 0))]
    #                = [2.416, 4]
    # auc = [2.416, 4]/(tp[1:]+fn[1:])
    expected_result = (2.416/7 + 4/7)
    self.assertAllClose(self.evaluate(result), expected_result, 1e-3)

  def test_invalid_num_thresholds(self):
    with self.assertRaisesRegexp(ValueError, '`num_thresholds` must be > 1.'):
      metrics.AUC(num_thresholds=-1)

    with self.assertRaisesRegexp(ValueError, '`num_thresholds` must be > 1.'):
      metrics.AUC(num_thresholds=1)

  def test_invalid_curve(self):
    with self.assertRaisesRegexp(ValueError,
                                 'Invalid AUC curve value "Invalid".'):
      metrics.AUC(curve='Invalid')

  def test_invalid_summation_method(self):
    with self.assertRaisesRegexp(
        ValueError, 'Invalid AUC summation method value "Invalid".'):
      metrics.AUC(summation_method='Invalid')

if __name__ == '__main__':
  test.main()

max_stars_repo_name: abhaikollara/tensorflow

max_issues_repo_name: abhaikollara/tensorflow

max_forks_repo_name: abhaikollara/tensorflow


text:
from markdown import markdown
import numpy as np

def joke():
    return markdown(u'hello world, this is a real test')

max_stars_repo_name: walkacross/stockgo

max_issues_repo_name: walkacross/stockgo

max_forks_repo_name: walkacross/stockgo