text:
import numpy
import pylab
from pImagePlots import PImagePlots

def ComputeStructureFunction(x0=0.0, x1=1.75, y1=0.04, ymm=0.08, xmax=10.):
    """Create the structure function based on an analytical model.

    This analytical model is determined by fitting data (done elsewhere).
    The variables x0, x1, y1, and ymm describe the structure function.
    x0=0, x1=1.75,y1=0.04, ymm=0.08 are defaults fit to the SDSS (?) structure function.
    ymm is related to the long-range grayscale extinction, while x1 and y2 are related to
    the inflection points in the structure function. """
    # define the x range
    xsf = numpy.arange(0, xmax, 0.1)
    # Calculate variable for the structure function.
    al = -(1/(x1-x0))*numpy.log(1-(y1/ymm))
    # Calculate the actual structure function.
    SF = ymm*(1.-numpy.exp(-al*xsf))
    return xsf, SF

def readSF(file):
    import useful_input as ui
    data = ui.readDatafile(file, ('theta', 'sf'))
    # Convert arcminutes to degrees
    data['theta'] = data['theta'] / 60.0
    # Convert Tim's SF**2 to SF (mag**2 -> mag)
    data['sf'] = numpy.sqrt(data['sf'])
    return data['theta'], data['sf']

def rescaleImage(image, sigma_goal, kappa):
    sigma = numpy.std(image)
    mean = numpy.mean(image)
    print 'Before: mean/sigma/min/max:', mean, sigma, image.min(), image.max()
    image = image - mean
    image *= sigma_goal / sigma
    image += kappa
    # make sure no 'negative' clouds
    image = numpy.where(image<0, 0, image)
    sigma = numpy.std(image)
    mean = numpy.mean(image)
    print 'After: mean/sigma/min/max:', mean, sigma, image.min(), image.max()
    return image

def imageStats(image):
    print 'Image: mean/sigma/min/max:', numpy.std(image), numpy.mean(image), image.min(), image.max()
    return

if __name__ == "__main__":
    #xsf, SF = ComputeStructureFunction()
    xsf, SF = readSF('sf_arma.dat')
    # Plot result - looks like X in degrees.
    pylab.figure()
    pylab.plot(xsf, SF)
    pylab.xlabel('Degrees')
    pylab.ylabel('Structure Function (mag)')
    pylab.title('ARMA structure function')
    pylab.grid()
    pylab.savefig('clouds_sf_ARMA.png', format='png')

    goal_rad_fov = 1.75 * numpy.sqrt(2)  # radius to furthest corner of fov for LSST
    sigma_goal = SF[numpy.abs(xsf-goal_rad_fov*numpy.sqrt(2))<0.1].mean()
    print 'sigma_goal', sigma_goal
    kappa = 1.

    # And also try to make an image with the same (final) pixel scale, but that will be created (first)
    #  larger and then scaled down, to avoid introducing artifacts from the ACovF1d being truncated.
    final_imsize = 1500 # pixels desired in final image
    final_rad_fov = 1.75*numpy.sqrt(2) # degrees
    final_pixscale = 2*final_rad_fov / float(final_imsize)
    # Start with larger image
    pixscale = final_pixscale
    rad_fov = final_rad_fov
    imsize = int(2*rad_fov / pixscale)
    if (imsize%2 != 0):
        imsize = imsize + 1
    xr = xsf / pixscale # xr = in pixels, over range that want to simulate
    print 'Image: Rad_fov', rad_fov, 'Imsize', imsize, 'Pixscale', pixscale, 'deg/pix', '(', pixscale*60.*60., 'arcsec/pix)'
    im = PImagePlots(shift=True, nx=imsize, ny=imsize)
    im.makeImageFromSf(sfx=xr, sf=SF)
    im.showAcovf1d()
    im.showPsd2dI()
    # Trim image to desired final size
    trim = round((imsize - final_imsize)/2.0)
    print 'Trimming about %d pixels from each side' %(trim)
    image = im.imageI[trim:trim+final_imsize, trim:trim+final_imsize]
    image = rescaleImage(image.real, sigma_goal, kappa)
    imsize = len(image)
    pixscale = pixscale
    rad_fov = imsize/2.0*pixscale
    print 'Image After Trimming: Rad_fov', rad_fov, 'Imsize', imsize, 'Pixscale', pixscale, 'deg/pix', '(', pixscale*60.*60., 'arcsec/pix)'
    im.setImage(image.real)
    imageStats(im.image)
    im.showImage(copy=True)
    im.hanningFilter()
    im.calcAll()
    im.showPsd2d()
    im.showAcovf2d()
    im.showAcovf1d(comparison=im)

    # And also try to make an image with the same (final) pixel scale, but that will be created (first)
    #  larger and then scaled down, to avoid introducing artifacts from the ACovF1d being truncated.
    final_imsize = 1500 # pixels desired in final image
    final_rad_fov = 1.75*numpy.sqrt(2) # degrees
    final_pixscale = 2*final_rad_fov / float(final_imsize)
    # Start with larger image
    pixscale = final_pixscale
    rad_fov = final_rad_fov*2.
    imsize = int(2*rad_fov / pixscale)
    if (imsize%2 != 0):
        imsize = imsize + 1
    xr = xsf / pixscale # xr = in pixels, over range that want to simulate
    print 'Image: Rad_fov', rad_fov, 'Imsize', imsize, 'Pixscale', pixscale, 'deg/pix', '(', pixscale*60.*60., 'arcsec/pix)'
    im2 = PImagePlots(shift=True, nx=imsize, ny=imsize)
    im2.makeImageFromSf(sfx=xr, sf=SF)
    im2.showAcovf1d()
    im2.showPsd2dI()
    # Trim image to desired final size
    trim = round((imsize - final_imsize)/2.0)
    print 'Trimming about %d pixels from each side' %(trim)
    image = im2.imageI[trim:trim+final_imsize, trim:trim+final_imsize]
    image = rescaleImage(image.real, sigma_goal, kappa)
    imsize = len(image)
    pixscale = pixscale
    rad_fov = imsize/2.0*pixscale
    print 'Image After Trimming: Rad_fov', rad_fov, 'Imsize', imsize, 'Pixscale', pixscale, 'deg/pix', '(', pixscale*60.*60., 'arcsec/pix)'
    im2.setImage(image.real)
    imageStats(im2.image)
    im2.showImage(copy=True)
    im2.hanningFilter()
    im2.calcAll()
    im2.showPsd2d()
    im2.showAcovf2d()
    im2.showAcovf1d(comparison=im)

    # Compare structure functions, but scaled (due to loss of amplitude info with random phases)
    im.sfx = im.sfx * pixscale
    im2.sfx = im2.sfx * pixscale
    scaleval = im.sf[numpy.abs(im.sfx - goal_rad_fov)<0.05].mean()
    print 'Cloud 1 scaleval', scaleval
    im.sf = im.sf/scaleval
    scaleval = im2.sf[numpy.abs(im2.sfx - goal_rad_fov)<0.05].mean()
    print 'Cloud 2 scaleval', scaleval
    im2.sf = im2.sf/scaleval
    im.showSf(linear=True, comparison=im2, legendlabels=['Clouds 1', 'Clouds 2'])
    scaleval = SF[numpy.where(numpy.abs(xsf - goal_rad_fov)<0.2)].mean()
    print 'ARMA scaleval', scaleval
    SF2 = SF / scaleval
    pylab.plot(xsf, SF2, 'k:', label='Original/ARMA')
    pylab.xlim(0, goal_rad_fov*1.2)
    pylab.ylim(0, 1.2)
    pylab.legend(numpoints=1, fancybox=True, fontsize='smaller')
    pylab.title('Structure Function (mag)')
    pylab.xlabel('Degrees')

    pylab.show()
    exit()

max_stars_repo_name: rhiannonlynne/powerspectrum

max_issues_repo_name: rhiannonlynne/powerspectrum

max_forks_repo_name: rhiannonlynne/powerspectrum


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

import pandas as pd
import numpy as np
import matplotlib.pylab as plt
from sklearn import linear_model
from sklearn.neighbors import KNeighborsRegressor
from service.base import ServiceInterface
from service.tools import plot_service_model, plot_service_values
from service.video import VideoServiceKNN
from service.video import VideoServiceLinear

#%%

VideoService = VideoServiceKNN
#VideoService = VideoServiceLinear

#%%

if __name__ == "__main__":

    df = pd.read_csv("data/metrics_dash_1sec.csv")

    X = df[['cfg_client_maxrate_kbit', 'cfg_delay']]
    y = df.mean_ql_mean

    service_model = VideoService(X, y)

    zlabel = r"Mean quality level"

    mgrid = np.mgrid[X.iloc[:,0].min():X.iloc[:,0].max():19j,
                     X.iloc[:,1].min():X.iloc[:,1].max():19j]

    x1, y1 = mgrid

    #%%

    from scipy.interpolate import griddata

    z_cubic = griddata(X, y, (x1, y1), method='cubic')

    vmin = 1
    vmax = 5

    plot_service_model(x1, y1, z_cubic, zlabel, "Data (Cubic Interpolation)",
                       vmin=vmin, vmax=vmax, fn="service_data_video_live.png",
                       datapoints=X)

    z_pred = np.zeros(x1.shape)

    for i in np.ndindex(x1.shape):
        z_pred[i] = service_model.predict(x1[i], y1[i])

    plot_service_model(x1, y1, z_pred, zlabel, "Predicted (%s)" % service_model.METHOD,
                       vmin=vmin, vmax=vmax, fn="service_pred_video_live.png",
                       datapoints=X)

    #%% Raw values
    plot_service_values(X, y, zlabel, "Data (Raw)",
                        vmin=vmin, vmax=vmax, fn="service_values_video_live.png")

ext: py

max_stars_repo_name: tum-lkn/appaware

max_issues_repo_name: tum-lkn/appaware

max_forks_repo_name: tum-lkn/appaware

reason: import numpy,from scipy


text:
# -*- coding: utf-8 -*-
# Copyright 2016 The TensorFlow Authors. All Rights Reserved.
# Modifications Copyright 2017 Abigail See
#
# 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.
# ==============================================================================
"""This file contains code to process data into batches"""

import Queue
from random import shuffle
from threading import Thread
import time
import numpy as np
import tensorflow as tf
from data import *

class Example(object):
    """Class representing a train/val/test example for text summarization."""

    def __init__(self, title, summarization, vocab, hps):
        """Initializes the Example,
        performing tokenization and truncation to produce the encoder, decoder and target sequences, which are stored in self.

        Args:
            article: source text; a string. each token is separated by a single space.
            summarization_sentences: list of strings, one per summarization sentence.
            In each sentence, each token is separated by a single space.
            vocab: Vocabulary object
            hps: hyperparameters
        """
        self.hps = hps

        # Get ids of special tokens
        start_decoding = vocab.word2id(MARK_GO)
        stop_decoding = vocab.word2id(MARK_EOS)

        title_words = split_text_with_whitespace(title) + [MARK_EOS]
        # if len(title_words) > hps.max_enc_steps:
        #     title_words = title_words[:hps.max_enc_steps]
        self.enc_len = len(title_words)
        self.enc_input = [vocab.word2id(w) for w in title_words]

        # Process the summarization
        summarization_words = split_text_with_whitespace(summarization)
        summarization_ids = [vocab.word2id(w) for w in summarization_words]

        # Get the decoder input sequence and target sequence
        #todo: max_dec_steps
        self.dec_input, self.target = self.get_dec_inp_targ_seqs(
            summarization_ids, hps.max_dec_steps, start_decoding,
            stop_decoding)
        self.dec_len = len(self.dec_input)

        # If using pointer-generator mode, we need to store some extra info
        if hps.pointer_gen:
            # Store a version of the enc_input where in-article OOVs are represented by their temporary OOV id; also store the in-article OOVs words themselves
            self.enc_input_extend_vocab, self.title_oovs = title2ids(
                title_words, vocab)
            # Get a verison of the reference summary where in-article OOVs are represented by their temporary article OOV id
            abs_ids_extend_vocab = summarization2ids(summarization_words,
                                                     vocab, self.title_oovs)

            # Overwrite decoder target sequence so it uses the temp article OOV ids
            #todo: 为什么 decoder input不用重写扩展
            _, self.target = self.get_dec_inp_targ_seqs(
                abs_ids_extend_vocab, hps.max_dec_steps, start_decoding,
                stop_decoding)

        # Store the original strings
        self.original_title = title
        self.original_summarization = summarization

    def get_dec_inp_targ_seqs(self, sequence, max_len, start_id, stop_id):
        """Given the reference summary as a sequence of tokens,
        return the input sequence for the decoder, and the target sequence which we will use to calculate loss.
        The sequence will be truncated if it is longer than max_len.
        The input sequence must start with the start_id and the target sequence must end with the stop_id (but not if it's been truncated).

        Args:
            sequence: List of ids (integers)
            max_len: integer
            start_id: integer
            stop_id: integer

        Returns:
            inp: sequence length <=max_len starting with start_id
            target: sequence same length as input, ending with stop_id only if there was no truncation
        """
        inp = [start_id] + sequence[:]
        target = sequence[:]
        if len(inp) > max_len:  # truncate
            inp = inp[:max_len]
            target = target[:max_len]  # no end_token
        else:  # no truncation
            target.append(stop_id)  # end token
        assert len(inp) == len(target)
        return inp, target

    def pad_decoder_inp_targ(self, max_len, pad_id):
        """Pad decoder input and target sequences with pad_id up to max_len."""
        while len(self.dec_input) < max_len:
            self.dec_input.append(pad_id)
        while len(self.target) < max_len:
            self.target.append(pad_id)

    def pad_encoder_input(self, max_len, pad_id):
        """Pad the encoder input sequence with pad_id up to max_len."""
        #while len(self.enc_input) < max_len:
        self.enc_input += [pad_id] * (max_len - len(self.enc_input))
        if self.hps.pointer_gen:
            self.enc_input_extend_vocab += [pad_id] * (
                max_len - len(self.enc_input_extend_vocab))
            # while len(self.enc_input_extend_vocab) < max_len:
            #     self.enc_input_extend_vocab.append(pad_id)

class Batch(object):
    """Class representing a minibatch of train/val/test examples for text summarization."""

    def __init__(self, example_list, hps, vocab):
        """Turns the example_list into a Batch object.

        Args:
          example_list: List of Example objects
          hps: hyperparameters
          vocab: Vocabulary object
        """
        self.pad_id = vocab.word2id(MARK_PAD)
        self.init_encoder_seq(example_list, hps)
        self.init_decoder_seq(example_list, hps)
        self.store_orig_strings(example_list)  # store the original strings

    def init_encoder_seq(self, example_list, hps):
        """Initializes the following:
        self.enc_batch:
            numpy array of shape (batch_size, <=max_enc_steps) containing integer ids (all OOVs represented by UNK id), padded to length of longest sequence in the batch
        self.enc_lens:
            numpy array of shape (batch_size) containing integers. The (truncated) length of each encoder input sequence (pre-padding).
        self.enc_padding_mask:
            numpy array of shape (batch_size, <=max_enc_steps), containing 1s and 0s. 1s correspond to real tokens in enc_batch and target_batch; 0s correspond to padding.

        If hps.pointer_gen, additionally initializes the following:
            self.max_art_oovs:
                maximum number of in-article OOVs in the batch
            self.art_oovs:
                list of list of in-article OOVs (strings), for each example in the batch
            self.enc_batch_extend_vocab:
                Same as self.enc_batch, but in-article OOVs are represented by their temporary article OOV number.
        """
        # Determine the maximum length of the encoder input sequence in this batch
        max_enc_seq_len = max([ex.enc_len for ex in example_list])

        # Pad the encoder input sequences up to the length of the longest sequence
        for ex in example_list:
            ex.pad_encoder_input(max_enc_seq_len, self.pad_id)

        # Initialize the numpy arrays
        # Note: our enc_batch can have different length (second dimension) for each batch because we use dynamic_rnn for the encoder.
        self.enc_batch = np.zeros(
            (hps.batch_size, max_enc_seq_len), dtype=np.int32)
        self.enc_lens = np.zeros((hps.batch_size), dtype=np.int32)
        self.enc_padding_mask = np.zeros(
            (hps.batch_size, max_enc_seq_len), dtype=np.float32)

        # Fill in the numpy arrays
        for i, ex in enumerate(example_list):
            self.enc_batch[i, :] = ex.enc_input[:]
            self.enc_lens[i] = ex.enc_len
            for j in xrange(ex.enc_len):
                self.enc_padding_mask[i][j] = 1

        # For pointer-generator mode, need to store some extra info
        if hps.pointer_gen:
            # Determine the max number of in-article OOVs in this batch
            self.max_art_oovs = max(
                [len(ex.title_oovs) for ex in example_list])
            # Store the in-article OOVs themselves
            self.art_oovs = [ex.title_oovs for ex in example_list]
            # Store the version of the enc_batch that uses the article OOV ids
            self.enc_batch_extend_vocab = np.zeros(
                (hps.batch_size, max_enc_seq_len), dtype=np.int32)
            for i, ex in enumerate(example_list):
                self.enc_batch_extend_vocab[
                    i, :] = ex.enc_input_extend_vocab[:]

    def init_decoder_seq(self, example_list, hps):
        """Initializes the following:
        self.dec_batch:
            numpy array of shape (batch_size, max_dec_steps), containing integer ids as input for the decoder, padded to max_dec_steps length.
        self.target_batch:
            numpy array of shape (batch_size, max_dec_steps), containing integer ids for the target sequence, padded to max_dec_steps length.
        self.dec_padding_mask:
            numpy array of shape (batch_size, max_dec_steps), containing 1s and 0s. 1s correspond to real tokens in dec_batch and target_batch; 0s correspond to padding.
        """
        # Pad the inputs and targets
        for ex in example_list:
            ex.pad_decoder_inp_targ(hps.max_dec_steps, self.pad_id)

        # Initialize the numpy arrays.
        # Note: our decoder inputs and targets must be the same length for each batch (second dimension = max_dec_steps) because we do not use a dynamic_rnn for decoding. However I believe this is possible, or will soon be possible, with Tensorflow 1.0, in which case it may be best to upgrade to that.
        self.dec_batch = np.zeros(
            (hps.batch_size, hps.max_dec_steps), dtype=np.int32)
        self.target_batch = np.zeros(
            (hps.batch_size, hps.max_dec_steps), dtype=np.int32)
        self.dec_padding_mask = np.zeros(
            (hps.batch_size, hps.max_dec_steps), dtype=np.float32)

        # Fill in the numpy arrays
        for i, ex in enumerate(example_list):
            self.dec_batch[i, :] = ex.dec_input[:]
            self.target_batch[i, :] = ex.target[:]
            for j in xrange(ex.dec_len):
                self.dec_padding_mask[i][j] = 1

    def store_orig_strings(self, example_list):
        """Store the original article and abstract strings in the Batch object"""
        self.original_titles = [ex.original_title for ex in example_list]
        self.original_summarizations = [
            ex.original_summarization for ex in example_list
        ]
        #todo: need to check
        # self.original_summarizations_sents = [
        #     ex.original_abstract_sents for ex in example_list
        # ]  # list of list of lists

class Batcher(object):
    """A class to generate minibatches of data.
    Buckets examples together based on length of the encoder sequence."""

    BATCH_QUEUE_MAX = 100  # max number of batches the batch_queue can hold

    def __init__(self, data_path, vocab, hps, single_pass):
        """Initialize the batcher. Start threads that process the data into batches.
        Args:
          data_path: tf.Example filepattern.
          vocab: Vocabulary object
          hps: hyperparameters
          single_pass: If True, run through the dataset exactly once
                      (useful for when you want to run evaluation on the dev or test set).
          Otherwise generate random batches indefinitely (useful for training).
        """
        self._data_path = data_path
        self._vocab = vocab
        self._hps = hps
        self._single_pass = single_pass

        # Initialize a queue of Batches waiting to be used, and a queue of Examples waiting to be batched
        self._batch_queue = Queue.Queue(self.BATCH_QUEUE_MAX)
        self._example_queue = Queue.Queue(
            self.BATCH_QUEUE_MAX * self._hps.batch_size)

        # Different settings depending on whether we're in single_pass mode or not
        if single_pass:
            self._num_example_q_threads = 1  # just one thread, so we read through the dataset just once
            self._num_batch_q_threads = 1  # just one thread to batch examples
            self._bucketing_cache_size = 1  # only load one batch's worth of examples before bucketing; this essentially means no bucketing
            self._finished_reading = False  # this will tell us when we're finished reading the dataset
        else:
            self._num_example_q_threads = 1  #16 num threads to fill example queue
            self._num_batch_q_threads = 1  # 4 num threads to fill batch queue
            self._bucketing_cache_size = 100  # how many batches-worth of examples to load into cache before bucketing

        # Start the threads that load the queues
        self._example_q_threads = []
        for _ in xrange(self._num_example_q_threads):
            self._example_q_threads.append(
                Thread(target=self.fill_example_queue))
            self._example_q_threads[-1].daemon = True
            self._example_q_threads[-1].start()
        self._batch_q_threads = []
        for _ in xrange(self._num_batch_q_threads):
            self._batch_q_threads.append(Thread(target=self.fill_batch_queue))
            self._batch_q_threads[-1].daemon = True
            self._batch_q_threads[-1].start()

        # Start a thread that watches the other threads and restarts them if they're dead
        if not single_pass:  # We don't want a watcher in single_pass mode because the threads shouldn't run forever
            self._watch_thread = Thread(target=self.watch_threads)
            self._watch_thread.daemon = True
            self._watch_thread.start()

    def next_batch(self):
        """Return a Batch from the batch queue.

        If mode='decode' then each batch contains a single example repeated beam_size-many times; this is necessary for beam search.

        Returns:
          batch: a Batch object, or None if we're in single_pass mode and we've exhausted the dataset.
        """
        # If the batch queue is empty, print a warning
        if self._batch_queue.qsize() == 0:
            tf.logging.warning(
                'Bucket input queue is empty when calling next_batch. Bucket queue size: %i, Input queue size: %i',
                self._batch_queue.qsize(), self._example_queue.qsize())
            if self._single_pass and self._finished_reading:
                tf.logging.info(
                    "Finished reading dataset in single_pass mode.")
                return None

        batch = self._batch_queue.get()  # get the next Batch
        return batch

    def fill_example_queue(self):
        """Reads data from file and processes into Examples which are then placed into the example queue."""

        input_gen = self.text_generator(self._data_path, self._single_pass)

        while True:
            try:
                (title, summarization) = input_gen.next()
            except StopIteration:  # if there are no more examples:
                tf.logging.info(
                    "The example generator for this example queue filling thread has exhausted data."
                )
                if self._single_pass:
                    tf.logging.info(
                        "single_pass mode is on, so we've finished reading dataset. This thread is stopping."
                    )
                    self._finished_reading = True
                    break
                else:
                    raise Exception(
                        "single_pass mode is off but the example generator is out of data; error."
                    )

            # abstract_sentences = [
            #     sent.strip() for sent in data.abstract2sents(abstract)
            # ]  # Use the <s> and </s> tags in abstract to get a list of sentences.
            example = Example(title, summarization, self._vocab,
                              self._hps)  # Process into an Example.
            self._example_queue.put(
                example)  # place the Example in the example queue.

    def fill_batch_queue(self):
        """Takes Examples out of example queue,
        sorts them by encoder sequence length,
        processes into Batches and places them in the batch queue.

        In decode mode, makes batches that each contain a single example repeated.
        todo: why? 为什么decode时，重复
        """
        while True:
            if self._hps.mode != 'decode':
                # Get bucketing_cache_size-many batches of Examples into a list, then sort
                inputs = []
                for _ in xrange(
                        self._hps.batch_size * self._bucketing_cache_size):
                    inputs.append(self._example_queue.get())
                inputs = sorted(inputs, key=lambda inp: inp.enc_len)

                # Group the sorted Examples into batches, optionally shuffle the batches, and place in the batch queue.
                batches = []
                for i in xrange(0, len(inputs), self._hps.batch_size):
                    batches.append(inputs[i:i + self._hps.batch_size])
                if not self._single_pass:
                    shuffle(batches)
                for b in batches:  # each b is a list of Example objects
                    self._batch_queue.put(Batch(b, self._hps, self._vocab))

            else:  # beam search decode mode
                ex = self._example_queue.get()
                b = [ex for _ in xrange(self._hps.batch_size)]
                self._batch_queue.put(Batch(b, self._hps, self._vocab))

    def watch_threads(self):
        """Watch example queue and batch queue threads and restart if dead."""
        while True:
            time.sleep(60)
            for idx, t in enumerate(self._example_q_threads):
                if not t.is_alive():  # if the thread is dead
                    tf.logging.error(
                        'Found example queue thread dead. Restarting.')
                    new_t = Thread(target=self.fill_example_queue)
                    self._example_q_threads[idx] = new_t
                    new_t.daemon = True
                    new_t.start()
            for idx, t in enumerate(self._batch_q_threads):
                if not t.is_alive():  # if the thread is dead
                    tf.logging.error(
                        'Found batch queue thread dead. Restarting.')
                    new_t = Thread(target=self.fill_batch_queue)
                    self._batch_q_threads[idx] = new_t
                    new_t.daemon = True
                    new_t.start()

    def text_generator(self, data_path, single_pass):
        """Generates article and abstract text from tf.Example.

        Args:
            data_path:
            single_pass:
        """
        while True:
            filelist = glob.glob(data_path)  # get the list of datafiles
            assert filelist, ('Error: Empty filelist at %s' % data_path)
            if single_pass:
                filelist = sorted(filelist)
            else:
                random.shuffle(filelist)
            for f in filelist:
                with open(f) as train_f:
                    for line in train_f:
                        record = line.decode('utf-8').strip().split('\t\t')

                        yield (record[0], record[1])
                    #yield example_pb2.Example.FromString(example_str)
            if single_pass:
                print "text_generator completed reading all datafiles. No more data."
                break

reason: import numpy


text:
import os
from glob import glob
import numpy as np

from .utils import info, suppress_stdout

try:
    with suppress_stdout():
        from actin.actin import actin as ACTIN  # actinception
    _actin_available = True
except ImportError:
    _actin_available = False

def actin(self):
    if not _actin_available:
        print('Please (pip) install actin to use this function')
        return

    # download and load S2D spectra
    self.enable_spectra(s1d=False)

    # call actin
    output_dir = os.path.join(self.spec_dir, 'actin')
    indices = ['I_CaII', 'I_NaI', 'I_Ha16', 'I_Ha06', 'I_HeI', 'I_CaI']

    if all([hasattr(self, index) for index in indices]):
        if self.verbose:
            info('actin alread calculated')
        return

    ACTIN(files=[s.filename for s in self.spectra],
          calc_index=indices,
          obj_name=self.star,
          save_output=output_dir,
          del_out=False,
          verbose=False)

    results = load_actin_results(output_dir, indices)
    for index, v in results.items():
        # add attribute to self
        setattr(self, index, v)
        # add attribute to self.each
        for i, individual in enumerate(self.each):
            m = self.obs == i + 1
            setattr(individual, index, v[m])
        # keep track of attributes in _activity_indicators and print
        if not index.endswith('_err'):
            self._activity_indicators.add(index)
            if self.verbose:
                info(f'Adding {index} and {index}_err as attributes')

def load_actin_results(directory, indices):
    rdb = glob(os.path.join(directory, '*/*.rdb'))[0]
    data = np.genfromtxt(rdb,
                         names=True,
                         comments='--',
                         dtype=None,
                         encoding=None)
    r = {}
    for i in indices:
        r.update(as_dict(data[[i, i + '_err']]))
    return r

def as_dict(rec):
    """ Turn a numpy recarray record into a dict. """
    return {name: rec[name] for name in rec.dtype.names}

max_stars_repo_name: j-faria/vera

max_issues_repo_name: j-faria/vera

max_forks_repo_name: j-faria/vera


text:
import pandas as pd
import numpy as np
from matplotlib import pyplot as plt
import torch
from torchvision import datasets, transforms
import os

from RBM import RBM

if __name__ == '__main__':
    test_dataset = datasets.MNIST('dataset', download=True, train=False, transform=transforms.ToTensor())
    test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=len(test_dataset))
    for test_x, test_y in test_loader:
        test_x = test_x.view(-1, 784)

    vn = test_x.shape[1]
    hn = 2500

    rbm = RBM(vn, hn)
    rbm.load_rbm('models/mnist_trained_rbm.pt')

    plt.figure(figsize=(12, 60))

    for n in range(10):
        x = test_x[np.where(test_y==n)[0][0]]
        x = x.unsqueeze(0)
        hidden_image = []
        gen_image = []

        for k in range(rbm.k):
            _, hk = rbm.sample_h(x)
            _, vk = rbm.sample_v(hk)
            gen_image.append(vk.cpu().numpy())
            hidden_image.append(hk.cpu().numpy())

        hidden_image = np.array(hidden_image)
        hidden_image = np.mean(hidden_image, axis=0)
        gen_image = np.array(gen_image)
        gen_image = np.mean(gen_image, axis=0)
        image = x.numpy()

        # revert transforms.ToTensor() scaling
        image = (image*255)[0]
        hidden_image = (hidden_image*255)[0]
        gen_image = (gen_image*255)[0]

        image = np.reshape(image, (28, 28))
        hidden_image = np.reshape(hidden_image, (50, 50))
        gen_image = np.reshape(gen_image, (28, 28))

        image = image.astype(np.int32)
        hidden_image = hidden_image.astype(np.int32)
        gen_image = gen_image.astype(np.int32)

        plt.subplot(10, 3, 1 + n*3)
        plt.imshow(image, cmap='gray')
        plt.title('original image')

        plt.subplot(10, 3, 2 + n*3)
        plt.imshow(hidden_image, cmap='gray')
        plt.title('hidden image')

        plt.subplot(10, 3, 3 + n*3)
        plt.imshow(gen_image, cmap='gray')
        plt.title('reconstructed image')

        print('generated images for digit %d' % (n))

    plt.tight_layout()
    plt.savefig('images/RBM_digits.jpg')


text:
# Import necessary packages here
import numpy as np
import pandas as pd
import sys
import os
import platform
#import yfinance as yf
sys.path.insert(0, os.path.abspath('../src'))
from core_utilities.plotting import two_d_line_matplot, two_d_scatter_matplot
from core_utilities.plotting import two_d_scatter_line_matplot, one_d_histogram_plot
from core_utilities.plotting import text_date_plot, MatPlotDataFrame
# ============================================================================
# ============================================================================
# Date:    December 18, 2020
# Purpose: This file contains functions that tests functions and classes
#          in the plotting.py file

# Source Code Metadata
__author__ = "Jonathan A. Webb"
__copyright__ = "Copyright 2020, Jon Webb Inc."
__version__ = "1.0"
# ============================================================================
# ============================================================================
# Test plotting routines in plotting.py

def test_date_plot():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/date.eps'
    else:
        plt_name = r'..\data\test\date.eps'
    # Use stock data for example
    tickers = ['AAPL', 'WMT']
    data = yf.download(tickers, '2015-1-1')['Adj Close']
    # transform Timestamps to string
    dates = list(data.index.strftime('%Y-%m-%d'))
    date_list = [dates, dates]
    y_list = [list(data[tickers[0]]), list(data[tickers[1]])]
    colors = ['red', 'green']
    line_style = ['-', '-']
    weight = [1.0, 1.0]
    text_date_plot(date_list, y_list, colors, line_style, weight, 'Date',
                   '$', tickers, 'upper left', save=True, plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ----------------------------------------------------------------------------

def test_two_d_line_plot():
    """

    This function tests the two_d_line_matplot function to ensure it
    produces a plot
    """
    x_dat = np.linspace(0, 10, 15)
    y1_dat = x_dat
    y2_dat = x_dat ** 2.0
    y3_dat = x_dat ** 3.0
    x_list = [x_dat, x_dat, x_dat]
    y_list = [y1_dat, y2_dat, y3_dat]
    colors = ['red', 'blue', 'black']
    line_style = ['-', '-', '--']
    labels = ['linear', 'squared', 'cubed']
    weight = [1, 2, 3]
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/test.png'
    else:
        plt_name = r'..\data\test\test.png'
    two_d_line_matplot(x_list, y_list, colors, line_style, weight, 'x-data',
                       'y-data', labels, 'upper left', save=True,
                       plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ----------------------------------------------------------------------------

def test_two_d_scatter_matplot():
    x_dat = np.linspace(0, 10, 15)
    y1_dat = x_dat
    y2_dat = x_dat ** 2.0
    y3_dat = x_dat ** 3.0
    x_list = [x_dat, x_dat, x_dat]
    y_list = [y1_dat, y2_dat, y3_dat]
    colors = ['red', 'blue', 'black']
    marker_style = ['^', 'o', 'd']
    labels = ['linear', 'squared', 'cubed']
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/test.eps'
    else:
        plt_name = r'..\data\test\test.eps'
    two_d_scatter_matplot(x_list, y_list, colors, marker_style, 'x-data',
                          'y-data', labels, 'upper left', save=True,
                          plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ----------------------------------------------------------------------------

def test_two_d_scatter_line_matplot():
    """

    This function tests the two_d_scatter_line_matplot function to determine if
    it generates an output file
    """
    x_dat = np.linspace(0, 10, 15)
    y1_dat = x_dat
    y2_dat = x_dat ** 2.0
    y3_dat = x_dat ** 3.0
    x_list = [x_dat, x_dat, x_dat]
    y_list = [y1_dat, y2_dat, y3_dat]
    colors = ['red', 'blue', 'black']
    line_style = ['-', '-', '--']
    labels = ['linear', 'squared', 'cubed']
    weight = [1, 2, 3]
    marker_style = ['^', 'o', 'd']
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/line_mark.eps'
    else:
        plt_name = r'..\data\test\line_mark.eps'
    two_d_scatter_line_matplot(x_list, y_list, colors, marker_style,
                               line_style, weight, 'x-axis', 'y-axis',
                               labels, 'upper left', save=True, plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ----------------------------------------------------------------------------

def test_histogram_plot():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/hist1.eps'
    else:
        plt_name = r'..\data\test\hist1.eps'
    np.random.seed(19680801)
    x = np.random.normal(15.0, 3.0, 1000)
    y = np.random.normal(20.0, 3.0, 1000)
    data = [x, y]
    labels = ['one', 'two']
    colors = ['blue', 'green']
    edge_colors = ['black', 'black']
    alpha = [0.9, 0.4]
    x_label = 'x-axis'
    y_label = 'y-axis'
    one_d_histogram_plot(data, labels, x_label, y_label, colors, edge_colors,
                         alpha, 'upper left', num_bins=50, save=True,
                         plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ================================================================================
# ================================================================================
# Test MatPlotDataFrame

def test_matplot_scatter_plot_parse_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/scatter1.eps'
    else:
        plt_name = r'..\data\test\scatter1.eps'
    length = 20
    x = np.linspace(0, 20, num=20)
    linear = x
    squared = x ** 2.0
    lin = np.repeat('linear', 20)
    sq = np.repeat('squared', 20)

    # Combine arrays into one
    x = np.hstack((x, x))
    y = np.hstack((linear, squared))
    power = np.hstack((lin, sq))

    # Create dataframe
    dictionary = {'x': x, 'y': y, 'power': power}
    df = pd.DataFrame(dictionary)

    # Plot data
    obj = MatPlotDataFrame(df)

    parsing_header = 'power'
    column_values = ['linear', 'squared']
    obj.scatter_plot_parse_column('x', 'y', parsing_header, column_values,
                                  x_label='x-axis', y_label='y-axis', title='Test',
                                  style_name='default',marker_colors=['red', 'green'],
                                  fill_alpha=0.7, marker_style=['o', '^'],
                                  label_pos='upper left', grid=True, save=True,
                                  plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_matplot_scatter_plot_columns():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/scatter2.eps'
    else:
        plt_name = r'..\data\test\scatter2.eps'
    length = 20
    x = np.linspace(0, 20, num=20)
    linear = x
    squared = x ** 2.0

    # create dataframe
    dictionary = {'x': x, 'linear': linear, 'squared': squared}
    df = pd.DataFrame(dictionary)

    # plot data
    obj = MatPlotDataFrame(df)
    x_headers = ['x', 'x']
    y_headers = ['linear', 'squared']
    obj.scatter_plot_columns(x_headers, y_headers, y_headers,
                             x_label='x-axis', y_label='y-axis', title='Test',
                             style_name='default',marker_colors=['red', 'green'],
                             fill_alpha=0.7, marker_style=['o', '^'],
                             label_pos='upper left', grid=False, save=True,
                             plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_matplot_line_plot_parse_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/line1.eps'
    else:
        plt_name = r'..\data\test\line.eps'
    length = 20
    x = np.linspace(0, 20, num=20)
    linear = x
    squared = x ** 2.0
    lin = np.repeat('linear', 20)
    sq = np.repeat('squared', 20)

    # Combine arrays into one
    x = np.hstack((x, x))
    y = np.hstack((linear, squared))
    power = np.hstack((lin, sq))

    # Create dataframe
    dictionary = {'x': x, 'y': y, 'power': power}
    df = pd.DataFrame(dictionary)

    # Plot data
    obj = MatPlotDataFrame(df)

    parsing_header = 'power'
    column_values = ['linear', 'squared']
    obj.line_plot_parse_column('x', 'y', parsing_header, column_values,
                               x_label='x-axis', y_label='y-axis', title='Test',
                               style_name='default', line_colors=['red', 'green'],
                               fill_alpha=0.7,
                               label_pos='upper left', grid=True, save=True,
                               plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_matplot_line_plot_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/line2.eps'
    else:
        plt_name = r'..\data\test\line2.eps'
    length = 20
    x = np.linspace(0, 20, num=20)
    linear = x
    squared = x ** 2.0

    # create dataframe
    dictionary = {'x': x, 'linear': linear, 'squared': squared}
    df = pd.DataFrame(dictionary)

    # plot data
    obj = MatPlotDataFrame(df)
    x_headers = ['x', 'x']
    y_headers = ['linear', 'squared']
    obj.line_plot_columns(x_headers, y_headers, y_headers,
                          x_label='x-axis', y_label='y-axis', title='Test',
                          style_name='default', line_colors=['red', 'green'],
                          fill_alpha=0.7,
                          label_pos='upper left', grid=True, save=True,
                          plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_matplot_timedate_plot_parse_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/time1.eps'
    else:
        plt_name = r'..\data\test\time.eps'
    length = 6
    dates = pd.date_range(start=pd.to_datetime('2016-09-24'),
                          periods = length, freq='y')
    x = np.linspace(0, length, num=length)
    linear = x
    squared = x ** 2.0
    lin = np.repeat('linear', length)
    sq = np.repeat('squared', length)

    # Combine arrays into one
    x = np.hstack((dates, dates))
    y = np.hstack((linear, squared))
    power = np.hstack((lin, sq))

    # Create dataframe
    dictionary = {'dates': x, 'y': y, 'power': power}
    df = pd.DataFrame(dictionary)
    # Plot data
    obj = MatPlotDataFrame(df)

    parsing_header = 'power'
    column_values = ['linear', 'squared']
    obj.timedate_plot_parse_column('dates', 'y', parsing_header, column_values,
                                   x_label='x-axis', y_label='y-axis', title='Test',
                                   style_name='default', line_colors=['red', 'green'],
                                   fill_alpha=0.7,
                                   label_pos='upper left', grid=True, save=True,
                                   plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_matplot_timedate_plot_parse_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/line1.eps'
    else:
        plt_name = r'..\data\test\line.eps'
    length = 6
    dates = pd.date_range(start=pd.to_datetime('2016-09-24'),
                          periods = length, freq='y')
    x = np.linspace(0, length, num=length)
    linear = x
    squared = x ** 2.0

    dictionary = {'dates': dates, 'squared': squared,
                  'linear': linear}
    df = pd.DataFrame(dictionary)
    # Plot data
    obj = MatPlotDataFrame(df)
    time_axis = ['dates', 'dates']
    y_axis = ['linear', 'squared']
    obj.timedate_plot_columns(time_axis, y_axis, y_axis,
                              x_label='x-axis', y_label='y-axis', title='Test',
                              style_name='default', line_colors=['red', 'green'],
                              fill_alpha=0.7,
                              label_pos='upper left', grid=True, save=True,
                              plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_histogram_plot_parse_column():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/hist1.eps'
    else:
        plt_name = r'..\data\test\hist1.eps'
    np.random.seed(19680801)
    x = np.random.normal(15.0, 3.0, 1000)
    y = np.random.normal(20.0, 3.0, 1000)
    data = [x, y]
    labels = ['one', 'two']
    one = np.repeat('one', len(x))
    two = np.repeat('two', len(x))
    x = np.hstack((x, y))
    y = np.hstack((one, two))

    dictionary = {'data': x, 'type': y}
    df = pd.DataFrame(dictionary)
    obj = MatPlotDataFrame(df)
    obj.histogram_plot_parse_column('data', 'type', labels, x_label='x-axis',
                                    y_label='y-axis', shading=[0.9, 0.4], save=True,
                                    plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# --------------------------------------------------------------------------------

def test_histogram_plot_columns():
    plat = platform.system()
    if plat == 'Darwin':
        plt_name = '../data/test/hist2.eps'
    else:
        plt_name = r'..\data\test\hist2.eps'
    np.random.seed(19680801)
    x = np.random.normal(15.0, 3.0, 1000)
    y = np.random.normal(20.0, 3.0, 1000)
    data = [x, y]
    labels = ['one', 'two']
    dictionary = {'x': x, 'y': y}
    df = pd.DataFrame(dictionary)
    obj = MatPlotDataFrame(df)
    obj.histogram_plot_columns(['x', 'y'], labels, x_label='x-axis',
                                y_label='y-axis', shading=[0.9, 0.4],
                                save=True, plot_name=plt_name)
    assert os.path.isfile(plt_name)
    if os.path.isfile(plt_name):
        os.remove(plt_name)
# ================================================================================
# ================================================================================
# eof


text:
"""Process Cascades

MIT License (MIT)

Copyright (c) 2015 Julien BLEGEAN <julien.blegean@aalto.fi>
"""

import networkx as nx
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import re
import datetime
import redis
import string
import numpy as np
import math
import Image
import community
import itertools
import os
from operator import itemgetter
import sys
from collections import Counter
import random
from correlations import correlations
import clusters

# add to graph
def process(current,rtlist) :
  """Process tweet to add his hashtags and user to the graphs
  :param tid : tweet id
  :param uid : user id
  :param tags : hashtags list
  """
  global graphs,found

  orig = rtlist[-1]

  rtlist.append(current)

  if orig not in graphs :
    graphs[orig] = nx.Graph()

  for node in rtlist :
    graphs[orig].add_node(node)

  for i in range(len(rtlist)-1):
    graphs[orig].add_edge(rtlist[i], rtlist[i+1])

# redis connection
rtc = redis.StrictRedis(host='localhost', port=6379, db=1)
r = redis.StrictRedis(host='localhost', port=6379, db=0)

found = 0
totalt = 0

# graph table
graphs = {}
tweets = {}

# input tweets
filepath = str(sys.argv[1])

# check time window
time = []
i = 0
for ar in sys.argv :
  i = i + 1
  if ar == "-h" :
    time.append(sys.argv[i])
    time.append(sys.argv[i+1])
if len(time) > 0 :
  dfrom = datetime.datetime.strptime(time[0], '%H:%M')
  dfrom = datetime.time(dfrom.hour,dfrom.minute)

  dto = datetime.datetime.strptime(time[1], '%H:%M')
  dto = datetime.time(dto.hour,dto.minute)

topusers = open("../../../data/users/users_fw_11_08_10msg+").readlines()
topusers = set([u.strip().split(",")[1] for u in topusers])

def username2id(id) :
  user = r.get(tweet[2])
  if user == None :
    return None
  user = re.findall('"((?:(?!(?:",")).)*)"', user)
  user = user[1].lower()
  return user

rtclusters = {}

# counter
k = 0

topusers11 = ["michaelskolnik","antoniofrench","pzfeed","pdpj","youranonnews","khaledbeydoun","womenonthemove1"]

# iterate tweets
with open(filepath) as f:
    for tweet in f:
      if math.fmod(k,100000) == 0 :
        print(k)

      # get the content
      tweet = re.findall('"((?:(?!(?:",")).)*)"', tweet)

      # parse date
      tdate = datetime.datetime.strptime(tweet[1], '%Y-%m-%d %H:%M:%S')
      tdate = datetime.time(tdate.hour,tdate.minute)

      # if the time is good
      if len(time) == 0 or len(time) > 0 and tdate >= dfrom and tdate <= dto :

        uname = username2id(tweet[2])
        current = (uname,len(tweets))
        tweets[(uname,tweet[3])] = current

        # check if it is an RT
        rtlist = []
        rt = re.findall(r"RT @([a-zA-Z0-9-_]*):? (.*)",tweet[3])

        while len(rt) > 0 :
          rt = rt[0]

          rtname = rt[0]
          mess = rt[1]

          if (rtname.lower(),mess) not in tweets :
            tweets[(rtname.lower(),mess)] = (rtname.lower(),len(tweets))
          rtlist.append(tweets[(rtname.lower(),mess)])

          rt = re.findall(r"RT @([a-zA-Z0-9-_]*):? (.*)",mess)

        if len(rtlist) > 0 :
          # add to graph
          if tweet[2] in topusers and rtlist[0][0] in topusers11 :
            if rtlist[0][0] not in rtclusters :
              rtclusters[rtlist[0][0]] = []
            rtclusters[rtlist[0][0]].append(uname)
          process(current,rtlist)
          found += 1

      k = k + 1

for c in rtclusters :
  rtclusters[c] = set(rtclusters[c])

def distJ(a,b) :
  return float(len(a & b)) / len(a | b)

distM = []

for u,i in zip(topusers11,range(len(topusers11))) :
  tmp = []
  for v,j in zip(topusers11,range(len(topusers11))) :
    dd = distJ(rtclusters[u],rtclusters[v])
    print(u)
    print(v)
    print(dd)
    tmp.append(dd)
  distM.append(tmp)

distM = np.array(distM)

print(distM)

#plt.clf()
#plt.figure(figsize=[7,7])
#plt.pcolor(distM,cmap=plt.cm.Blues)
#plt.colorbar(orientation='vertical')
#plt.savefig("heatmap11.png",dpi = 200)

graphs = sorted(graphs.items(), key=lambda x: len(x[1].nodes()), reverse = True)

if "-draw" in sys.argv :
  #for g in graphs[:100] :
    #print("%s : %s" %(g[0],len(g[1].nodes())))
  tweets = {v: k for k, v in tweets.items()}
  #graphs = graphs[:10]
  #print([tuple(tweets[g[0]]) for g in graphs])
  #G = nx.union_all([u[1] for u in graphs])
  print(tweets[graphs[0][0]])
  G = graphs[0][1]
  print("now create layout..")
  pos = nx.graphviz_layout(G,prog="sfdp")
  nx.draw(G,pos, linewidths=0, node_size = 5, with_labels = False, alpha = 0.5,font_size = 6, node_color='#862B59', edge_color='#cccccc',arrows=True)
  plt.legend(prop={'size':6})
  plt.savefig("graph.png",dpi=200)

if "-counter" in sys.argv :

  counter = Counter()

  for g in graphs[:100] :
    uname = g[0][0]
    counter[uname] += 1

  print(counter)

  vals = Counter([v[1] for v in counter.most_common()]).most_common()

  vals = sorted(vals,key=itemgetter(0), reverse=True)

  x = [v[1] for v in vals]
  y = [v[0] for v in vals]

  plt.clf()
  plt.figure(figsize=[7,7])
  plt.plot(x,y,'-o')
  plt.xlabel('number of users')
  plt.ylabel('number of tweets')
  plt.suptitle('User frequencies in the top 100 retweets')
  plt.savefig("freqInfluentUsers.png",dpi = 200)

max_stars_repo_name: jwheatp/twitter-riots

max_issues_repo_name: jwheatp/twitter-riots

max_forks_repo_name: jwheatp/twitter-riots

reason: import numpy,import networkx


text:
#!/usr/bin/env python3
#------------------------------------------------------------------------------#
#  fortnet-python: Python Tools for the Fortnet Software Package               #
#  Copyright (C) 2021 - 2022 T. W. van der Heide                               #
#                                                                              #
#  See the LICENSE file for terms of usage and distribution.                   #
#------------------------------------------------------------------------------#

'''
Regression tests covering the Fnetdata class of Fortformat.
'''

import os
import pytest
import numpy as np
from fortformat import Fnetdata

from common import Hdf5, get_cluster_geometries, get_bulk_geometries, \
    get_mixed_geometries, get_properties_byatoms, get_atomicweights_byatoms, \
    get_batomicweights_byatoms

REFPATH = os.path.join(os.getcwd(), 'test', 'references', 'Fnetdata')

def test_csgeometries(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'csgeometries.hdf5'
    csatoms = get_mixed_geometries()

    fnetdata = Fnetdata(atoms=csatoms)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries.hdf5'
    catoms = get_cluster_geometries()

    fnetdata = Fnetdata(atoms=catoms)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: No
        periodic: /
        global targets: No
        atomic targets: No
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'extfeatures.hdf5'
    features = []

    # fix random seed for reproduction purposes
    np.random.seed(42)

    for natom in [3, 7, 13]:
        features.append(np.random.random((natom, 4)))

    fnetdata = Fnetdata(features=features)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: No
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_extfeatures.hdf5'
    catoms = get_cluster_geometries()
    features = get_properties_byatoms(catoms, 5, True)

    fnetdata = Fnetdata(atoms=catoms, features=features)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: No
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'cgeometries_weights.hdf5'
    catoms = get_cluster_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(catoms), dtype=int)

    fnetdata = Fnetdata(atoms=catoms)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_atomictargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_atomictargets.hdf5'
    catoms = get_cluster_geometries()
    targets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, atomictargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_atomictargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: Yes
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_atomictargets_extfeatures.hdf5'
    catoms = get_cluster_geometries()
    targets = get_properties_byatoms(catoms, 3, True)
    features = get_properties_byatoms(catoms, 13, True)

    fnetdata = Fnetdata(atoms=catoms, features=features, atomictargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_atomictargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'cgeometries_atomictargets_weights.hdf5'
    catoms = get_cluster_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(catoms), dtype=int)
    targets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, atomictargets=targets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_atomictargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_atomictargets_atomicweights.hdf5'
    catoms = get_cluster_geometries()
    atomicweights = get_atomicweights_byatoms(catoms)
    targets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, atomictargets=targets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_atomictargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_atomictargets_batomicweights.hdf5'
    catoms = get_cluster_geometries()
    batomicweights = get_batomicweights_byatoms(catoms)
    targets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, atomictargets=targets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets.hdf5'
    catoms = get_cluster_geometries()
    targets = get_properties_byatoms(catoms, 3, False)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: No
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets_extfeatures.hdf5'
    catoms = get_cluster_geometries()
    targets = get_properties_byatoms(catoms, 3, False)
    features = get_properties_byatoms(catoms, 13, True)

    fnetdata = Fnetdata(atoms=catoms, features=features, globaltargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets_weights.hdf5'
    catoms = get_cluster_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(catoms), dtype=int)
    targets = get_properties_byatoms(catoms, 3, False)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=targets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_globaltargets_atomicweights.hdf5'
    catoms = get_cluster_geometries()
    atomicweights = get_atomicweights_byatoms(catoms)
    targets = get_properties_byatoms(catoms, 3, False)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=targets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_globaltargets_batomicweights.hdf5'
    catoms = get_cluster_geometries()
    batomicweights = get_batomicweights_byatoms(catoms)
    targets = get_properties_byatoms(catoms, 3, False)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=targets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: No
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_extfeatures.hdf5'
    satoms = get_bulk_geometries()
    features = get_properties_byatoms(satoms, 16, True)

    fnetdata = Fnetdata(atoms=satoms, features=features)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_atomictargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_atomictargets.hdf5'
    satoms = get_bulk_geometries()
    targets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, atomictargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_atomictargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: Yes
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_atomictargets_extfeatures.hdf5'
    satoms = get_bulk_geometries()
    targets = get_properties_byatoms(satoms, 3, True)
    features = get_properties_byatoms(satoms, 13, True)

    fnetdata = Fnetdata(atoms=satoms, features=features, atomictargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_atomictargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'sgeometries_atomictargets_weights.hdf5'
    satoms = get_bulk_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(satoms), dtype=int)
    targets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, atomictargets=targets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_atomictargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_atomictargets_atomicweights.hdf5'
    satoms = get_bulk_geometries()
    atomicweights = get_atomicweights_byatoms(satoms)
    targets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, atomictargets=targets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_atomictargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: No
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_atomictargets_batomicweights.hdf5'
    satoms = get_bulk_geometries()
    batomicweights = get_batomicweights_byatoms(satoms)
    targets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, atomictargets=targets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets.hdf5'
    satoms = get_bulk_geometries()
    targets = get_properties_byatoms(satoms, 3, False)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: No
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets_extfeatures.hdf5'
    satoms = get_bulk_geometries()
    targets = get_properties_byatoms(satoms, 3, False)
    features = get_properties_byatoms(satoms, 13, True)

    fnetdata = Fnetdata(atoms=satoms, features=features, globaltargets=targets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets_weights.hdf5'
    satoms = get_bulk_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(satoms), dtype=int)
    targets = get_properties_byatoms(satoms, 3, False)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=targets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_globaltargets_atomicweights.hdf5'
    satoms = get_bulk_geometries()
    atomicweights = get_atomicweights_byatoms(satoms)
    targets = get_properties_byatoms(satoms, 3, False)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=targets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: No
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_globaltargets_batomicweights.hdf5'
    satoms = get_bulk_geometries()
    batomicweights = get_batomicweights_byatoms(satoms)
    targets = get_properties_byatoms(satoms, 3, False)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=targets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

#######################################################################

def test_cgeometries_globaltargets_atomictargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets_atomictargets.hdf5'
    catoms = get_cluster_geometries()
    globaltargets = get_properties_byatoms(catoms, 3, False)
    atomictargets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_atomictargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: Yes
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets_atomictargets_extfeatures.hdf5'
    catoms = get_cluster_geometries()
    globaltargets = get_properties_byatoms(catoms, 3, False)
    atomictargets = get_properties_byatoms(catoms, 3, True)
    features = get_properties_byatoms(catoms, 13, True)

    fnetdata = Fnetdata(atoms=catoms, features=features,
                        globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_atomictargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'cgeometries_globaltargets_atomictargets_weights.hdf5'
    catoms = get_cluster_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(catoms), dtype=int)
    globaltargets = get_properties_byatoms(catoms, 3, False)
    atomictargets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_atomictargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_globaltargets_atomictargets_atomicweights.hdf5'
    catoms = get_cluster_geometries()
    atomicweights = get_atomicweights_byatoms(catoms)
    globaltargets = get_properties_byatoms(catoms, 3, False)
    atomictargets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_cgeometries_globaltargets_atomictargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: No
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'cgeometries_globaltargets_atomictargets_batomicweights.hdf5'
    catoms = get_cluster_geometries()
    batomicweights = get_batomicweights_byatoms(catoms)
    globaltargets = get_properties_byatoms(catoms, 3, False)
    atomictargets = get_properties_byatoms(catoms, 3, True)

    fnetdata = Fnetdata(atoms=catoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomictargets(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets_atomictargets.hdf5'
    satoms = get_bulk_geometries()
    globaltargets = get_properties_byatoms(satoms, 3, False)
    atomictargets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomictargets_extfeatures(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: Yes
        external features: Yes
        manual dataset weights: No
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets_atomictargets_extfeatures.hdf5'
    satoms = get_bulk_geometries()
    globaltargets = get_properties_byatoms(satoms, 3, False)
    atomictargets = get_properties_byatoms(satoms, 3, True)
    features = get_properties_byatoms(satoms, 13, True)

    fnetdata = Fnetdata(atoms=satoms, features=features,
                        globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomictargets_weights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: Yes
        manual gradient weights: No

    '''

    fname = 'sgeometries_globaltargets_atomictargets_weights.hdf5'
    satoms = get_bulk_geometries()
    # fix random seed for reproduction purposes
    np.random.seed(42)
    weights = np.random.randint(1, 100, len(satoms), dtype=int)
    globaltargets = get_properties_byatoms(satoms, 3, False)
    atomictargets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.weights = weights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomictargets_atomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_globaltargets_atomictargets_atomicweights.hdf5'
    satoms = get_bulk_geometries()
    atomicweights = get_atomicweights_byatoms(satoms)
    globaltargets = get_properties_byatoms(satoms, 3, False)
    atomictargets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.atomicweights = atomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

def test_sgeometries_globaltargets_atomictargets_batomicweights(tmpdir):
    '''Test dataset generation for configuration:

        structures: Yes
        periodic: Yes
        global targets: Yes
        atomic targets: Yes
        external features: No
        manual dataset weights: No
        manual gradient weights: Yes

    '''

    fname = 'sgeometries_globaltargets_atomictargets_batomicweights.hdf5'
    satoms = get_bulk_geometries()
    batomicweights = get_batomicweights_byatoms(satoms)
    globaltargets = get_properties_byatoms(satoms, 3, False)
    atomictargets = get_properties_byatoms(satoms, 3, True)

    fnetdata = Fnetdata(atoms=satoms, globaltargets=globaltargets,
                        atomictargets=atomictargets)
    fnetdata.atomicweights = batomicweights
    fnetdata.dump(os.path.join(tmpdir, fname))

    hdf5 = Hdf5(os.path.join(tmpdir, fname))
    equal = hdf5.equals(os.path.join(REFPATH, '_' + fname))

    assert equal, 'h5diff reports mismatch in generated datasets.'

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

max_stars_repo_name: vanderhe/fortnet-python

max_issues_repo_name: vanderhe/fortnet-python

max_forks_repo_name: vanderhe/fortnet-python


text:
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# (C) British Crown Copyright 2017-2021 Met Office.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# * Redistributions of source code must retain the above copyright notice, this
#   list of conditions and the following disclaimer.
#
# * Redistributions in binary form must reproduce the above copyright notice,
#   this list of conditions and the following disclaimer in the documentation
#   and/or other materials provided with the distribution.
#
# * Neither the name of the copyright holder nor the names of its
#   contributors may be used to endorse or promote products derived from
#   this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
"""Unit tests for the RegridWithLandSeaMask class"""
# set up a special data set and corresponding land-sea mask info
# set up target grid and its land-sea mask info
# it is designed to cover different scenarios for regridding with land-sea
# the regridding reference results are manually checked for different methods
# not using "set_up_variable_cube" because of different spacing at lat/lon

import numpy as np

from improver.regrid.bilinear import basic_indexes
from improver.regrid.grid import calculate_input_grid_spacing, latlon_from_cube
from improver.regrid.landsea import RegridLandSea
from improver.synthetic_data.set_up_test_cubes import set_up_variable_cube

def modify_cube_coordinate_value(cube, coord_x, coord_y):
    """modify x(longitude) & y(latitude) andcoordinates for a cube"""
    cube.coord(axis="x").points = coord_x
    cube.coord(axis="x").bounds = None
    cube.coord(axis="x").guess_bounds()
    cube.coord(axis="y").points = coord_y
    cube.coord(axis="y").bounds = None
    cube.coord(axis="y").guess_bounds()
    return cube

def define_source_target_grid_data():
    """ define cube_in, cube_in_mask,cube_out_mask using assumed data """
    # source (input) grid
    in_lats = np.linspace(0, 15, 4)
    in_lons = np.linspace(0, 40, 5)

    # target (output) grid
    out_lats = np.linspace(0, 14, 8)
    out_lons = np.linspace(5, 35, 11)

    # assume a set of nwp data
    data = np.arange(20).reshape(4, 5).astype(np.float32)

    # input grid mask info
    in_mask = np.empty((4, 5), dtype=np.int)
    in_mask[:, :] = 1
    in_mask[0, 2] = 0
    in_mask[2, 2:4] = 0
    in_mask[3, 2:4] = 0

    # output grid mask info
    out_mask = np.empty((8, 11), dtype=np.int)
    out_mask[:, :] = 1
    out_mask[0, 4:7] = 0
    out_mask[1, 5] = 0
    out_mask[5:9, 4:10] = 0

    out_mask[6, 6] = 1
    out_mask[7, 6] = 1
    out_mask[1, 0] = 0

    # create cube with default spacing
    cube_in = set_up_variable_cube(data, "air_temperature", "Celsius")
    cube_in_mask = set_up_variable_cube(in_mask, "Land_Binary_Mask", "1")
    cube_out_mask = set_up_variable_cube(out_mask, "Land_Binary_Mask", "1")

    # modify cube coordinates to the designed value
    cube_in = modify_cube_coordinate_value(cube_in, in_lons, in_lats)
    cube_in_mask = modify_cube_coordinate_value(cube_in_mask, in_lons, in_lats)
    cube_out_mask = modify_cube_coordinate_value(cube_out_mask, out_lons, out_lats)

    return cube_in, cube_out_mask, cube_in_mask

def define_source_target_grid_data_same_domain():
    """ define cube_in, cube_in_mask,cube_out_mask, assume the same domain  """
    # source (input) grid
    in_lats = np.linspace(0, 15, 4)
    in_lons = np.linspace(0, 40, 5)

    # target (output) grid
    out_lats = np.linspace(0, 15, 7)
    out_lons = np.linspace(5, 40, 9)

    # assume a set of nwp data
    data = np.arange(20).reshape(4, 5).astype(np.float32)

    # input grid mask info
    in_mask = np.empty((4, 5), dtype=np.int)
    in_mask[:, :] = 1
    in_mask[0, 2] = 0
    in_mask[2, 2:4] = 0
    in_mask[3, 2:4] = 0

    # output grid mask info
    out_mask = np.empty((7, 9), dtype=np.int)
    out_mask[:, :] = 1
    out_mask[0, 3:6] = 0
    out_mask[1, 4] = 0
    out_mask[4:9, 4:8] = 0
    out_mask[6, 6] = 1
    out_mask[1, 0] = 0

    # create cube with default spacing
    cube_in = set_up_variable_cube(data, "air_temperature", "Celsius")
    cube_in_mask = set_up_variable_cube(in_mask, "Land_Binary_Mask", "1")
    cube_out_mask = set_up_variable_cube(out_mask, "Land_Binary_Mask", "1")

    # modify cube coordinates to the designed value
    cube_in = modify_cube_coordinate_value(cube_in, in_lons, in_lats)
    cube_in_mask = modify_cube_coordinate_value(cube_in_mask, in_lons, in_lats)
    cube_out_mask = modify_cube_coordinate_value(cube_out_mask, out_lons, out_lats)

    return cube_in, cube_out_mask, cube_in_mask

def test_basic_indexes():
    """Test basic_indexes for identical source and target domain case """
    cube_in, cube_out_mask, _ = define_source_target_grid_data_same_domain()
    in_latlons = latlon_from_cube(cube_in)
    out_latlons = latlon_from_cube(cube_out_mask)
    in_lons_size = cube_in.coord(axis="x").shape[0]
    lat_spacing, lon_spacing = calculate_input_grid_spacing(cube_in)
    indexes = basic_indexes(
        out_latlons, in_latlons, in_lons_size, lat_spacing, lon_spacing
    )
    test_results = indexes[58:63, :]
    expected_results = np.array(
        [
            [12, 17, 18, 13],
            [12, 17, 18, 13],
            [13, 18, 19, 14],
            [13, 18, 19, 14],
            [13, 18, 19, 14],
        ]
    )
    np.testing.assert_array_equal(test_results, expected_results)

def test_regrid_nearest_2():
    """Test nearest neighbour regridding option 'nearest-2'"""

    cube_in, cube_out_mask, _ = define_source_target_grid_data()
    regrid_nearest = RegridLandSea(regrid_mode="nearest-2",)(cube_in, cube_out_mask)
    expected_results = np.array(
        [
            [0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 3],
            [0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 3],
            [5, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8],
            [5, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8],
            [10, 11, 11, 11, 12, 12, 12, 13, 13, 13, 13],
            [10, 11, 11, 11, 12, 12, 12, 13, 13, 13, 13],
            [10, 11, 11, 11, 12, 12, 12, 13, 13, 13, 13],
            [15, 16, 16, 16, 17, 17, 17, 18, 18, 18, 18],
        ]
    )
    np.testing.assert_allclose(regrid_nearest.data, expected_results, atol=1e-3)

def test_regrid_bilinear_2():
    """Test bilinear regridding option 'bilinear-2'"""

    cube_in, cube_out_mask, _ = define_source_target_grid_data()
    regrid_bilinear = RegridLandSea(regrid_mode="bilinear-2",)(cube_in, cube_out_mask)

    expected_results = np.array(
        [
            [0.5, 0.8, 1.1, 1.4, 1.7, 2.0, 2.3, 2.6, 2.9, 3.2, 3.5],
            [2.5, 2.8, 3.1, 3.4, 3.7, 4.0, 4.3, 4.6, 4.9, 5.2, 5.5],
            [4.5, 4.8, 5.1, 5.4, 5.7, 6.0, 6.3, 6.6, 6.9, 7.2, 7.5],
            [6.5, 6.8, 7.1, 7.4, 7.7, 8.0, 8.3, 8.6, 8.9, 9.2, 9.5],
            [8.5, 8.8, 9.1, 9.4, 9.7, 10.0, 10.3, 10.6, 10.9, 11.2, 11.5],
            [10.5, 10.8, 11.1, 11.4, 11.7, 12.0, 12.3, 12.6, 12.9, 13.2, 13.5],
            [12.5, 12.8, 13.1, 13.4, 13.7, 14.0, 14.3, 14.6, 14.9, 15.2, 15.5],
            [14.5, 14.8, 15.1, 15.4, 15.7, 16.0, 16.3, 16.6, 16.9, 17.2, 17.5],
        ]
    )

    np.testing.assert_allclose(regrid_bilinear.data, expected_results, atol=1e-3)

def test_regrid_nearest_with_mask_2():
    """Test nearest-with-mask-2 regridding"""

    cube_in, cube_out_mask, cube_in_mask = define_source_target_grid_data()
    regrid_nearest_with_mask = RegridLandSea(
        regrid_mode="nearest-with-mask-2",
        landmask=cube_in_mask,
        landmask_vicinity=250000000,
    )(cube_in, cube_out_mask)

    expected_results = np.array(
        [
            [0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 3],
            [0, 1, 1, 1, 7, 2, 7, 3, 3, 3, 3],
            [5, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8],
            [5, 6, 6, 6, 7, 7, 7, 8, 8, 8, 9],
            [10, 11, 11, 11, 7, 7, 7, 8, 8, 8, 14],
            [10, 11, 11, 11, 12, 12, 12, 13, 13, 13, 14],
            [10, 11, 11, 11, 12, 12, 7, 13, 13, 13, 14],
            [15, 16, 16, 16, 17, 17, 7, 18, 18, 18, 19],
        ]
    )

    np.testing.assert_allclose(
        regrid_nearest_with_mask.data, expected_results, atol=1e-3
    )

    # consider constant field
    cube_in.data = np.repeat(1.0, 20).reshape(4, 5).astype(np.float32)
    regrid_nearest_with_mask = RegridLandSea(
        regrid_mode="nearest-with-mask-2",
        landmask=cube_in_mask,
        landmask_vicinity=250000000,
    )(cube_in, cube_out_mask)

    expected_results = np.repeat(1.0, 88).reshape(8, 11).astype(np.float32)
    np.testing.assert_allclose(
        regrid_nearest_with_mask.data, expected_results, atol=1e-3
    )

def test_regrid_bilinear_with_mask_2():
    """Test bilinear-with-mask-2 regridding """

    cube_in, cube_out_mask, cube_in_mask = define_source_target_grid_data()
    regrid_bilinear_with_mask = RegridLandSea(
        regrid_mode="bilinear-with-mask-2",
        landmask=cube_in_mask,
        landmask_vicinity=250000000,
    )(cube_in, cube_out_mask)

    expected_results = np.array(
        [
            [0.5, 0.8, 1.40096, 3.2916, 2.0, 2.0, 2.0, 4.94333, 3.25586, 3.2, 3.5],
            [2.5, 2.8, 3.1, 3.4, 5.48911, 2.76267, 6.32926, 4.6, 4.9, 5.2, 5.5],
            [4.5, 4.8, 5.1, 5.4, 5.7, 7.0154, 6.3, 6.6, 6.9, 7.2, 7.5],
            [6.5, 6.8, 7.1, 7.4, 7.7, 7.0, 7.19033, 7.6681, 7.6618, 9.2, 9.5],
            [
                8.5,
                8.8,
                9.1,
                9.4,
                8.10633,
                7.0,
                7.0,
                7.62915,
                7.21672,
                9.11434,
                10.52363,
            ],
            [
                10.5,
                10.8,
                11.00012,
                11.01183,
                13.15439,
                12.0,
                12.3,
                12.6,
                12.9,
                13.71286,
                15.74504,
            ],
            [
                12.5,
                12.8,
                12.23411,
                13.25881,
                14.14155,
                14.0,
                8.07328,
                14.6,
                14.9,
                14.96332,
                16.3334,
            ],
            [
                14.5,
                14.8,
                15.0997,
                14.22659,
                15.50905,
                16.0,
                9.8733,
                16.6,
                16.9,
                16.91114,
                17.03773,
            ],
        ]
    )

    np.testing.assert_allclose(
        regrid_bilinear_with_mask.data, expected_results, atol=1e-3
    )

    # consider constant field
    cube_in.data = np.repeat(1.0, 20).reshape(4, 5).astype(np.float32)
    regrid_bilinear_with_mask = RegridLandSea(
        regrid_mode="bilinear-with-mask-2",
        landmask=cube_in_mask,
        landmask_vicinity=250000000,
    )(cube_in, cube_out_mask)

    expected_results = np.repeat(1.0, 88).reshape(8, 11).astype(np.float32)

    np.testing.assert_allclose(
        regrid_bilinear_with_mask.data, expected_results, atol=1e-3
    )

max_stars_repo_name: yzhaobom/improver

max_issues_repo_name: yzhaobom/improver

max_forks_repo_name: MoseleyS/improver