text:
from PIL import Image, ImageOps
import numpy as np
import skimage.io as io
from src.models.class_patcher import patcher
from src.utils.imgproc import *
from skimage.color import rgb2hsv, hsv2rgb

class patcher(patcher):
    def __init__(self, body='./body/body_inakokoa.png', **options):
        super().__init__('いな屋さんのここあ下着', body=body, pantie_position=[48, 1117], **options)
        self.mask = io.imread('./mask/mask_inakokoa.png')

        self.bra_position = [0, 0]
        self.bra = np.float32(io.imread('./mask/bra_inakokoa.png') / 255)
        self.bra_center = np.float32(io.imread('./mask/bra_inakokoa_center.png') / 255)
        self.bra_shade = np.float32(io.imread('./material/bra_inakokoa_shade.png') / 255)
        self.bra_lace = np.float32(io.imread('./material/bra_inakokoa_lace.png') / 255)
        self.bra_shade_alpha = self.bra_shade[:, :, -1]
        self.bra_lace_mask = self.bra_lace[:, :, -1] > 0.3

        self.ribbon_position = [882, 685]
        self.ribbon = np.float32(io.imread('./material/ribbon_inakokoa.png') / 255)
        self.ribbon_shade = np.float32(io.imread('./material/ribbon_inakokoa_shade.png') / 255)
        self.ribbon_shade_alpha = self.ribbon_shade[:, :, -1]

        self.extra_position = [630, 1410]

    def pick_color(self, arr):
        return np.mean(np.mean(arr, axis=0), axis=0)

    def extract_base_color(self, pantie):
        front = pantie[20:100, 30:80, :3] / 255.0
        front_shade = pantie[130:150, 0:40, :3] / 255.0
        front_color = self.pick_color(front)
        front_shade_color = self.pick_color(front_shade)
        front_shade_color = rgb2hsv(front_shade_color[None, None])
        front_shade_color[0, 0, 1] *= front_shade_color[0, 0, 2] / 0.3
        front_shade_color[0, 0, 1] *= 4
        front_shade_color[0, 0, 2] *= front_shade_color[0, 0, 2] / 0.2
        front_shade_color = np.clip(hsv2rgb(front_shade_color)[0, 0], 0, 1)
        return front_color, front_shade_color

    def gen_ribbon(self, image):
        pantie = np.array(image)
        ribbon = pantie[24:32, 15:27, :3] / 255.0
        ribbon_color = self.pick_color(ribbon)
        ribbon_shade = pantie[26:30, 12:15, :3] / 255.0
        ribbon_shade_color = self.pick_color(ribbon_shade)
        ribbon = self.ribbon[:, :, :3] * ribbon_color
        ribbon_shade = (self.ribbon_shade[:, :, -1])[:, :, None] * ribbon_shade_color
        ribbon = alpha_brend(ribbon_shade, ribbon, self.ribbon_shade_alpha)
        ribbon = np.dstack((ribbon, self.ribbon[:, :, -1] > 0.5))
        return Image.fromarray(np.uint8(np.clip(ribbon, 0, 1) * 255))

    def gen_extra(self, image):
        pantie = np.array(image)
        ribbon = pantie[24:32, 15:27, :3] / 255.0
        ribbon_color = self.pick_color(ribbon)
        extra = np.ones((600, 600, 4), dtype=np.float32)
        extra[:, :, :3] *= ribbon_color
        return Image.fromarray(np.uint8(extra * 255))

    def gen_bra(self, image):
        pantie = np.array(image)
        front_color, front_shade_color = self.extract_base_color(pantie)
        ribbon = pantie[24:32, 15:27, :3] / 255.0
        ribbon_color = self.pick_color(ribbon)
        center = np.float32(pantie[20:170, -250:-15, :3][:, ::-1]) / 255
        center = resize(center, [4.3, 4.3])
        bra_center = np.copy(self.bra_center)
        bra_center[150:150 + center.shape[0], 100:100 + center.shape[1], :3] = center * np.float32(bra_center[150:150 + center.shape[0], 100:100 + center.shape[1], :3] > 0)
        bra = self.bra[:, :, :3] * front_color
        bra_shade = (self.bra_shade[:, :, -1])[:, :, None] * front_shade_color
        bra_lace = self.bra_lace[:, :, :3] * ribbon_color
        bra = alpha_brend(bra_center[:, :, :3], bra[:, :, :3], bra_center[:, :, 0] > 0.1)
        bra = alpha_brend(bra_lace, bra, self.bra_lace_mask)
        bra = alpha_brend(bra_shade, bra, self.bra_shade_alpha)
        bra = np.dstack((bra, self.bra[:, :, 0]))
        return Image.fromarray(np.uint8(np.clip(bra, 0, 1) * 255))

    def convert(self, image):
        pantie = np.array(image)
        pantie = ribbon_inpaint(pantie)
        patch = np.copy(pantie[-190:-5, 546:, :])
        [pr, pc, d] = patch.shape
        pantie[120:120 + pr, :pc, :] = patch[::-1, ::-1]
        pantie[165:165 + pr, -pc:, :] = patch[::-1]
        pantie = pantie[:220]
        arrx = np.zeros(100)
        arry = np.zeros(100)
        arry[10:] += np.linspace(0, 1, 90)**1 * 230
        arry[10:30] += np.sin(np.linspace(0, np.pi, 20))**1 * -30
        pantie = affine_transform_by_arr(pantie, arrx, arry)
        pantie = resize(pantie[:, :470], [4.18, 4.18])
        pantie = np.uint8(pantie * 255)
        pantie = np.bitwise_and(pantie, self.mask)
        return Image.fromarray(pantie)

    def patch(self, image, transparent=False):
        pantie = self.convert(image)
        if transparent:
            patched = Image.new("RGBA", (2048, 2048))
        else:
            patched = self.body.copy()
        patched = self.paste(patched, pantie, self.pantie_position)
        patched = self.paste(patched, self.gen_bra(image), self.bra_position)
        patched = self.paste(patched, self.gen_ribbon(image), self.ribbon_position)
        patched = self.paste(patched, self.gen_extra(image), self.extra_position)
        return patched


text:
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

class SaveResults:
    def __init__(self, name):
        self.test_name = name
        self.results = {}

    def save_bearing_array(self, bearing_array):
        with open("bearings.txt", "ab") as f:
            np.savetxt(f, np.degrees(bearing_array), fmt="%.1f", newline=" ")
            f.write(b"\n")

    def save_distance_array(self, distance_array):
        with open("distances.txt", "ab") as f:
            np.savetxt(f, distance_array / 1000, fmt="%.3f", newline=" ")
            f.write(b"\n")

    def save_bearings_fig(self, true_distances, pred_distances):
        plt.plot(true_distances, linewidth=5.0)
        plt.plot(2)
        ax = plt.gca()
        ax.set_xlabel("время (с)")
        ax.set_ylabel("дистанция (м)")
        ax.legend(["Истинное расстояние", "Расчетное расстояние"])
        plt.xlim([0, len(true_distances) - 1])
        plt.savefig("Пеленги" + self.test_name + ".png")

ext: py

reason: import numpy


text:
import os
import re
import numpy as np

def line2float(line):
    data=list()
    newstr=str()
    data.append(newstr)
    index=0
    for i in line:
        if((i<='9'and i>='0') or i =='.'):
           data[index]+=i
        elif(i ==','):
            index+=1
            newstr=str()
            data.append(newstr)
    for i in range(len(data)):
        data[i]=float(data[i])
    return data

max_stars_repo_name: Shiduo-zh/pybulletSim

max_issues_repo_name: Shiduo-zh/pybulletSim

max_forks_repo_name: Shiduo-zh/pybulletSim


text:
import cv2
import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import re
import shutil
import glob
import json

def load_video_names(paths):
    save_folder = 'raw_video'
    filepath = os.path.join(paths[save_folder], "searched_videos")
    with open(filepath, "r") as f:
        videos = list(json.load(f))
        return videos

def load_videos_into_np(folder):
    # Load files into a dict of numpy arrays using opencv
    video_dict = {}
    for filename in glob.glob(folder + '*.mp4'):
        try:
            video_name = re.split(r"\\|/", filename)[-1]
            video_dict[video_name] = mp4_to_npy(filename)
        except Exception as e:
            print(f"Could not convert {filename}  to numpy array due to {e}")

    return video_dict

def download_video_and_convert_to_numpy(local_folder, s3_profile, bucket, filenames: list):
    """Downloads videos from s3 to a local temp directory and then loads them into numpy arrays, before
    deleting the temp directory (default behavior).

    Args:
        local_folder: local folder to temporarily download the videos to.
        s3_profile: s3 profile on amazon aws
        bucket: bucket name where the videos are stored
        filenames: list of filenames to download from s3
    Returns:
        video_dict: dictionary of numpy arrays containing all the jamcam videos between the selected dates,
                    with the key being the filename
    Raises:

    """
    my_bucket = connect_to_bucket(s3_profile, bucket)
    delete_and_recreate_dir(local_folder)

    # Download the files
    for filename in filenames:
        try:
            my_bucket.download_blob(filename, local_folder + filename.split('/')[-1].replace(
                ':', '-').replace(" ", "_"))
        except:
            print("Could not download " + filename)
    return load_videos_into_np(local_folder)

def delete_and_recreate_dir(temp_dir):
    """
    Creates an empty local directory for downloading from s3.
    Will wipe local_dir if already a directory
    """
    if os.path.isdir(temp_dir):
        shutil.rmtree(temp_dir)

    os.makedirs(temp_dir, exist_ok=True)
    return

def mp4_to_npy(local_mp4_path):
    """Load mp4 filename from the local directory and turn it into a numpy array"""
    cap = cv2.VideoCapture(local_mp4_path)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    buf = np.empty((frame_count, frame_height, frame_width, 3),
                   np.dtype('uint8'))
    fc = 0
    ret = True
    while fc < frame_count and ret:
        ret, buf[fc] = cap.read()
        fc += 1
    cap.release()

    if (buf.size == 0):
        raise Exception('Numpy array is empty')

    return buf

def connect_to_bucket(profile_dir, bucket_name):
    """Connects to the s3 bucket"""
    # Set up boto3 session
    s3_session = boto3.Session(profile_name=profile_dir)
    s3_resource = s3_session.resource('s3')
    my_bucket = s3_resource.Bucket(bucket_name)

    return my_bucket

def describe_s3_bucket(paths):
    """Plot the number of videos in the s3 bucket for each date.
    Plot is saved locally under plots/01_exploratory.
    """
    my_bucket = connect_to_bucket(paths['s3_profile'], paths['bucket_name'])

    # Get list of all dates in the s3 bucket
    objects = my_bucket.objects.filter(Prefix="raw/video_data_new/")
    files = [obj.key for obj in objects]
    dates = []
    for filename in files:
        res = re.search(r"([0-9]{4}\-[0-9]{2}\-[0-9]{2})", filename)
        dates.append(res.group())

    # Plot a bar chart of the dates in the s3 bucket
    unique_dates, counts = np.unique(dates, return_counts=True)
    plt.figure()
    plt.bar(np.arange(unique_dates.shape[0]), counts)
    plt.xticks(np.arange(unique_dates.shape[0]),
               unique_dates, rotation='vertical')
    plt.xlabel('Date')
    plt.ylabel('Number of Videos')
    plt.tight_layout()
    plt.savefig(paths['plots'] + '01_exploratory/s3_description.pdf')
    plt.close()

    return


text:
from sympy import symbols
from sympy.physics.continuum_mechanics.beam import Beam
E, I = symbols('E, I')
R_0, R_8 = symbols('R_0, R_8')
b = Beam(12, E, I)
b.apply_support(0, 'roller')
b.apply_support(8, 'roller')
b.solve_for_ild_reactions(1, R_0, R_8)
b.solve_for_ild_shear(4, 1, R_0, R_8)
b.ild_shear
# Piecewise((x/8, x < 4), (x/8 - 1, x > 4))
b.plot_ild_shear()
# Plot object containing:
# [0]: cartesian line: Piecewise((x/8, x < 4), (x/8 - 1, x > 4)) for x over (0.0, 12.0)

reason: from sympy


text:
### Import basic packages
import pandas as pd
import os
from datetime import datetime as dt
from datetime import timedelta

###Import package used to get features timeseries
#tsfresh is a python package. It automatically calculates a large number of time series characteristics, the so called features.
#Further the package contains methods to evaluate the explaining power and importance of such characteristics for regression or classification tasks.
from tsfresh.utilities.dataframe_functions import roll_time_series
from tsfresh import extract_features
from tsfresh.feature_extraction import MinimalFCParameters

from statsmodels.tsa.stattools import adfuller, kpss
from statsmodels.tsa.stattools import acf, pacf
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf

def get_resample_features(data, window, settings=MinimalFCParameters()):
    """ Make rolling in time series to extrated daily features
	The number of windows to make the rolling and in this way get daily features

    Parameters
    ----------
    data : DataFrame
        The DataFrame contains events update by CI
    settings : Object
        A object that maps feature calculator names in tsfresh. There are
        two options: ComprehensiveFCParameters() or MinimalFCParameters()
	window : int

    Returns
    -------
    DataFrame
        DataFrame with daily features
    """

    data["id"] = 1
    df_roll_time = roll_time_series(
                                    data,
                                    column_id="id",
                                    column_sort='Timestamp',
                                    column_kind=None,
                                    rolling_direction=1,
                                    max_timeshift=window
                                    )

    X_features = extract_features(df_roll_time,
                                  column_id="id",
                                  column_sort='Timestamp',
                                  default_fc_parameters=settings,
  #                                n_jobs=4
                                 )

    X_features.index = pd.to_datetime(X_features.index)
    return X_features

def get_features_lag(df, lag=1):
	"""Get the time series lag

	Parameters
	----------
	df : dataframe
		DataFrame with the time series to lag
	lag : int
		Number of times the series will lag behind

	Returns
	-------
		df_lag
			A DataFrame whit the lags of time series
	"""

	df_lag = df.shift(lag)
	df_lag.columns = ["{}_lag_{}".format(column, lag) for column in df_lag.columns]
	return df_lag

def extraction_features_ts(df, window, settings, TARGET, list_features):

	"""Feature extraction with tsfresh package

	Parameters
	----------
	df : dataframe
		DataFrame with the time series to extraction time
	window : int
		Number of windows to calculate features
	settings : str
	TARGET : str
	list_features : list

	Returns
	-------
		df_lag
			A DataFrame whit feature extraction
	"""

	df_features = df.filter(list_features)

	df_features = get_resample_features(df_features, window, settings). \
	  drop(columns=["{}__length".format(TARGET), "{}__sum_values".format(TARGET)])

	df = df.set_index("Timestamp")

	df_features = pd.concat([df_features, df[TARGET]], axis=1)

	###Get lag 1
	df_features_lag = pd.concat([get_features_lag(df_features, lag=1), df_features[TARGET]], axis=1).dropna()
	###Get lag 2
	df_features_lag = pd.concat([get_features_lag(df_features, lag=2), df_features_lag], axis=1).dropna()

	return df_features_lag

max_stars_repo_name: MFrancys/ai-platform

max_issues_repo_name: MFrancys/ai-platform

max_forks_repo_name: MFrancys/ai-platform

reason: from statsmodels


text:
import numpy as np
import os
from PIL import Image
import sys
import torchvision
import pathlib
import requests

from datasets import base
from platforms.platform import get_platform

class CIFAR10(torchvision.datasets.CIFAR10):
    """A subclass to suppress an annoying print statement in the torchvision CIFAR-10 library.

    Not strictly necessary - you can just use `torchvision.datasets.CIFAR10 if the print
    message doesn't bother you.
    """

    def download(self):
        if get_platform().is_primary_process:
            with get_platform().open(os.devnull, 'w') as fp:
                sys.stdout = fp
                super(CIFAR10, self).download()
                sys.stdout = sys.__stdout__
        get_platform().barrier()

# Modified from https://github.com/modestyachts/CIFAR-10.1
def load_new_test_data():
    data_path = get_platform().dataset_root
    filename = os.path.join('cifar10.1', 'v6')
    label_filename = filename + '_labels.npy'
    imagedata_filename = filename + '_data.npy'
    label_filepath = os.path.abspath(os.path.join(data_path, label_filename))
    imagedata_filepath = os.path.abspath(os.path.join(data_path, imagedata_filename))
    # Download the data if it isn't already there
    if not pathlib.Path(label_filepath).is_file():
        url = 'https://raw.github.com/modestyachts/CIFAR-10.1/master/datasets/cifar10.1_v6_labels.npy'
        r = requests.get(url)
        os.makedirs(os.path.dirname(label_filepath), exist_ok=True)
        f = open(label_filepath, 'wb')
        f.write(r.content)
    if not pathlib.Path(imagedata_filepath).is_file():
        url = 'https://raw.github.com/modestyachts/CIFAR-10.1/master/datasets/cifar10.1_v6_data.npy'
        r = requests.get(url)
        os.makedirs(os.path.dirname(imagedata_filepath), exist_ok=True)
        f = open(imagedata_filepath, 'wb')
        f.write(r.content)
    assert pathlib.Path(label_filepath).is_file()
    labels = np.load(label_filepath)
    assert pathlib.Path(imagedata_filepath).is_file()
    imagedata = np.load(imagedata_filepath)
    assert len(labels.shape) == 1
    assert len(imagedata.shape) == 4
    assert labels.shape[0] == imagedata.shape[0]
    assert imagedata.shape[1] == 32
    assert imagedata.shape[2] == 32
    assert imagedata.shape[3] == 3
    assert labels.shape[0] == 2000
    return imagedata, labels

class Dataset(base.ImageDataset):
    """The CIFAR-10 train set with the CIFAR-10.1 test set."""

    @staticmethod
    def num_train_examples(): return 50000

    @staticmethod
    def num_test_examples(): return 2000

    @staticmethod
    def num_classes(): return 10

    @staticmethod
    def get_train_set(use_augmentation):
        augment = [torchvision.transforms.RandomHorizontalFlip(), torchvision.transforms.RandomCrop(32, 4)]
        train_set = CIFAR10(train=True, root=os.path.join(get_platform().dataset_root, 'cifar10'), download=True)
        return Dataset(train_set.data, np.array(train_set.targets), augment if use_augmentation else [])

    @staticmethod
    def get_test_set():
        data, labels = load_new_test_data()
        return Dataset(data, labels.astype(np.int64))

    def __init__(self,  examples, labels, image_transforms=None):
        super(Dataset, self).__init__(examples, labels, image_transforms or [],
                                      [torchvision.transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])])

    def example_to_image(self, example):
        return Image.fromarray(example)

DataLoader = base.DataLoader


text:
"""
Behavioural Cloning agent that trains on MineRLTreechop data. It is then evaluated on MineRLObtainDiamond by running it
for a certain number of steps and then switching to the scripted part that crafts a wooden_pickaxe and digs down to get
some cobblestone.
With default parameters it trains in 5-10 mins on a machine with a GeForce RTX 2080 Ti GPU.
It uses less than 8GB RAM and achieves an average reward of 8.6.
"""

from tqdm import tqdm
import numpy as np
import torch as th
from torch import nn
import gym
import minerl

# Parameters:
DATA_DIR = "data"  # path to where MineRL dataset resides (should contain "MineRLTreechop-v0" directory).
EPOCHS = 3  # How many times we train over the dataset.
LEARNING_RATE = 0.0001  # Learning rate for the neural network.

TRAIN_MODEL_NAME = 'another_potato.pth'  # name to use when saving the trained agent.
TEST_MODEL_NAME = 'another_potato.pth'  # name to use when loading the trained agent.

TEST_EPISODES = 10  # number of episodes to test the agent for.
MAX_TEST_EPISODE_LEN = 18000  # 18k is the default for MineRLObtainDiamond.
TREECHOP_STEPS = 2000  # number of steps to run BC lumberjack for in evaluations.

class NatureCNN(nn.Module):
    """
    CNN from DQN nature paper:
        Mnih, Volodymyr, et al.
        "Human-level control through deep reinforcement learning."
        Nature 518.7540 (2015): 529-533.

    :param input_shape: A three-item tuple telling image dimensions in (C, H, W)
    :param output_dim: Dimensionality of the output vector
    """

    def __init__(self, input_shape, output_dim):
        super().__init__()
        n_input_channels = input_shape[0]
        self.cnn = nn.Sequential(
            nn.Conv2d(n_input_channels, 32, kernel_size=8, stride=4, padding=0),
            nn.ReLU(),
            nn.Conv2d(32, 64, kernel_size=4, stride=2, padding=0),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, stride=1, padding=0),
            nn.ReLU(),
            nn.Flatten(),
        )

        # Compute shape by doing one forward pass
        with th.no_grad():
            n_flatten = self.cnn(th.zeros(1, *input_shape)).shape[1]

        self.linear = nn.Sequential(
            nn.Linear(n_flatten, 512),
            nn.ReLU(),
            nn.Linear(512, output_dim)
        )

    def forward(self, observations: th.Tensor) -> th.Tensor:
        return self.linear(self.cnn(observations))

class ActionShaping(gym.ActionWrapper):
    """
    The default MineRL action space is the following dict:

    Dict(attack:Discrete(2),
         back:Discrete(2),
         camera:Box(low=-180.0, high=180.0, shape=(2,)),
         craft:Enum(crafting_table,none,planks,stick,torch),
         equip:Enum(air,iron_axe,iron_pickaxe,none,stone_axe,stone_pickaxe,wooden_axe,wooden_pickaxe),
         forward:Discrete(2),
         jump:Discrete(2),
         left:Discrete(2),
         nearbyCraft:Enum(furnace,iron_axe,iron_pickaxe,none,stone_axe,stone_pickaxe,wooden_axe,wooden_pickaxe),
         nearbySmelt:Enum(coal,iron_ingot,none),
         place:Enum(cobblestone,crafting_table,dirt,furnace,none,stone,torch),
         right:Discrete(2),
         sneak:Discrete(2),
         sprint:Discrete(2))

    It can be viewed as:
         - buttons, like attack, back, forward, sprint that are either pressed or not.
         - mouse, i.e. the continuous camera action in degrees. The two values are pitch (up/down), where up is
           negative, down is positive, and yaw (left/right), where left is negative, right is positive.
         - craft/equip/place actions for items specified above.
    So an example action could be sprint + forward + jump + attack + turn camera, all in one action.

    This wrapper makes the action space much smaller by selecting a few common actions and making the camera actions
    discrete. You can change these actions by changing self._actions below. That should just work with the RL agent,
    but would require some further tinkering below with the BC one.
    """
    def __init__(self, env, camera_angle=10, always_attack=False):
        super().__init__(env)

        self.camera_angle = camera_angle
        self.always_attack = always_attack
        self._actions = [
            [('attack', 1)],
            [('forward', 1)],
            # [('back', 1)],
            # [('left', 1)],
            # [('right', 1)],
            # [('jump', 1)],
            # [('forward', 1), ('attack', 1)],
            # [('craft', 'planks')],
            [('forward', 1), ('jump', 1)],
            [('camera', [-self.camera_angle, 0])],
            [('camera', [self.camera_angle, 0])],
            [('camera', [0, self.camera_angle])],
            [('camera', [0, -self.camera_angle])],
        ]

        self.actions = []
        for actions in self._actions:
            act = self.env.action_space.noop()
            for a, v in actions:
                act[a] = v
            if self.always_attack:
                act['attack'] = 1
            self.actions.append(act)

        self.action_space = gym.spaces.Discrete(len(self.actions))

    def action(self, action):
        return self.actions[action]

def dataset_action_batch_to_actions(dataset_actions, camera_margin=5):
    """
    Turn a batch of actions from dataset (`batch_iter`) to a numpy
    array that corresponds to batch of actions of ActionShaping wrapper (_actions).

    Camera margin sets the threshold what is considered "moving camera".

    Note: Hardcoded to work for actions in ActionShaping._actions, with "intuitive"
        ordering of actions.
        If you change ActionShaping._actions, remember to change this!

    Array elements are integers corresponding to actions, or "-1"
    for actions that did not have any corresponding discrete match.
    """
    # There are dummy dimensions of shape one
    camera_actions = dataset_actions["camera"].squeeze()
    attack_actions = dataset_actions["attack"].squeeze()
    forward_actions = dataset_actions["forward"].squeeze()
    jump_actions = dataset_actions["jump"].squeeze()
    batch_size = len(camera_actions)
    actions = np.zeros((batch_size,), dtype=np.int)

    for i in range(len(camera_actions)):
        # Moving camera is most important (horizontal first)
        if camera_actions[i][0] < -camera_margin:
            actions[i] = 3
        elif camera_actions[i][0] > camera_margin:
            actions[i] = 4
        elif camera_actions[i][1] > camera_margin:
            actions[i] = 5
        elif camera_actions[i][1] < -camera_margin:
            actions[i] = 6
        elif forward_actions[i] == 1:
            if jump_actions[i] == 1:
                actions[i] = 2
            else:
                actions[i] = 1
        elif attack_actions[i] == 1:
            actions[i] = 0
        else:
            # No reasonable mapping (would be no-op)
            actions[i] = -1
    return actions

def train():
    data = minerl.data.make("MineRLTreechop-v0",  data_dir=DATA_DIR, num_workers=4)

    # We know ActionShaping has seven discrete actions, so we create
    # a network to map images to seven values (logits), which represent
    # likelihoods of selecting those actions
    network = NatureCNN((3, 64, 64), 7).cuda()
    optimizer = th.optim.Adam(network.parameters(), lr=LEARNING_RATE)
    loss_function = nn.CrossEntropyLoss()

    iter_count = 0
    losses = []
    for dataset_obs, dataset_actions, _, _, _ in tqdm(data.batch_iter(num_epochs=EPOCHS, batch_size=32, seq_len=1)):
        # We only use pov observations (also remove dummy dimensions)
        obs = dataset_obs["pov"].squeeze().astype(np.float32)
        # Transpose observations to be channel-first (BCHW instead of BHWC)
        obs = obs.transpose(0, 3, 1, 2)
        # Normalize observations
        obs /= 255.0

        # Actions need bit more work
        actions = dataset_action_batch_to_actions(dataset_actions)

        # Remove samples that had no corresponding action
        mask = actions != -1
        obs = obs[mask]
        actions = actions[mask]

        # Obtain logits of each action
        logits = network(th.from_numpy(obs).float().cuda())

        # Minimize cross-entropy with target labels.
        # We could also compute the probability of demonstration actions and
        # maximize them.
        loss = loss_function(logits, th.from_numpy(actions).long().cuda())

        # Standard PyTorch update
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        iter_count += 1
        losses.append(loss.item())
        if (iter_count % 1000) == 0:
            mean_loss = sum(losses) / len(losses)
            tqdm.write("Iteration {}. Loss {:<10.3f}".format(iter_count, mean_loss))
            losses.clear()

    th.save(network.state_dict(), TRAIN_MODEL_NAME)
    del data

def str_to_act(env, actions):
    """
    Simplifies specifying actions for the scripted part of the agent.
    Some examples for a string with a single action:
        'craft:planks'
        'camera:[10,0]'
        'attack'
        'jump'
        ''
    There should be no spaces in single actions, as we use spaces to separate actions with multiple "buttons" pressed:
        'attack sprint forward'
        'forward camera:[0,10]'

    :param env: base MineRL environment.
    :param actions: string of actions.
    :return: dict action, compatible with the base MineRL environment.
    """
    act = env.action_space.noop()
    for action in actions.split():
        if ":" in action:
            k, v = action.split(':')
            if k == 'camera':
                act[k] = eval(v)
            else:
                act[k] = v
        else:
            act[action] = 1
    return act

def get_action_sequence():
    """
    Specify the action sequence for the scripted part of the agent.
    """
    # make planks, sticks, crafting table and wooden pickaxe:
    action_sequence = []
    action_sequence += [''] * 100
    action_sequence += ['craft:planks'] * 4
    action_sequence += ['craft:stick'] * 2
    action_sequence += ['craft:crafting_table']
    action_sequence += ['camera:[10,0]'] * 18
    action_sequence += ['attack'] * 20
    action_sequence += [''] * 10
    action_sequence += ['jump']
    action_sequence += [''] * 5
    action_sequence += ['place:crafting_table']
    action_sequence += [''] * 10

    # bug: looking straight down at a crafting table doesn't let you craft. So we look up a bit before crafting.
    action_sequence += ['camera:[-1,0]']
    action_sequence += ['nearbyCraft:wooden_pickaxe']
    action_sequence += ['camera:[1,0]']
    action_sequence += [''] * 10
    action_sequence += ['equip:wooden_pickaxe']
    action_sequence += [''] * 10

    # dig down:
    action_sequence += ['attack'] * 600
    action_sequence += [''] * 10

    return action_sequence

def test():
    network = NatureCNN((3, 64, 64), 7).cuda()
    network.load_state_dict(th.load(TEST_MODEL_NAME))

    env = gym.make('MineRLObtainDiamond-v0')

    # optional interactive mode, where you can connect to your agent and play together (see link for details):
    # https://minerl.io/docs/tutorials/minerl_tools.html#interactive-mode-minerl-interactor
    # env.make_interactive(port=6666, realtime=True)

    env = ActionShaping(env, always_attack=True)
    env1 = env.unwrapped

    num_actions = env.action_space.n
    action_list = np.arange(num_actions)

    action_sequence = get_action_sequence()

    for episode in range(TEST_EPISODES):
        obs = env.reset()
        done = False
        total_reward = 0
        steps = 0

        # BC part to get some logs:
        for i in range(TREECHOP_STEPS):
            # Process the action:
            #   - Add/remove batch dimensions
            #   - Transpose image (needs to be channels-last)
            #   - Normalize image
            obs = th.from_numpy(obs['pov'].transpose(2, 0, 1)[None].astype(np.float32) / 255).cuda()
            # Turn logits into probabilities
            probabilities = th.softmax(network(obs), dim=1)[0]
            # Into numpy
            probabilities = probabilities.detach().cpu().numpy()
            # Sample action according to the probabilities
            action = np.random.choice(action_list, p=probabilities)

            obs, reward, done, info = env.step(action)
            total_reward += reward
            steps += 1
            if done:
                break

        # scripted part to use the logs:
        if not done:
            for i, action in enumerate(action_sequence[:MAX_TEST_EPISODE_LEN - TREECHOP_STEPS]):
                obs, reward, done, _ = env1.step(str_to_act(env1, action))
                total_reward += reward
                steps += 1
                if done:
                    break

        print(f'Episode #{episode + 1} reward: {total_reward}\t\t episode length: {steps}')

    env.close()

def main():
    # train()
    test()

if __name__ == '__main__':
    main()


text:
"""
NCL_panel_15.py
===============
This script illustrates the following concepts:
   - Paneling three plots vertically
   - Making a color bar span over two axes
   - Selecting a different colormap to abide by best practices. See the `color examples <https://geocat-examples.readthedocs.io/en/latest/gallery/index.html#colors>`_ for more information.

See following URLs to see the reproduced NCL plot & script:
    - Original NCL script: http://www.ncl.ucar.edu/Applications/Scripts/panel_15.ncl
    - Original NCL plot: http://www.ncl.ucar.edu/Applications/Images/panel_15_lg.png
"""

##############################################################################
# Import packages:

import cartopy.crs as ccrs
from cartopy.mpl.gridliner import LongitudeFormatter, LatitudeFormatter
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
import numpy as np
import xarray as xr

import geocat.datafiles as gdf
import geocat.viz.util as gvutil

##############################################################################
# Read in data:

# Open a netCDF data file using xarray default engine and load the data into
# xarrays
ds = xr.open_dataset(gdf.get("netcdf_files/h_avg_Y0191_D000.00.nc"),
                     decode_times=False)

# Ensure longitudes range from 0 to 360 degrees
t = gvutil.xr_add_cyclic_longitudes(ds.T, "lon_t")

# Selecting the first time step and then the three levels of interest
t = t.isel(time=0)
t_1 = t.isel(z_t=0)
t_2 = t.isel(z_t=1)
t_6 = t.isel(z_t=5)

##############################################################################
# Plot:
fig = plt.figure(figsize=(8, 12))

grid = gridspec.GridSpec(nrows=3, ncols=1, figure=fig)

# Choose the map projection
proj = ccrs.PlateCarree()

# Add the subplots
ax1 = fig.add_subplot(grid[0], projection=proj)  # upper cell of grid
ax2 = fig.add_subplot(grid[1], projection=proj)  # middle cell of grid
ax3 = fig.add_subplot(grid[2], projection=proj)  # lower cell of grid

for (ax, title) in [(ax1, 'level 0'), (ax2, 'level 1'), (ax3, 'level 6')]:
    # Use geocat.viz.util convenience function to set axes tick values
    gvutil.set_axes_limits_and_ticks(ax=ax,
                                     xlim=(-180, 180),
                                     ylim=(-90, 90),
                                     xticks=np.linspace(-180, 180, 13),
                                     yticks=np.linspace(-90, 90, 7))

    # Use geocat.viz.util convenience function to make plots look like NCL
    # plots by using latitude, longitude tick labels
    gvutil.add_lat_lon_ticklabels(ax)

    # Remove the degree symbol from tick labels
    ax.yaxis.set_major_formatter(LatitudeFormatter(degree_symbol=''))
    ax.xaxis.set_major_formatter(LongitudeFormatter(degree_symbol=''))

    # Use geocat.viz.util convenience function to add minor and major ticks
    gvutil.add_major_minor_ticks(ax)

    # Draw coastlines
    ax.coastlines(linewidth=0.5)

    # Use geocat.viz.util convenience function to set titles
    gvutil.set_titles_and_labels(ax,
                                 lefttitle=t_1.long_name,
                                 righttitle=t_1.units,
                                 lefttitlefontsize=10,
                                 righttitlefontsize=10)
    # Add center title
    ax.set_title(title, loc='center', y=1.04, fontsize=10)

# Select an appropriate colormap
cmap = 'magma'

# Plot data
C = ax1.contourf(t_1['lon_t'],
                 t_1['lat_t'],
                 t_1.data,
                 levels=np.arange(0, 30, 2),
                 cmap=cmap,
                 extend='both')
ax2.contourf(t_2['lon_t'],
             t_2['lat_t'],
             t_2.data,
             levels=np.arange(0, 30, 2),
             cmap=cmap,
             extend='both')
C_2 = ax3.contourf(t_6['lon_t'],
                   t_6['lat_t'],
                   t_6.data,
                   levels=np.arange(0, 22, 2),
                   cmap=cmap,
                   extend='both')

# Add colorbars
# By specifying two axes for `ax` the colorbar will span both of them
plt.colorbar(C,
             ax=[ax1, ax2],
             ticks=range(0, 30, 2),
             extendrect=True,
             extendfrac='auto',
             shrink=0.85,
             aspect=13,
             drawedges=True)
plt.colorbar(C_2,
             ax=ax3,
             ticks=range(0, 22, 2),
             extendrect=True,
             extendfrac='auto',
             shrink=0.85,
             aspect=5.5,
             drawedges=True)

plt.show()

max_stars_repo_name: NCAR/GeoCAT-examples

max_issues_repo_name: netgodz/GeoCAT-examples

max_forks_repo_name: netgodz/GeoCAT-examples


text:
"""
Runs optimisation of hyperparameters of neural networks. This will be fleshed out to use Bayesian optimisation.

TODO:
    - Add learning rate to optimiser? Currently models have learning rate decay schedulers, but optim could find best
      starting position.
    - Add number of epochs

@author: Tyler Bikaun
"""

import numpy as np
import yaml
from ax import optimize
from datetime import datetime

import torch

from experimenter import Experimenter
from connections import load_config, Mongo

class Optimiser:
    def __init__(self, trials=3):
        self.exp = Experimenter()
        self.mongo_coll_conn = Mongo(collection_name='optimisation')
        self.trials = trials
        self.config = load_config()
        self.task_type = self.config['Utils']['task_type']
        self.data_name = self.config['Utils'][self.task_type]['data_name']

    def _opt_full_data_performance(self, objective_name="f1_macro", minimise=None):
        """ Optimisation routine for full data performance of task learner """
        start_time = datetime.now()
        best_parameters, best_values, _, _ = optimize(
                                                    parameters=[
                                                        {
                                                            "name": "epochs",
                                                            "type": "range",
                                                            "value_type": "int",
                                                            "bounds": [16, 128]
                                                        },
                                                        {
                                                            "name": "batch_size",
                                                            "type": "range",
                                                            "value_type": "int",
                                                            "bounds": [16, 64],
                                                        },
                                                        {
                                                            "name": "tl_rnn_type",
                                                            "type": "fixed",
                                                            "value_type": "str",
                                                            "value": "gru",
                                                        },
                                                        {
                                                            "name": "tl_embedding_dim",
                                                            "type": "range",
                                                            "value_type": "int",
                                                            "bounds": [128, 1024],
                                                        },
                                                        {
                                                            "name": "tl_hidden_dim",
                                                            "type": "range",
                                                            "value_type": "int",
                                                            "bounds": [128, 1024],
                                                        },
                                                        {
                                                            "name": "learning_rate",    # TODO: Will need to update to tl_ in the future
                                                            "type": "range",
                                                            "value_type": "float",
                                                            "bounds": [0.00001, 0.1]
                                                        },
                                                        ],
                                                        evaluation_function= self.exp._full_data_performance,
                                                        minimize=minimise,
                                                        objective_name=objective_name,
                                                        total_trials=self.trials
                                                )
        finish_time = datetime.now()
        run_time = (finish_time-start_time).total_seconds()/60

        # TODO: Will put into a decorated in the future...
        data = {"name": f"FDP-{self.task_type}-{self.data_name}",
                "info": {"start timestamp": start_time,
                         "finish timestamp": finish_time,
                         "run time": run_time},
                "config": self.config,
                "settings": {"trials": self.trials, "object name": objective_name, "minimise": minimise},
                "results": {"best parameters": best_parameters, "best value": best_values[0][objective_name]}}
        # # Post results to mongodb
        self.mongo_coll_conn.post(data)

    def _opt_svae(self, objective_name="train_loss", minimise=None):
        """ Optimisation routine for full data performance of SVAE """

        start_time = datetime.now()
        best_parameters, best_values, _, _ = optimize(
                                                    parameters=[
                                                    {
                                                        "name": "epochs",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [25,100],
                                                    },
                                                    {
                                                        "name": "batch_size",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [16, 64],
                                                    },
                                                    {
                                                        "name": "svae_embedding_dim",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [128, 1024],
                                                    },
                                                    {
                                                        "name": "svae_hidden_dim",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [128, 1024],
                                                    },
                                                    {
                                                        "name": "svae_rnn_type",
                                                        "type": "fixed",
                                                        "value_type": "str",
                                                        "value": "gru",
                                                    },
                                                    {
                                                        "name": "svae_num_layers",
                                                        "type": "fixed",
                                                        "value_type": "int",
                                                        "value": 1,
                                                    },
                                                    {
                                                        "name": "svae_bidirectional",
                                                        "type": "fixed",
                                                        "value_type": "bool",
                                                        "value": True,
                                                    },
                                                    {
                                                        "name": "svae_latent_size",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [16, 256],
                                                    },
                                                    {
                                                        "name": "svae_word_dropout",
                                                        "type": "range",
                                                        "value_type": "float",
                                                        "bounds": [0, 1],
                                                    },
                                                    {
                                                        "name": "svae_embedding_dropout",
                                                        "type": "range",
                                                        "value_type": "float",
                                                        "bounds": [0, 1],
                                                    },
                                                    {
                                                        "name": "svae_k",
                                                        "type": "range",
                                                        "value_type": "float",
                                                        "bounds": [0, 1],
                                                    },
                                                    {
                                                        "name": "svae_x0",
                                                        "type": "range",
                                                        "value_type": "int",
                                                        "bounds": [250, 5000],
                                                    },
                                                    ],
                                                    evaluation_function= self.exp._svae,
                                                    minimize=minimise,
                                                    objective_name=objective_name,
                                                    total_trials=self.trials
                                                )
        finish_time = datetime.now()
        run_time = (finish_time-start_time).total_seconds()/60

        # TODO: Will put into a decorated in the future...
        data = {"name": "SVAE",
                "info": {"start timestamp": start_time,
                            "finish timestamp": finish_time,
                            "run time": run_time},
                "settings": {"trials": self.trials, "object name": objective_name, "minimise": minimise},
                "results": {"best parameters": best_parameters, "best value": best_values[0][objective_name]}}
        # # Post results to mongodb
        self.mongo_coll_conn.post(data)

    def _opt_svaal(self, objective_name="val_f1_macro", minimise=None):
        """ Optimisation routine for SVAAL model

        Notes
        -----
        This model differs from SVAE and FDP as SVAAL is not necessarily trained on full data.
        """

        start_time = datetime.now()
        best_parameters, best_values, experiment, model = optimize(
                parameters=[
                {
                    "name": "epochs",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [10,100],
                },
                {
                    "name": "batch_size",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [16, 64],
                },
                # {
                #     "name": "tl_embedding_dim",
                #     "type": "fixed",
                #     "value_type": "int",
                #     "value": 64,
                # },
                # {
                #     "name": "tl_hidden_dim",
                #     "type": "fixed",
                #     "value_type": "int",
                #     "value": 64,
                # },
                # {
                #     "name": "tl_rnn_type",
                #     "type": "fixed",
                #     "value_type": "str",
                #     "value": "gru",
                # },
                # {
                #     "name": "tl_learning_rate",
                #     "type": "fixed",
                #     "value_type": "float",
                #     "value": 0.01,
                # },
                {
                    "name": "latent_size",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [16,128],
                },
                {
                    "name": "disc_fc_dim",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [64,256]
                },
                {
                    "name": "disc_learning_rate",
                    "type": "range",
                    "value_type": "float",
                    "bounds": [0.0001, 0.1]
                },
                # {
                #     "name": "svae_rnn_type",
                #     "type": "fixed",
                #     "value_type": "str",
                #     "value": "gru",
                # },
                {
                    "name": "svae_embedding_dim",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [128, 1024],
                },
                {
                    "name": "svae_hidden_dim",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [128, 1024],
                },
                # {
                #     "name": "svae_num_layers",
                #     "type": "fixed",
                #     "value_type": "int",
                #     "value": 1,
                # },
                # {
                #     "name": "svae_bidirectional",
                #     "type": "fixed",
                #     "value_type": "bool",
                #     "value": False,
                # },
                {
                    "name": "svae_word_dropout",
                    "type": "range",
                    "value_type": "float",
                    "bounds": [0, 1],
                },
                {
                    "name": "svae_embedding_dropout",
                    "type": "range",
                    "value_type": "float",
                    "bounds": [0, 1],
                },
                {
                    "name": "svae_k",
                    "type": "range",
                    "value_type": "float",
                    "bounds": [0, 1],
                },
                {
                    "name": "svae_x0",
                    "type": "range",
                    "value_type": "int",
                    "bounds": [250, 5000],
                },
                # {
                #     "name": "svae_adv_hyperparameter",
                #     "type": "fixed",
                #     "value_type": "int",
                #     "value": 1,
                # },
                {
                    "name": "svae_learning_rate",
                    "type": "range",
                    "value_type": "float",
                    "bounds": [0.0001, 0.1]
                },
                ],
                evaluation_function= self.exp.train_single_cycle,
                minimize=minimise,
                objective_name=objective_name,
                total_trials=self.trials
            )
        finish_time = datetime.now()
        run_time = (finish_time-start_time).total_seconds()/60

        data = {"name": "SEQ-SVAE-BBN-GRU",
                "info": {"start timestamp": start_time,
                            "finish timestamp": finish_time,
                            "run time": run_time},
                "settings": {"trials": self.trials, "object name": objective_name, "minimise": minimise},
                "results": {"best parameters": best_parameters, "best value": best_values[0][objective_name]}}

        print(data)
        # Post results to mongodb
        self.mongo_coll_conn.post(data)

if __name__ == '__main__':
    Optimiser(trials=10)._opt_full_data_performance(objective_name="test_f1_macro", minimise=False)
    # Optimiser(trials=10)._opt_svae(objective_name="train_loss", minimise=True)
    # Optimiser(trials=10)._opt_svaal(objective_name="test_f1_macro_diff", minimise=False)


text:
from multiprocessing import Queue, Process, Event, Value
from auspex.instruments import AlazarATS9870, AlazarChannel

import time
import unittest
import tempfile
import numpy as np
from QGL import *
import QGL.config
import auspex.config

# Dummy mode
auspex.config.auspex_dummy_mode = True

# Set temporary output directories
awg_dir = tempfile.TemporaryDirectory()
kern_dir = tempfile.TemporaryDirectory()
auspex.config.AWGDir = QGL.config.AWGDir = awg_dir.name
auspex.config.KernelDir = kern_dir.name

from auspex.qubit import *
import bbndb

class AlazarTestCase(unittest.TestCase):

    def basic(self, delay=1.0, averages=10, segments=20, record=1024):
        oc   = Queue()
        exit = Event()
        run  = Event()
        alz  = AlazarATS9870(resource_name="1")
        ch   = AlazarChannel()
        ch.phys_channel = 1
        alz.add_channel(ch)
        alz.gen_fake_data = True
        alz.connect()

        config_dict = {
            'acquireMode': 'digitizer',
            'bandwidth': "Full" ,
            'clockType': "int",
            'delay': 0.0,
            'enabled': True,
            'label': 'Alazar',
            'recordLength': record,
            'nbrSegments': segments,
            'nbrWaveforms': 1,
            'nbrRoundRobins': averages,
            'samplingRate': 500e6,
            'triggerCoupling': "DC",
            'triggerLevel': 100,
            'triggerSlope': "rising",
            'triggerSource': "Ext",
            'verticalCoupling': "DC",
            'verticalOffset': 0.0,
            'verticalScale': 1.0
        }
        alz._lib.setAll(config_dict)
        alz.record_length           = record
        alz.number_acquisitions     = alz._lib.numberAcquisitions
        alz.samples_per_acquisition = alz._lib.samplesPerAcquisition
        alz.number_segments         = segments
        alz.number_waveforms        = 1
        alz.number_averages         = averages
        alz.ch1_buffer              = alz._lib.ch1Buffer
        alz.ch2_buffer              = alz._lib.ch2Buffer
        # print("asdasd", alz.number_averages)

        class OC(object):
            def __init__(self):
                self.queue = Queue()
                self.points_taken = Value('i', 0)
            def push(self, data):
                # print(f"Got data {len(data)}")
                self.queue.put(data)
                self.points_taken.value += data.size
        oc = OC()
        ready = Value('i', 0)
        proc = Process(target=alz.receive_data, args=(ch, oc, exit, ready, run))
        proc.start()
        while ready.value < 1:
            time.sleep(delay)

        run.set()
        alz.wait_for_acquisition(run, timeout=5, ocs=[oc])

        exit.set()
        time.sleep(0.1)
        proc.join(3.0)
        if proc.is_alive():
            proc.terminate()
        alz.disconnect()

        self.assertTrue(oc.points_taken.value == averages*segments*record)

    def test_1sec(self):
        self.basic(delay=1.0)
    def test_100msec(self):
        self.basic(delay=0.1)
    def test_10msec(self):
        self.basic(delay=0.01)
    def test_1sec_100avg(self):
        self.basic(delay=1.0, averages=100)

    def test_qubit_experiment(self):
        cl = ChannelLibrary(":memory:")
        pl = PipelineManager()
        cl.clear()
        q1     = cl.new_qubit("q1")
        aps2_1 = cl.new_APS2("BBNAPS1", address="192.168.5.101")
        aps2_2 = cl.new_APS2("BBNAPS2", address="192.168.5.102")
        dig_1  = cl.new_Alazar("Alz1", address="1")
        h1     = cl.new_source("Holz1", "HolzworthHS9000", "HS9004A-009-1", power=-30, reference='10MHz')
        h2     = cl.new_source("Holz2", "HolzworthHS9000", "HS9004A-009-2", power=-30, reference='10MHz')
        cl.set_control(q1, aps2_1, generator=h1)
        cl.set_measure(q1, aps2_2, dig_1.ch("1"), generator=h2)
        cl.set_master(aps2_1, aps2_1.ch("m2"))
        pl.create_default_pipeline(buffers=True)
        pl["q1"]["Demodulate"].decimation_factor = 16
        pl["q1"]["Demodulate"]["Integrate"].box_car_stop = 1e-6
        exp = QubitExperiment(RabiAmp(cl["q1"], np.linspace(-1, 1, 51)), averages=100)
        exp.set_fake_data(dig_1, np.cos(np.linspace(-np.pi, np.pi, 51)))
        exp.run_sweeps()
        data, desc = exp.buffers[0].get_data()
        self.assertAlmostEqual(np.abs(data).sum(),459.1,places=0)

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

max_stars_repo_name: grahamrow/Auspex

max_issues_repo_name: grahamrow/Auspex

max_forks_repo_name: grahamrow/Auspex


text:
import numpy as np
import matplotlib.pyplot as plt

class Plotter():

	def __init__(self, e, g, a):
		self.e = e
		self.g = g
		self.a = a

	def plot_success(self):
	    result = np.load("result.npy").item()
	    stats = np.load("stats.npy").item()
	    colors = list("rgbcmyk")
	    print stats
	    x = result.keys()
	    y = result.values()
	    fig = plt.figure()
	    fig.suptitle('(Epsilon: {} - Gamma: {} - Alpha: {} \n Success Rate: %{})'.format(self.e, self.g, self.a, stats["winning"]), fontsize=14)
	    plt.scatter(x,y,color=colors.pop())
	    plt.xlabel("Trials", fontsize = 14)
	    plt.ylabel("Rewards", fontsize = 14)
	    plt.legend(result.keys())
	    fig.savefig('(Epsilon{}Gamma{}Alpha{}).png'.format(self.e, self.g, self.a), bbox_inches='tight')

	def plot_times(self):
	    times = np.load("times.npy").item()
	    colors = list("rgbcmyk")
	    x = times.keys()
	    y = times.values()
	    fig = plt.figure()
	    fig.suptitle('(Epsilon: {} - Gamma: {} - Alpha: {})'.format(self.e, self.g, self.a), fontsize=14)
	    plt.scatter(x,y,color=colors.pop())
	    plt.xlabel("Time", fontsize = 14)
	    plt.ylabel("Rewards", fontsize = 14)
	    plt.legend(times.keys())
	    fig.savefig('TIMES(Epsilon{}Gamma{}Alpha{}).png'.format(self.e, self.g, self.a), bbox_inches='tight')

	def plot_rewards(self):
	    result = np.load("rewards.npy").item()
	    colors = list("rgbcmyk")
	    x = result.keys()
	    y = result.values()
	    fig = plt.figure()
	    fig.suptitle('(Epsilon: {} - Gamma: {} - Alpha: {})'.format(self.e, self.g, self.a), fontsize=14)
	    plt.scatter(x,y,color=colors.pop())
	    plt.xlabel("Trials", fontsize = 14)
	    plt.ylabel("Positive Rewards %", fontsize = 14)
	    plt.legend(result.keys())
	    fig.savefig('REWARDS(Epsilon{}Gamma{}Alpha{}).png'.format(self.e, self.g, self.a), bbox_inches='tight')

max_stars_repo_name: ramankh/Smartcab

max_issues_repo_name: ramankh/Smartcab

max_forks_repo_name: ramankh/Smartcab


text:
from guacamol.benchmark_suites import goal_directed_benchmark_suite
from guacamol.distribution_matching_generator import DistributionMatchingGenerator
from guacamol.goal_directed_generator import GoalDirectedGenerator
import gzip, pickle
from guacamol.assess_distribution_learning import assess_distribution_learning
from guacamol.assess_goal_directed_generation import assess_goal_directed_generation
from abc import ABCMeta, abstractmethod
from typing import List, Optional
import numpy as np
import os, inspect
import glob
import random
from guacamol.scoring_function import ScoringFunction

root_location = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))

def full_cache_path(file):
    return os.path.realpath(root_location + '/train/pretrained/' + file)

version_name_list = ['v1','v2', 'trivial']

class GuacamolGoalWrapper:
    def __init__(self, benchmark):
        self.benchmark = benchmark

    def __call__(self, smiles_list):
        out = [self.benchmark.objective.score(s) for s in smiles_list]
        return out
    @property
    def name(self):
        return self.benchmark.name

def guacamol_goal_scoring_functions(version_name):
    assert version_name in version_name_list, "Version name must be in " + str(version_name_list)
    benchmarks = goal_directed_benchmark_suite(version_name=version_name)
    out = [GuacamolGoalWrapper(b) for b in benchmarks]
    return out

class DummyMoleculeGenerator(DistributionMatchingGenerator):
    def __init__(self, cache_files,
                 maximize_reward=False,
                 keep_last=None,
                 make_unique=False):
        '''

        :param cache_files: a filename or list of filenames, that must live in the pretrained directory
        :param maximize_reward: whether to return the molecules with the highest reward
        '''
        self.maximize_reward = maximize_reward
        self.cache_files = cache_files

        if type(cache_files) == str:
            cache_files = [cache_files]

        pre_data = []
        for cache_file in cache_files:
            if not os.path.isfile(cache_file):
                full_cache_file = full_cache_path(cache_file)
            else:
                full_cache_file = cache_file
            print(full_cache_file)
            with gzip.open(full_cache_file, 'rb') as f:
                this_data = pickle.load(f)
                pre_data += this_data

        if keep_last is not None:
            pre_data = pre_data[-int(keep_last):]

        if make_unique:
            tmp_dict = {}
            for s, r in pre_data:
                if s not in tmp_dict.keys():
                    tmp_dict[s] = r
                else:
                    tmp_dict = max(r, tmp_dict[s])

            pre_data = list(tmp_dict.items())

        if maximize_reward:
            pre_data = sorted(pre_data, key=lambda x: x[1], reverse=True)

        self.data = pre_data

    def generate(self, number_samples: int): # the benchmarks use 10K samples
        if self.maximize_reward:
            return [x[0] for x in self.data[:number_samples]]
        else:
            return [x[0] for x in random.sample(self.data, number_samples)]

class MyGoalDirectedGenerator(GoalDirectedGenerator):
    def __init__(self, version):
        self.version = version
        self.num_benchmarks = len(goal_directed_benchmark_suite(self.version))
        self.obj_num = 0

    def generate_optimized_molecules(self,
                                     scoring_function: ScoringFunction,
                                     number_molecules: int,
                                     starting_population: Optional[List[str]] = None) -> List[str]:
        """
        Given an objective function, generate molecules that score as high as possible.

        Args:
            scoring_function: scoring function
            number_molecules: number of molecules to generate
            starting_population: molecules to start the optimization from (optional)

        Returns:
            A list of SMILES strings for the generated molecules.
        """
        benchmarks = goal_directed_benchmark_suite(self.version)
        cache_files = ['guac_v2_{}do_0.5_lr4e-5_mark'.format(idx) for idx in range(20)]
        #TODO: put the directory for the cache file here
        gen = DummyMoleculeGenerator([cache_files[19]], maximize_reward=True)
        self.obj_num += 1
        if self.obj_num == self.num_benchmarks:
            self.obj_num == 0
        return gen.generate(number_molecules)


text:
"""Tests for the IMFV122 Parser class."""

from numpy.testing import assert_equal
from geomagio.imfv122 import IMFV122Parser
from obspy.core import UTCDateTime

def test_imfv122_parse_header__hour_of_day():
    """imfv122_test.test_imfv122_parse_header__minutes."""
    parser = IMFV122Parser()
    parser._parse_header(
        "KAK MAY0216 123 03 HDZF A KYO 05381402 000000 RRRRRRRRRRRRRRRR"
    )
    assert_equal(parser.channels, ["H", "D", "Z", "F"])
    metadata = parser.metadata
    assert_equal(metadata["declination_base"], 0)
    assert_equal(metadata["geodetic_latitude"], 53.8)
    assert_equal(metadata["geodetic_longitude"], 140.2)
    assert_equal(metadata["station"], "KAK")
    assert_equal(parser._delta, 60)
    assert_equal(parser._nexttime, UTCDateTime("2016-05-02T03:00:00Z"))

def test_imfv122_parse_header__minute_of_day():
    """imfv122_test.test_imfv122_parse_header__seconds."""
    parser = IMFV122Parser()
    parser._parse_header(
        "HER JAN0116 001 0123 HDZF R EDI 12440192 -14161 DRRRRRRRRRRRRRRR"
    )
    assert_equal(parser.channels, ["H", "D", "Z", "F"])
    metadata = parser.metadata
    assert_equal(metadata["declination_base"], -14161)
    assert_equal(metadata["geodetic_latitude"], 124.4)
    assert_equal(metadata["geodetic_longitude"], 19.2)
    assert_equal(metadata["station"], "HER")
    assert_equal(parser._delta, 60)
    assert_equal(parser._nexttime, UTCDateTime("2016-01-01T02:03:00Z"))

def test_imfv122_parse_data():
    """imfv122_test.test_imfv122_parse_data."""
    parser = IMFV122Parser()
    parser._parse_header(
        "HER JAN0116 001 0123 HDZF R EDI 12440192 -14161 DRRRRRRRRRRRRRRR"
    )
    parser._parse_data("1234 5678 9101 1121 3141 5161 7181 9202")
    import pprint

    pprint.pprint(parser._parsedata)
    assert_equal(parser._parsedata[0][0], UTCDateTime("2016-01-01T02:03:00Z"))
    assert_equal(parser._parsedata[1][0], "1234")
    assert_equal(parser._parsedata[2][0], "5678")
    assert_equal(parser._parsedata[3][0], "9101")
    assert_equal(parser._parsedata[4][0], "1121")
    assert_equal(parser._parsedata[0][1], UTCDateTime("2016-01-01T02:04:00Z"))
    assert_equal(parser._parsedata[1][1], "3141")
    assert_equal(parser._parsedata[2][1], "5161")
    assert_equal(parser._parsedata[3][1], "7181")
    assert_equal(parser._parsedata[4][1], "9202")

def test_imfv122_post_process():
    """imfv122_test.test_imfv122_post_process."""
    parser = IMFV122Parser()
    parser._parse_header(
        "HER JAN0116 001 0123 HDZF R EDI 12440192 -14161 DRRRRRRRRRRRRRRR"
    )
    parser._parse_data("1234 5678 9101 1121 3141 5161 7181 9202")
    parser._post_process()
    assert_equal(parser.times[0], UTCDateTime("2016-01-01T02:03:00Z"))
    assert_equal(parser.data["H"][0], 123.4)
    assert_equal(parser.data["D"][0], 56.78)
    assert_equal(parser.data["Z"][0], 910.1)
    assert_equal(parser.data["F"][0], 112.1)
    assert_equal(parser.times[1], UTCDateTime("2016-01-01T02:04:00Z"))
    assert_equal(parser.data["H"][1], 314.1)
    assert_equal(parser.data["D"][1], 51.61)
    assert_equal(parser.data["Z"][1], 718.1)
    assert_equal(parser.data["F"][1], 920.2)

max_stars_repo_name: usgs/geomag-algorithms

max_issues_repo_name: usgs/geomag-algorithms

max_forks_repo_name: alejandrodelcampillo/geomag-algorithms

reason: from numpy


text:
import numpy as np
from keras.models import Sequential
from keras.layers import Dense
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import OneHotEncoder

def execute(attributeLength, classLength, trainingFile, testingFile):
    X_train, y_train = readFromFile(trainingFile)

    X_test, y_test = readFromFile(testingFile)

    model = Sequential()
    model.add(Dense(12, input_dim=attributeLength, activation='relu'))
    model.add(Dense(8, activation='relu'))
    model.add(Dense(8, activation='relu'))
    model.add(Dense(8, activation='relu'))
    model.add(Dense(8, activation='relu'))
    model.add(Dense(classLength, activation='softmax'))

    model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])

    history = model.fit(X_train, y_train, epochs=150, batch_size=64)

    _, accuracy = model.evaluate(X_test, y_test)
    print('Accuracy: %.2f' % (accuracy * 100))

def readFromFile(files):
    X = []
    y = []

    matches = ['@RELATION', '@ATTRIBUTE', '@DATA']
    temp_X = []
    for file in files:
        temp_X = []
        y = []
        with open(file) as my_file:
            for line in my_file:
                if not any(x in line for x in matches):
                    result = [x.strip() for x in line.split(',')]
                    if len(result) > 1:
                        temp_X.append(result[:len(result) - 1])
                        y.append([result[len(result) - 1]])
        if len(X) == 0:
            X = temp_X

    if len(files) > 1:
        res_X = []
        for i in range(len(X)):
            a = X[i]
            b = temp_X[i]
            x = a + b
            res_X.append(x)

        X = res_X

    X = np.array(X, dtype=np.float128)
    y = np.array(y, dtype=np.float128)

    sc = StandardScaler()
    X = sc.fit_transform(X)

    ohe = OneHotEncoder()
    y = ohe.fit_transform(y).toarray()

    return [X, y]

def main():
    execute(5, 3, ['IrisGeometicFeatures_TrainingSet.txt'], ['IrisGeometicFeatures_TestingSet.txt'])
    # execute(9600, 3, ['IrisTextureFeatures_TrainingSet.txt'], ['IrisTextureFeatures_TestingSet.txt'])
    # execute(9605, 3, ['IrisGeometicFeatures_TrainingSet.txt', 'IrisTextureFeatures_TrainingSet.txt'], ['IrisTextureFeatures_TestingSet.txt', 'IrisGeometicFeatures_TestingSet.txt'])

if __name__ == '__main__':
    main()


text:
# Copyright 2019 The Cirq Developers
#
# 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
#
#     https://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.

import operator
from typing import Callable, Iterable, List, TYPE_CHECKING
import re

import networkx as nx

from cirq import circuits, ops
import cirq.contrib.acquaintance as cca
from cirq.contrib.routing.swap_network import SwapNetwork

if TYPE_CHECKING:
    import cirq

BINARY_OP_PREDICATE = Callable[[ops.Operation, ops.Operation], bool]

def get_time_slices(dag: circuits.CircuitDag) -> List[nx.Graph]:
    """Slices the DAG into logical graphs.

    Each time slice is a graph whose vertices are qubits and whose edges
    correspond to two-qubit gates. Single-qubit gates are ignored (and
    more-than-two-qubit gates are not supported).

    The edges of the first time slice correspond to the nodes of the DAG without
    predecessors. (Again, single-qubit gates are ignored.) The edges of the
    second slice correspond to the nodes of the DAG whose only predecessors are
    in the first time slice, and so on.
    """
    circuit = circuits.Circuit(op for op in dag.all_operations() if len(op.qubits) > 1)
    return [nx.Graph(op.qubits for op in moment.operations) for moment in circuit]

def ops_are_consistent_with_device_graph(
    ops: Iterable[ops.Operation], device_graph: nx.Graph
) -> bool:
    for op in ops:
        if not set(op.qubits).issubset(device_graph):
            return False
        if len(op.qubits) >= 2 and not device_graph.has_edge(*op.qubits):
            return False
    return True

def is_valid_routing(
    circuit: circuits.Circuit,
    swap_network: SwapNetwork,
    *,
    equals: BINARY_OP_PREDICATE = operator.eq,
    can_reorder: BINARY_OP_PREDICATE = circuits.circuit_dag._disjoint_qubits,
) -> bool:
    """Determines whether a swap network is consistent with a given circuit.

    Args:
        circuit: The circuit.
        swap_network: The swap network whose validity is to be checked.
        equals: The function to determine equality of operations. Defaults to
            `operator.eq`.
        can_reorder: A predicate that determines if two operations may be
            reordered.

    Raises:
        ValueError: If equals operator or can_reorder throws a ValueError.
    """
    circuit_dag = circuits.CircuitDag.from_circuit(circuit, can_reorder=can_reorder)
    logical_operations = swap_network.get_logical_operations()
    try:
        return cca.is_topologically_sorted(circuit_dag, logical_operations, equals)
    except ValueError as err:
        if re.match(r'Operation .* acts on unmapped qubit .*\.', str(err)):
            return False
        raise

def get_circuit_connectivity(circuit: 'cirq.Circuit') -> nx.Graph:
    """Return a graph of all 2q interactions in a circuit.

    Nodes are qubits and undirected edges correspond to any two-qubit
    operation.
    """
    g = nx.Graph()
    for op in circuit.all_operations():
        n_qubits = len(op.qubits)
        if n_qubits > 2:
            # coverage: ignore
            raise ValueError(
                f"Cannot build a graph out of a circuit that "
                f"contains {n_qubits}-qubit operations"
            )
        if n_qubits == 2:
            g.add_edge(*op.qubits)
    return g

max_issues_repo_name: bradyb/Cirq

max_forks_repo_name: bradyb/Cirq

reason: import networkx


text:
import re
import numpy as np

f = open("7.txt","r")

lines = f.readlines()
lines.sort()
instructions = [l.strip() for l in lines]

signals = {}

def printSignals():
    print("Filled Signals:")
    for s in signals:
        if signals[s]!=-1:
            print(s,end=': ')
            print(signals[s])
    print()

def runSimulation():
        #Populate signals dictionary with -1
    _abc = list("abcdefghijklmnopqrstuvwxyz")

    for i in _abc:
        signals[i] = -1 #For single letters
        #For double letters
        for j in _abc:
            signals[i+j] = -1
            continue
        continue

    while len(instructions):
        indicesToRemove = []

        for i in range(len(instructions)):
            #printSignals()

            cmd = instructions[i]

            _cmds = cmd.split(" ")
            _outputGate = _cmds[len(_cmds)-1]

            #Is a number assignment (ie. number -> "aa")
            if len(_cmds)==3 and re.search(r"(\d+)",_cmds[0])!=None:
                signals[_cmds[len(_cmds)-1]] = np.uint16(_cmds[0])
                indicesToRemove.append(i)
                continue

            #Is a gate assignment (ie. "aa" -> "bb")
            if len(_cmds)==3 and re.search(r"(\w)",_cmds[0])!=None:
                if signals[_cmds[0]]!=-1: #Must make sure that "aa" has a signal value
                    signals[_outputGate] = np.uint16(signals[_cmds[0]])
                    indicesToRemove.append(i)
                continue

            #NOT operation (ie. NOT "aa" -> "bb")
            if _cmds[0]=="NOT":
                if signals[_cmds[1]]!=-1:
                    signals[_outputGate] = ~np.uint16(signals[_cmds[1]])
                    indicesToRemove.append(i)
                continue

            #AND operation (ie. ("aa"|number) AND "aa" -> "bb")
            #   !NOTE: First might be a number or gate
            if _cmds[1]=="AND":
                if signals[_cmds[2]]!=-1:
                    if re.search(r"(\d+)",_cmds[0]): #First is actually a number
                        signals[_outputGate] = np.uint16(_cmds[0])&np.uint16(signals[_cmds[2]])
                        indicesToRemove.append(i)
                    elif signals[_cmds[0]]!=-1: #Normal gate
                        signals[_outputGate] = np.uint16(signals[_cmds[0]])&np.uint16(signals[_cmds[2]])
                        indicesToRemove.append(i)
                continue

            if _cmds[1]=="OR":
                if signals[_cmds[0]]!=-1 and signals[_cmds[2]]!=-1:
                    signals[_outputGate] = np.uint16(signals[_cmds[0]])|np.uint16(signals[_cmds[2]])
                    indicesToRemove.append(i)
                continue

            if _cmds[1]=="LSHIFT":
                if signals[_cmds[0]]!=-1:
                    signals[_outputGate] = np.uint16(signals[_cmds[0]]*pow(2,np.uint16(_cmds[2])))
                    indicesToRemove.append(i)
                continue

            if _cmds[1]=="RSHIFT":
                if signals[_cmds[0]]!=-1:
                    signals[_outputGate] = np.uint16(signals[_cmds[0]]/pow(2,np.uint16(_cmds[2])))
                    indicesToRemove.append(i)
                continue

            print("ERROR: Command "+cmd+" not recognised!")

        #Remove according to indices
        indicesToRemove.sort(reverse = True)
        for i in indicesToRemove:
            del instructions[i]

        #break #TODO: Remove

    #printSignals()

#Part A
runSimulation()
print("--Part A--")
printSignals()
print("Signal \'a\':",end=' ')
print(signals["a"])

#Part B
f = open("7b.txt","r")

lines = f.readlines()
lines.sort()
instructions = [l.strip() for l in lines]

runSimulation()
print("--Part B--")
printSignals()
print("Signal \'a\':",end=' ')
print(signals["a"])


text:
from sklearn.feature_extraction.image import _extract_patches
import numpy as np
import cv2
from datasets import download_and_prepare

def read_file(file, grayscale=False):
    """
    Loads and normalizes image.

    Parameters
        file (string): image file path
        grayscale (bool): True converts the image to grayscale

    Returns:
        (ndarray): image
    """
    image = cv2.imread(file)
    if grayscale:
        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    image = np.float32(image) / 255
    return image

files = download_and_prepare("utk-faces", "data")
Img = read_file(files[0])
print(Img.shape)
sample = _extract_patches(Img, patch_shape=(64,64,3), extraction_step=1)
print(sample.shape)


text:
import matplotlib
matplotlib.use('Agg')

import matplotlib.pyplot as plt
import numpy as np
import pickle
import sys
import corner
import matplotlib.gridspec as gridspec
from scipy import stats
from dart_board import sf_history
from dart_board.sf_history.sf_plotting import get_plot_polar
from dart_board.posterior import A_to_P, calculate_L_x

file_name = sys.argv[1]
in_file = "../data/" + file_name + "_chain.obj"

if len(sys.argv) == 2:
    delay = 200
else:
    delay = int(int(sys.argv[2]) / 100)

chains = pickle.load(open(in_file, "rb"))
if chains.ndim == 3:
    chains = chains[:,delay:,:]
elif chains.ndim == 4:
    chains = chains[0,:,delay:,:]
else:
    sys.exit()

n_chains, length, n_var = chains.shape
chains = chains.reshape((n_chains*length, n_var))
print(chains.shape)

chains[:,0] = np.exp(chains[:,0])
chains[:,1] = np.exp(chains[:,1])
chains[:,2] = np.exp(chains[:,2])
chains[:,-1] = np.exp(chains[:,-1])

one_sigma_low = (1.0-0.68)/2.0
one_sigma_high = 1.0 - one_sigma_low

median_idx = int(0.5*n_chains*length)
one_sigma_low_idx = int(one_sigma_low*n_chains*length)
one_sigma_high_idx = int(one_sigma_high*n_chains*length)

for i in range(n_var):
    tmp_x = np.sort(chains[:,i])

    median = tmp_x[median_idx]
    low = median - tmp_x[one_sigma_low_idx]
    high = tmp_x[one_sigma_high_idx] - median

    print("Median=", "{0:.2f}".format(median), "-", "{0:.2f}".format(low), "+", "{0:.2f}".format(high))

reason: import numpy,from scipy


text:
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# (C) British Crown Copyright 2017-2019 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 AdjustLandSeaPoints class"""

import unittest

import iris
import numpy as np
from iris.coords import DimCoord
from iris.cube import Cube
from iris.tests import IrisTest
from iris.util import squeeze

from improver.grids import ELLIPSOID
from improver.standardise import AdjustLandSeaPoints
from improver.utilities.spatial import OccurrenceWithinVicinity
from improver.utilities.warnings_handler import ManageWarnings

from ..nbhood.nbhood.test_BaseNeighbourhoodProcessing import set_up_cube

# The warning messages are internal to the iris.analysis module v2.2.0
IGNORED_MESSAGES = ["Using a non-tuple sequence for multidimensional indexing"]
WARNING_TYPES = [FutureWarning]

class Test__init__(IrisTest):
    """Tests for the initiation of the AdjustLandSeaPoints class."""

    def test_basic(self):
        """Test that instantiating the class results in an object with
        expected variables."""
        expected_members = {'nearest_cube': None,
                            'input_land': None,
                            'output_land': None,
                            'output_cube': None}
        result = AdjustLandSeaPoints()
        members = {attr: getattr(result, attr) for attr in dir(result)
                   if not attr.startswith("__")}
        non_methods = {key: val for key, val in members.items()
                       if not callable(val)}
        regridder = non_methods.pop('regridder')
        vicinity = members.pop('vicinity')
        self.assertDictEqual(non_methods, expected_members)
        self.assertTrue(isinstance(regridder, iris.analysis.Nearest))
        self.assertTrue(isinstance(vicinity, OccurrenceWithinVicinity))

    def test_extrap_arg(self):
        """Test with extrapolation_mode argument."""
        result = AdjustLandSeaPoints(extrapolation_mode="mask")
        regridder = getattr(result, 'regridder')
        self.assertTrue(isinstance(regridder, iris.analysis.Nearest))

    def test_extrap_arg_error(self):
        """Test with invalid extrapolation_mode argument."""
        msg = "Extrapolation mode 'not_valid' not supported"
        with self.assertRaisesRegex(ValueError, msg):
            AdjustLandSeaPoints(extrapolation_mode="not_valid")

    def test_vicinity_arg(self):
        """Test with vicinity_radius argument."""
        result = AdjustLandSeaPoints(vicinity_radius=30000.)
        vicinity = getattr(result, 'vicinity')
        self.assertTrue(isinstance(vicinity, OccurrenceWithinVicinity))
        self.assertEqual(vicinity.distance, 30000.)

    def test_vicinity_arg_error(self):
        """Test with invalid vicinity_radius argument.
        This is not possible as OccurrenceWithinVicinity does not check the
        input value."""
        pass

class Test__repr__(IrisTest):
    """Tests the __repr__ method of the AdjustLandSeaPoints class."""

    def test_basic(self):
        """Test that the expected string is returned."""
        expected = ("<AdjustLandSeaPoints: regridder: Nearest('nanmask'); "
                    "vicinity: <OccurrenceWithinVicinity: distance: 25000.0>>")
        result = repr(AdjustLandSeaPoints())
        self.assertEqual(result, expected)

class Test_correct_where_input_true(IrisTest):
    """Tests the correct_where_input_true method of the AdjustLandSeaPoints
    class."""

    def setUp(self):
        """Create a class-object containing the necessary cubes.
        All cubes are on the target grid. Here this is defined as a 3x3 grid.
        The grid contains ones everywhere except the centre point (a zero).
        The output_cube has a value of 0.5 at [0, 1].
        The move_sea_point cube has the zero value at [0, 1] instead of [1, 1],
        this allows it to be used in place of input_land to trigger the
        expected behaviour in the function.
        """
        self.plugin = AdjustLandSeaPoints(vicinity_radius=2200.)
        cube = squeeze(
            set_up_cube(num_grid_points=3,
                        zero_point_indices=((0, 0, 1, 1),)))
        self.plugin.input_land = cube.copy()
        self.plugin.output_land = cube.copy()
        self.plugin.nearest_cube = cube.copy()
        self.plugin.nearest_cube.data[0, 1] = 0.5
        self.plugin.output_cube = self.plugin.nearest_cube.copy()
        self.move_sea_point = squeeze(
            set_up_cube(num_grid_points=3,
                        zero_point_indices=((0, 0, 0, 1),)))

    def test_basic_sea(self):
        """Test that nothing changes with argument zero (sea)."""
        input_land = self.plugin.input_land.copy()
        output_land = self.plugin.output_land.copy()
        output_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(0)
        self.assertArrayEqual(input_land.data, self.plugin.input_land.data)
        self.assertArrayEqual(output_land.data, self.plugin.output_land.data)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_basic_land(self):
        """Test that nothing changes with argument one (land)."""
        input_land = self.plugin.input_land.copy()
        output_land = self.plugin.output_land.copy()
        output_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(1)
        self.assertArrayEqual(input_land.data, self.plugin.input_land.data)
        self.assertArrayEqual(output_land.data, self.plugin.output_land.data)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_work_sea(self):
        """Test for expected change with argument zero (sea)."""
        self.plugin.input_land = self.move_sea_point
        output_cube = self.plugin.output_cube.copy()
        # The output sea point should have been changed to the value from the
        # input sea point in the same grid.
        output_cube.data[1, 1] = 0.5
        self.plugin.correct_where_input_true(0)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_work_land(self):
        """Test for expected change with argument one (land)."""
        self.plugin.input_land = self.move_sea_point
        output_cube = self.plugin.output_cube.copy()
        # The input sea point should have been changed to the value from an
        # input land point in the same grid.
        output_cube.data[0, 1] = 1.0
        self.plugin.correct_where_input_true(1)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_work_sealand_eq_landsea(self):
        """Test result is independent of order of sea/land handling."""
        self.plugin.input_land = self.move_sea_point.copy()
        reset_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(0)
        self.plugin.correct_where_input_true(1)
        attempt_01 = self.plugin.output_cube.data.copy()

        self.plugin.output_cube = reset_cube
        self.plugin.correct_where_input_true(1)
        self.plugin.correct_where_input_true(0)
        attempt_10 = self.plugin.output_cube.data.copy()

        self.assertArrayEqual(attempt_01, attempt_10)

    def test_not_in_vicinity(self):
        """Test for no change if the matching point is too far away."""
        # We need larger arrays for this.
        # Define 5 x 5 arrays with output sea point at [1, 1] and input sea
        # point at [4, 4]. The alternative value of 0.5 at [4, 4] should not
        # be selected with a small vicinity_radius.
        self.plugin = AdjustLandSeaPoints(vicinity_radius=2200.)
        cube = squeeze(
            set_up_cube(num_grid_points=5,
                        zero_point_indices=((0, 0, 1, 1),)))
        self.plugin.output_land = cube.copy()
        self.plugin.nearest_cube = cube.copy()
        self.plugin.nearest_cube.data[4, 4] = 0.5
        self.plugin.output_cube = self.plugin.nearest_cube.copy()
        self.plugin.input_land = squeeze(
            set_up_cube(num_grid_points=5,
                        zero_point_indices=((0, 0, 4, 4),)))

        output_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(0)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_no_matching_points(self):
        """Test code runs and makes no changes if no sea points are present."""
        self.plugin.input_land.data = np.ones_like(
            self.plugin.input_land.data)
        self.plugin.output_land.data = np.ones_like(
            self.plugin.output_land.data)
        output_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(0)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

    def test_all_matching_points(self):
        """Test code runs and makes no changes if all land points are
        present."""
        self.plugin.input_land.data = np.ones_like(
            self.plugin.input_land.data)
        self.plugin.output_land.data = np.ones_like(
            self.plugin.output_land.data)
        output_cube = self.plugin.output_cube.copy()
        self.plugin.correct_where_input_true(1)
        self.assertArrayEqual(output_cube.data, self.plugin.output_cube.data)

class Test_process(IrisTest):
    """Tests the process method of the AdjustLandSeaPoints class."""

    def setUp(self):
        """Create a class-object containing the necessary cubes.
        All cubes are on the target grid. Here this is defined as a 5x5 grid.
        The cubes have values of one everywhere except:
        input_land: zeroes (sea points) at [0, 1], [4, 4]
        output_land: zeroes (sea points) at [0, 0], [1, 1]
        input_cube: 0. at [1, 1]; 0.5 at [0, 1]; 0.1 at [4, 4]
        These should trigger all the behavior we expect.
        """
        self.plugin = AdjustLandSeaPoints(vicinity_radius=2200.)

        self.output_land = squeeze(
            set_up_cube(num_grid_points=5,
                        zero_point_indices=((0, 0, 1, 1),
                                            (0, 0, 0, 0))))

        self.cube = squeeze(
            set_up_cube(num_grid_points=5,
                        zero_point_indices=((0, 0, 1, 1),)))
        self.cube.data[0, 1] = 0.5
        self.cube.data[4, 4] = 0.1

        self.input_land = squeeze(
            set_up_cube(num_grid_points=5,
                        zero_point_indices=((0, 0, 0, 1),
                                            (0, 0, 4, 4))))

        # Lat-lon coords for reprojection
        # These coords result in a 1:1 regridding with the above cubes.
        x_coord = DimCoord(np.linspace(-3.281, -3.153, 5),
                           standard_name='longitude',
                           units='degrees',
                           coord_system=ELLIPSOID)
        y_coord = DimCoord(np.linspace(54.896, 54.971, 5),
                           standard_name='latitude',
                           units='degrees',
                           coord_system=ELLIPSOID)
        self.input_land_ll = Cube(self.input_land.data,
                                  long_name='land_sea_mask',
                                  units='1',
                                  dim_coords_and_dims=[(y_coord, 0),
                                                       (x_coord, 1)])

    @ManageWarnings(ignored_messages=["Using a non-tuple sequence for "],
                    warning_types=[FutureWarning])
    def test_basic(self):
        """Test that the expected changes occur and meta-data are unchanged."""
        expected = self.cube.data.copy()
        # Output sea-point populated with data from input sea-point:
        expected[0, 0] = 0.5
        # Output sea-point populated with data from input sea-point:
        expected[1, 1] = 0.5
        # Output land-point populated with data from input land-point:
        expected[0, 1] = 1.
        # Output land-point populated with data from input sea-point due to
        # vicinity-constraint:
        expected[4, 4] = 1.
        result = self.plugin.process(self.cube,
                                     self.input_land,
                                     self.output_land)
        self.assertIsInstance(result, Cube)
        self.assertArrayEqual(result.data, expected)
        self.assertDictEqual(result.attributes, self.cube.attributes)
        self.assertEqual(result.name(), self.cube.name())

    @ManageWarnings(ignored_messages=IGNORED_MESSAGES,
                    warning_types=WARNING_TYPES)
    def test_with_regridding(self):
        """Test when input grid is on a different projection."""
        self.input_land = self.input_land_ll
        expected = self.cube.data.copy()
        # Output sea-point populated with data from input sea-point:
        expected[0, 0] = 0.5
        # Output sea-point populated with data from input sea-point:
        expected[1, 1] = 0.5
        # Output land-point populated with data from input land-point:
        expected[0, 1] = 1.
        # Output land-point populated with data from input sea-point due to
        # vicinity-constraint:
        expected[4, 4] = 1.
        result = self.plugin.process(self.cube,
                                     self.input_land,
                                     self.output_land)
        self.assertIsInstance(result, Cube)
        self.assertArrayEqual(result.data, expected)
        self.assertDictEqual(result.attributes, self.cube.attributes)
        self.assertEqual(result.name(), self.cube.name())

    @ManageWarnings(ignored_messages=IGNORED_MESSAGES,
                    warning_types=WARNING_TYPES)
    def test_multi_realization(self):
        """Test that the expected changes occur and meta-data are unchanged
        when handling a multi-realization cube."""
        cube = self.cube.copy()
        cube.coord('realization').points = [1]
        cubes = iris.cube.CubeList([self.cube, cube])
        cube = cubes.merge_cube()

        expected = cube.data.copy()

        # Output sea-point populated with data from input sea-point:
        expected[:, 0, 0] = 0.5
        # Output sea-point populated with data from input sea-point:
        expected[:, 1, 1] = 0.5
        # Output land-point populated with data from input land-point:
        expected[:, 0, 1] = 1.
        # Output land-point populated with data from input sea-point due to
        # vicinity-constraint:
        expected[:, 4, 4] = 1.
        result = self.plugin.process(cube,
                                     self.input_land,
                                     self.output_land)
        self.assertIsInstance(result, Cube)
        self.assertArrayEqual(result.data, expected)
        self.assertDictEqual(result.attributes, self.cube.attributes)
        self.assertEqual(result.name(), self.cube.name())

    def test_raises_gridding_error(self):
        """Test error raised when cube and output grids don't match."""
        self.cube = self.input_land_ll
        msg = "X and Y coordinates do not match for cubes"
        with self.assertRaisesRegex(ValueError, msg):
            self.plugin.process(self.cube,
                                self.input_land,
                                self.output_land)

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

max_stars_repo_name: LaurenceBeard/improver

max_issues_repo_name: LaurenceBeard/improver

max_forks_repo_name: LaurenceBeard/improver