text:
import numpy as np
import random as rand
import pdb

class POMDPSolver:
    def __init__(self,gw,
                 state_shape,
                 start_positions,
                 num_actions,
                 num_obs,
                 depth=2,
                 gamma = 1
                 ):
        self.shape = state_shape
        self.start_positions = start_positions
        self._num_states = self.shape[0] * self.shape[1] * self.shape[2]
        self._num_actions = num_actions
        self._num_obs = num_obs
        self.eps = .001
        self.gw = gw
        self.d = depth
        self.gamma = gamma

    def init_belief(self):
        b0 = np.zeros(self.shape)
        for start_candidate in self.start_positions:
            b0[start_candidate]=1.0/len(self.start_positions)
        return b0

    def update_belief(self,o):
        #update belief for the entire state based on the observation.
        print('update')
        z, y, x = self.gw.grid_indices_to_coordinates()
        new_belief = np.zeros_like(self.b)
        for (z0,(y0,x0)) in zip(z,zip(y,x)): #for all current state
            key = (z0,y0,x0)
            neighbors = self.gw.neighbor_dict[str(key)]
            tb= 0
            # pdb.set_trace()
            for neighbor in neighbors: #sbar
                zbar, ybar, xbar = neighbor
                for action in range(self._num_actions):
                    tb+=self.gw._T[z0,y0,x0,action,zbar,ybar,xbar] * self.b[z0,y0,x0]
                tb*=self.gw._O[o,z0,y0,x0]
            new_belief[z0,y0,x0] = tb
        new_belief = np.divide(new_belief,np.sum(new_belief))
        return new_belief

    def get_reward(b,a):
        print('fill me in')
        return 0

    #how to init alpha vectors/
    def qmdp(self):
        #eqn 6.26
        new_alpha = 0
        return new_alpha

    def getU(self,b):
        print('fill me in')
        #eqn 6.24
        return 1

    def obs_prob(self,b,a):
        #∑_s P(o,|s,a)b(s)
        #return P(o|b,a)
        return 1

    def selectAction(self,b,d):
        if d == 0:
            return (None, self.getU(b))
        best_action, best_utility = (None, -flaot('inf'))
        for a in range(self._num_actions)):
            current_utility = self.get_reward(bp,a)
            for o in range(self._num_obs):
                bp= update_beleif(o)
                new_a, new_u = selectAction(bp,d-1)
                current_utility = current_utiltiy + self.gamma * self.obs_prob(b,a) * new_u

            if current_utility > best_utility:
                best_action = a
                best_utility = current_utility

            return best_action, best_utility
            #how am i updating my belief?

    def learn_fwsearch(self,U):
        #each individual experience
        b = self.init_belief()
        true_state = self.gw.start
        while True:
            obs_idx = self.gw.getObservation(true_state)
            # obs = self.gw.landmark_idx_lookup[obs_idx]
            self.update_belief(obs_idx)
            pdb.set_trace()
            action, utility = self.selectAction(b,self.d) #twostep lookahead
            next_state, reward, done =generate_experience_with_grid(true_state,action)
            if done:
                print('do more stuff')
                break

        pass

    # def find_alpha_vector(self):
        # pass
    def solve(self,max_iters=100):
        iter = 0
        U = 0
        while iter < max_iters:
            print('iter %d'%iter)
            self.learn_fwsearch(U)
            iter +=1

        pdb.set_trace()

max_stars_repo_name: jennaon/grid-world-rl

max_issues_repo_name: jennaon/grid-world-rl

max_forks_repo_name: jennaon/grid-world-rl


text:
from argparse import ArgumentParser
import math

from convokit import Corpus, download
import numpy as np
from scipy.stats import mode
from sklearn.metrics import precision_recall_fscore_support
import torch
from torch import nn
from torch.cuda.amp import GradScaler, autocast
from torch.optim import Optimizer
from torch.utils.data import DataLoader
from transformers import AdamW, AutoModelForTokenClassification, AutoTokenizer, get_linear_schedule_with_warmup

from data import CoarseDiscourseDataset, ConversationPathBatchSampler, add_title_to_root, conversation_path_collate_fn
from model import ConversationClassificationHRNN, UtteranceClassificationHRNN

WARMUP_RATIO = 0.1
CLIPPING_GRADIENT_NORM = 1.0
DISPLAY_STEPS = 100
CORPUS = 'reddit-coarse-discourse-corpus'

parser = ArgumentParser(description='Discourse Evaluation')
parser.add_argument('-m', '--model-name', type=str, default='google/bigbird-roberta-base',
                    help='name of pretrained model to use')
parser.add_argument('-g', '--gpu', type=int, default=None,
                    help='index of gpu to use)')
parser.add_argument('-l', '--learning-rate', type=float, default=2e-5,
                    help='base learning rate used')
parser.add_argument('-b', '--batch-size', type=int, default=16,
                    help='training data batch size')
parser.add_argument('-e', '--epochs', type=int, default=3,
                    help='number of epochs to run')
parser.add_argument('-t', '--train-split', type=float, default=.8,
                    help='amount of data used for training')
parser.add_argument('--max-conversation-len', type=int, default=11,
                    help='maximum conversation length')
parser.add_argument('--utterance-max', type=int, default=1024,
                    help='maximum utterance length')
parser.add_argument('-p', '--pretrain-path', type=str, default=None,
                    help='path to pretrained model')

def main() -> None:

    args = parser.parse_args()

    if args.gpu is None:
        device = torch.device('cpu')
    else:
        device = torch.device('cuda:{}'.format(args.gpu))

    corpus = Corpus(filename=download(CORPUS))
    add_title_to_root(corpus)

    tokenizer = AutoTokenizer.from_pretrained(args.model_name)

    conversations = list(corpus.iter_conversations())
    train_ceil = math.ceil(len(conversations) * args.train_split)
    train_conversations = conversations[:train_ceil]
    val_conversations = conversations[train_ceil:]

    train_dataset = CoarseDiscourseDataset(corpus, train_conversations, tokenizer, max_len=args.max_conversation_len, max_tokenization_len=args.utterance_max)
    val_dataset = CoarseDiscourseDataset(corpus, val_conversations, tokenizer, max_len=args.max_conversation_len, max_tokenization_len=args.utterance_max)
    val_dataset.label_encoder = train_dataset.label_encoder
    train_sampler = ConversationPathBatchSampler(args.batch_size, 1, train_dataset.get_indices_by_len())
    val_sampler = ConversationPathBatchSampler(args.batch_size * 4, 1, val_dataset.get_indices_by_len())
    train_loader = DataLoader(train_dataset, batch_sampler=train_sampler, collate_fn=conversation_path_collate_fn, pin_memory=True)
    val_loader = DataLoader(val_dataset, batch_sampler=val_sampler, collate_fn=conversation_path_collate_fn, pin_memory=True)

    num_training_steps = args.epochs * len(train_dataset)

    model = AutoModelForTokenClassification.from_pretrained(args.model_name, num_labels=len(train_dataset.label_encoder.classes_))
    model.to(device)

    if args.pretrain_path is not None:
        checkpoint = torch.load(args.pretrain_path, map_location=device)
        model.bert.load_state_dict(checkpoint['state_dict'])

    optimizer = AdamW(model.parameters(), args.learning_rate)
    scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=WARMUP_RATIO * num_training_steps, num_training_steps=num_training_steps)
    scaler = GradScaler()

    for epoch in range(args.epochs):
        print('Epoch {}'.format(epoch))
        train(train_loader, model, optimizer, scheduler, scaler, device, tokenizer.sep_token_id)
        validate(val_loader, model, device, tokenizer.sep_token_id)

def train(loader: DataLoader, model: nn.Module, optimizer: Optimizer, scheduler: object, scaler: GradScaler, device: torch.device, sep_token_id: int):
    non_blocking = device.type != 'cpu'
    loss_fct = nn.CrossEntropyLoss()
    losses = []
    samples_seen = 0
    for i, (path, attention_mask, targets, ids) in enumerate(loader):
        path = path.squeeze(1).to(device, non_blocking=non_blocking)
        attention_mask = path.squeeze(1).to(device, non_blocking=non_blocking)
        targets = targets.to(device, non_blocking=non_blocking)

        with autocast():
            output = model(input_ids=path, attention_mask=attention_mask)
            logits = output.logits[path == sep_token_id]
            loss = loss_fct(logits, targets.reshape(-1))

        losses.append(loss.item())

        optimizer.zero_grad()
        scaler.scale(loss).backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), CLIPPING_GRADIENT_NORM)
        scaler.step(optimizer)
        scaler.update()
        scheduler.step()

        samples_seen += targets.shape[0]
        if (i + 1) % DISPLAY_STEPS == 0 or samples_seen == len(loader.dataset):
            print('Training [{}/{}] Loss: {}'.format(samples_seen, len(loader.dataset), np.mean(losses[-DISPLAY_STEPS:])))

def validate(loader: DataLoader, model: nn.Module, device: torch.device, sep_token_id: int):
    non_blocking = device.type != 'cpu'
    loss_fct = nn.CrossEntropyLoss()
    with torch.no_grad():
        losses = []
        prediction_votes = {}
        for i, (path, attention_mask, targets, ids) in enumerate(loader):
            path = path.squeeze(1).to(device, non_blocking=non_blocking)
            attention_mask = path.squeeze(1).to(device, non_blocking=non_blocking)
            targets = targets.to(device, non_blocking=non_blocking)

            with autocast():
                output = model(input_ids=path, attention_mask=attention_mask)
                logits = output.logits[path == sep_token_id]
                loss = loss_fct(logits, targets.reshape(-1))
                predictions = logits.argmax(1)

            losses.append(loss.item())

            # baseline
            # predictions = []
            # for id_batch in ids:
            #     predictions.append([])
            #     for id_ in id_batch:
            #         if id_ in loader.dataset.corpus.conversations or '?' in loader.dataset.corpus.get_utterance(id_).text:
            #             prediction = 'question'
            #         else:
            #             prediction = 'answer'
            #         predictions[-1].append(prediction)
            #     predictions[-1] = loader.dataset.label_encoder.transform(predictions[-1])
            #

            for utterance_id, prediction in zip([id for id_batch in ids for id in id_batch], predictions.to('cpu').numpy().tolist()):
                if utterance_id in prediction_votes:
                    prediction_votes[utterance_id].append(prediction)
                else:
                    prediction_votes[utterance_id] = [prediction]
        utterance_ids = list(prediction_votes.keys())
        predictions = np.array([mode(prediction_votes[utterance_id]).mode[0] for utterance_id in utterance_ids])
        labels = [loader.dataset.corpus.get_utterance(utterance_id).retrieve_meta('majority_type') for utterance_id in utterance_ids]
        targets = loader.dataset.label_encoder.transform(labels)
        accuracy = np.mean(targets == predictions)
        precision, recall, fscore, _ = precision_recall_fscore_support(targets, predictions, average='weighted')
        print('Validation Loss: {}, Acc: {}, Pre: {}, Rec: {}, F: {}'.format(np.mean(losses), accuracy, precision, recall, fscore))

if __name__ == '__main__':
    main()

ext: py

max_stars_repo_name: rlaboulaye/turn-of-phrase

max_issues_repo_name: rlaboulaye/turn-of-phrase

max_forks_repo_name: rlaboulaye/turn-of-phrase

reason: import numpy,from scipy


text:
from utils.tensors import *
import numpy as np
from scripts.latent_space import reverse_z
from threading import Thread
from os import makedirs, path

def forward(model, input_imgs, b=8, gpu=0, optimize_z=False, **kwargs):
    """
    Produces a full forward pass of the model (encoding + decoding).
    On the intermediate level, a z-space optimization can be done
    :param model:
    :param input_imgs: numpy array
    :param b:
    :param gpu:
    :param optimize_z:
    :param kwargs:
    :return:
    """
    gen = batch_gen(input_imgs, batch_size=b)
    output = []
    model.eval()
    for arr in gen:
        tensor = convert_numpy_to_tensor(arr, gpu)
        # reconstruct = model(tensor, only_decode=True)
        z = model.encoder(tensor)
        if optimize_z:
            z = reverse_z(model.decoder, tensor, cuda=gpu, z_size=model.z_size, initial_z=z, **kwargs)
        reconstruct = model.decoder(z, only_last=True)
        reconstruct_array = convert_tensor_to_numpy(reconstruct, squeeze=False)
        output.append(reconstruct_array)
    return np.concatenate(output, axis=0)

def decoding(model, z, b=8, gpu=0):
    """
    Decode a latent space variable using the decoder of the given model
    :param model:
    :param z:
    :param b:
    :param gpu:
    :return:
    """
    gen = batch_gen(z, batch_size=b, vector=True)
    output = []
    model.eval()
    for arr in gen:
        tens_out = model.decoder(convert_numpy_to_tensor(arr, gpu, vector=True), only_last=True)
        output.append(convert_tensor_to_numpy(tens_out))
    return np.concatenate(output, 0)

def encoding(model, arr, b=8, gpu=0, optimize_z=False, **kwargs):
    gen = batch_gen(arr, batch_size=b)
    z_size = model.z_size
    output = []
    model.eval()
    for arr in gen:
        goal_tens = convert_numpy_to_tensor(arr, gpu)
        tens_out = model.encoder(goal_tens)
        if optimize_z:
            tens_out = reverse_z(model.decoder, goal_tens, cuda=gpu,  z_size=z_size, **kwargs)
        out = convert_tensor_to_numpy(tens_out, squeeze=False)
        output.append(out)

    return np.concatenate(output, 0)

class MultiThreadingEncoder(Thread):
    def __init__(self, model, gpu, dataset, folder, overwrite=True, **options):

        super(MultiThreadingEncoder, self).__init__()
        self.model = model
        self.gpu = gpu
        self.dataset = dataset
        self.options = options
        self.folder = folder
        self.overwrite = overwrite
        if not path.exists(folder):
            makedirs(folder)

    def run(self):
        for sequence in self.dataset:
            name = str(sequence['name'][0])
            path_to_file = path.join(self.folder, name+'.npy')
            if not self.overwrite:
                if path.isfile(path_to_file):
                    continue
            arr = sequence['data'][0]
            z = encoding(model=self.model, arr=arr, gpu=self.gpu, **self.options)
            np.save(path_to_file, z)

reason: import numpy


text:
import numpy as np
from sklearn import metrics  # Importing metrics from SK-Learn

doc_string = """
Metrics module to evaluate models
"""

def __doc__():
    return doc_string

def regression_model_evaluation(y, y_predict, set_type='', features_count=None):
    """
    Utility/Helper method to calculate the evaluation parameters for a regression model
    """
    result = {}

    if set_type != '':
        set_type = '_' + set_type

    # Mean Absolute Error on train set.
    result['MAE'] = metrics.mean_absolute_error(y, y_predict)
    # Mean Squared Error on train set.
    result['MSE'] = metrics.mean_squared_error(y, y_predict)
    # Root Mean Squared Error on train set.
    result['RMSE'] = np.sqrt(result['MSE'])
    # R_squared on train set.
    result['R_squared'] = metrics.r2_score(y, y_predict)

    # Adj r2 = 1-(1-R2)*(n-1)/(n-p-1)
    if features_count:
        # Adjusted R_squared on train set.
        result['Adj_R_squared'] = 1 - (
                ((1 - result['R_squared']) * (len(y) - features_count)) / (len(y) - features_count - 1))
    # Returning with appending type to key and rounding value
    return {f'{k}' + set_type: round(v, 4) for k, v in result.items()}

max_stars_repo_name: raghavajoijode-ds/rj-ml-utils

max_issues_repo_name: raghavajoijode-ds/rj-ml-utils

max_forks_repo_name: raghavajoijode-ds/rj-ml-utils


text:
######
### This script only works with the version of networkx < 2.0
######
import numpy as np
import glob
import os
import argparse
import re
import scipy.io as sio
import networkx as nx

from networkx.utils import powerlaw_sequence

spath = r'F:\Research\ScaleVariant\exp_20190803\scalefree'
N = 200
#model ='conf'
model ='BA'

if model == 'BA':
  for a in range(1, 6):
    m = a
    for i in range(1, 21):
      filename = os.path.join(spath, r'{0}/nodes_{1}_{0}_m_{2}_index_{3}.txt'.format(model, N, m, i))
      G = nx.barabasi_albert_graph(N, m)
      nx.write_edgelist(G, filename, data=False)

if model == 'conf':
  for m in range(1, 101):
    z = nx.utils.create_degree_sequence(N, powerlaw_sequence, exponent=3.0)
    G = nx.configuration_model(z)
    filename = os.path.join(spath, r'{0}/nodes_{1}_{0}_index_{2}.txt'.format(model, N, m))

    # remove parallel edges and self-loops
    loops = G.selfloop_edges()
    G = nx.Graph(G)
    G.remove_edges_from(loops)

    nx.write_edgelist(G, filename, data=False)

max_stars_repo_name: OminiaVincit/scale-variant-topo

max_issues_repo_name: OminiaVincit/scale-variant-topo

max_forks_repo_name: OminiaVincit/scale-variant-topo

reason: import numpy,import scipy,import networkx,from networkx


text:
from __future__ import absolute_import
from collections import defaultdict

import numpy as np
import torch
from torch.utils.data.sampler import (
    Sampler, SequentialSampler, RandomSampler, SubsetRandomSampler,
    WeightedRandomSampler)

def _choose_from(start, end, excluded_range=None, size=1, replace=False):
    num = end - start + 1
    if excluded_range is None:
        return np.random.choice(num, size=size, replace=replace) + start
    ex_start, ex_end = excluded_range
    num_ex = ex_end - ex_start + 1
    num -= num_ex
    inds = np.random.choice(num, size=size, replace=replace) + start
    inds += (inds >= ex_start) * num_ex
    return inds

class RandomPairSampler(Sampler):
    def __init__(self, data_source, neg_pos_ratio=1):
        super(RandomPairSampler, self).__init__(data_source)
        self.data_source = data_source
        self.num_samples = len(data_source)
        self.neg_pos_ratio = neg_pos_ratio
        # Sort by pid
        indices = np.argsort(np.asarray(data_source)[:, 1])
        self.index_map = dict(zip(np.arange(self.num_samples), indices))
        # Get the range of indices for each pid
        self.index_range = defaultdict(lambda: [self.num_samples, -1])
        for i, j in enumerate(indices):
            _, pid, _ = data_source[j]
            self.index_range[pid][0] = min(self.index_range[pid][0], i)
            self.index_range[pid][1] = max(self.index_range[pid][1], i)

    def __iter__(self):
        indices = np.random.permutation(self.num_samples)
        for i in indices:
            # anchor sample
            anchor_index = self.index_map[i]
            _, pid, _ = self.data_source[anchor_index]
            # positive sample
            start, end = self.index_range[pid]
            pos_index = _choose_from(start, end, excluded_range=(i, i))[0]
            yield anchor_index, self.index_map[pos_index]
            # negative samples
            neg_indices = _choose_from(0, self.num_samples - 1,
                                       excluded_range=(start, end),
                                       size=self.neg_pos_ratio)
            for neg_index in neg_indices:
                yield anchor_index, self.index_map[neg_index]

    def __len__(self):
        return self.num_samples * (1 + self.neg_pos_ratio)


text:
# -*- coding: utf-8 -*-
# https://dsp.stackexchange.com/q/76463/50076 ################################
import numpy as np
from numpy.fft import fft, ifft
from scipy.io import wavfile
from ssqueezepy import ssq_cwt, Wavelet
from ssqueezepy.visuals import imshow, plot

#%%# Helper methods ##########################################################
def frequency_modulate(slc, fc=None, b=.3):
    N = len(slc)
    if fc is None:
        fc = N / 18  # arbitrary
    # track actual `b` for demodulation purposes
    b_effective = b

    t_min, t_max = start / fs, end / fs
    t = np.linspace(t_min, t_max, N, endpoint=False)
    assert np.allclose(fs, 1 / np.diff(t))

    x0 = slc[:N]
    # ensure it's [-.5, .5] so diff(phi) is b*[-pi, pi]
    x0max = np.abs(x0).max()
    x0 /= (2*x0max)
    b_effective /= (2*x0max)

    # generate phase
    phi0 = 2*np.pi * fc * t
    phi1 = 2*np.pi * b * np.cumsum(x0)
    phi = phi0 + phi1
    diffmax  = np.abs(np.diff(phi)).max()
    # `b` correction
    if diffmax > np.pi or np.allclose(phi, np.pi):
        diffmax0 = np.abs(np.diff(phi0)).max()
        diffmax1 = np.abs(np.diff(phi1)).max()
        # epsilon term for stable inversion / pi-unambiguity
        eps = 1e-7
        factor = ((np.pi - diffmax0 - eps) / diffmax1)
        phi1 *= factor
        b_effective *= factor
        phi = phi0 + phi1
    assert np.abs(np.diff(phi)).max() <= np.pi

    # modulate
    x = np.cos(phi)
    return x, t, phi0, phi1, b_effective

def analytic(x):
    N = len(x)
    xf = fft(x)

    xaf = np.zeros(N, dtype='complex128')
    xaf[:N//2 + 1] = 2 * xf[:N//2 + 1]
    xaf[0] /= 2
    xaf[N//2] /= 2

    xa = ifft(xaf)
    assert np.allclose(xa.real, x)
    return xa

#%%# Load data, select slice #################################################
fs, data = wavfile.read(r"C:\Desktop\recording.wav")
data = data.astype('float64')
data /= (2*np.abs(data).max())

start, end = 0, fs
slc = data[start:end]

#%%# Modulate & validate #####################################################
x, t, phi0, phi10, b_effective = frequency_modulate(slc)

# extreme time localization
wavelet = Wavelet(('gmw', {'gamma': 1, 'beta': 1}))
# synchrosqueezed CWT
Tx, Wx, ssq_freqs, *_ = ssq_cwt(x, wavelet)
# ints for better plotting
ssq_freqs = (ssq_freqs * fs).astype(int)

#%%# Visualize ##############################################################
def viz(t_min=None, t_max=None, f_min=None, f_max=None, show_original=False,
        show_modulated=False):
    freqs = ssq_freqs[::-1]
    a = int(t_min * fs) if t_min is not None else 0
    b = int(t_max * fs) if t_max is not None else None
    d = np.argmin(np.abs(freqs - f_min)) if f_min is not None else None
    c = np.argmin(np.abs(freqs - f_max)) if f_max is not None else 0

    imshow(Tx[c:d, a:b], xticks=t[a:b], yticks=freqs[c:d], **kw)

    if show_original:
        plot(t[a:b], slc[a:b], xlabel="time [sec]", title="original data",
             show=1)
    if show_modulated:
        plot(t[a:b], x[a:b],   xlabel="time [sec]", title="modulated data",
             show=1)

mx = np.abs(Tx).max() * .4
kw = dict(abs=1, norm=(0, mx), title="abs(SSQ_CWT)",
          xlabel="time [sec]", ylabel="frequency [Hz]")
viz()
viz(0, .2,  300, 8000, show_original=1)
viz(0, .02, 300, 8000, show_original=1, show_modulated=1)

#%%
# plot(x, show=1)
# plot(slc, show=1)
phie = np.arccos(x)[:2048]

phi = np.unwrap(np.angle(analytic(x)))[:2048]
phi_exact = (phi0 + phi10)[:2048]
phi[:10] = phi_exact[:10]
ae = np.abs(phi - phi_exact)
print(ae.mean(), ae.max(), ae.min())

phi1 = (phi - phi0[:2048])
x_inv_cs = phi1 / (2*np.pi * b_effective)
x_inv = np.diff(x_inv_cs)
plot(x_inv)

max_stars_repo_path: SignalProcessing/Q76463 - FM - Frequency Modulating a Signal/main.py

max_stars_repo_name: OverLordGoldDragon/StackExchangeAnswers

max_issues_repo_name: OverLordGoldDragon/StackExchangeAnswers

max_forks_repo_name: OverLordGoldDragon/StackExchangeAnswers

reason: import numpy,from numpy,from scipy


text:
import functools

import matplotlib.colors as colors
import matplotlib.pyplot as plt
import numpy as np
from matplotlib import cm
from mpl_toolkits.axes_grid1.inset_locator import inset_axes

from src.utils.camera import Camera

depth_image_cmap = 'gist_yarg'

def plotlive(func):
    """
    A decorator that updates the current figure
    instead of plotting into a new one.
    It requires that the figure is created before calling
    the plot function that is decorated.
    The figure should be passed as a parameter to the plot function.
    """
    plt.ion()

    @functools.wraps(func)
    def new_func(*args, **kwargs):
        # Clear current axis
        # plt.cla()

        # Clear all axes in the current figure.
        axes = plt.gcf().get_axes()
        for axis in axes:
            axis.cla()

        result = func(*args, **kwargs)

        plt.draw()
        plt.pause(0.01)

        return result

    return new_func

def plot_depth_image(image, fig_location=None, figsize=(3, 3)):
    """
    Plots the depth image in a new figure.
    """
    fig, ax = plt.subplots(1, figsize=figsize)
    _plot_depth_image(ax, image)
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    fig.subplots_adjust(left=0.02, right=0.98, bottom=0.02, top=0.98)
    # fig.tight_layout()
    save_show_fig(fig, fig_location, show_figure=True)

@plotlive
def plot_skeleton_with_jre_live(fig, axes, image, joints, jres, label=None,
                                norm_vec=None, mean_vec=None):
    """
    Plots the depth image, hand's skeleton, and a colorbar
    for the user, showing Joint Relation Errors.
    Replaces the contents of the currently active plot.
    """
    _plot_skeleton_with_jre(fig, axes, image, joints, jres, label, norm_vec, mean_vec)

def plot_skeleton_with_jre(image, joints, jres, label=None, show_fig=True, fig_location=None,
                           norm_vec=None, mean_vec=None):
    """
    Creates a new figure into which it plots the depth image,
    hand's skeleton, and a colorbar for the user, showing Joint Relation Errors.
    """
    fig, axes = plot_skeleton_with_jre_subplots()
    _plot_skeleton_with_jre(fig, axes, image, joints, jres, label, norm_vec, mean_vec)
    save_show_fig(fig, fig_location, show_fig)

def plot_skeleton_with_jre_subplots():
    """
    Creates a new figure for plotting the result
    of gesture recognition.
    The left axis contains the estimated hand pose,
    and the right axis displays colorbar.
    """
    fig, axes = plt.subplots(nrows=1, ncols=2, figsize=(6.5, 5),
                             gridspec_kw={"width_ratios": [5.0 / 6, 1.5 / 6]})
    bar_axis = axes[1]
    ax_inset = inset_axes(bar_axis, width="100%", height="100%", loc='upper center',
                          bbox_to_anchor=(0, 0.35, 0.15, 0.4), bbox_transform=bar_axis.transAxes)
    axes = [axes[0], axes[1], ax_inset]
    return fig, axes

def _plot_skeleton_with_jre(fig, axes, image, joints, jres, label=None,
                            norm_vec=None, mean_vec=None):
    """
    Plots the depth image, hand's skeleton, and a colorbar
    for the user, showing Joint Relation Errors.
    """
    hand_axis = axes[0]
    bar_axis = axes[2]
    _plot_depth_image(hand_axis, image)
    _plot_hand_skeleton(hand_axis, joints, wrist_index=0, scatter=False)
    _plot_joint_errors(fig, hand_axis, bar_axis, joints, joint_errors=jres)
    if norm_vec is not None and mean_vec is not None:
        _plot_hand_orientation(hand_axis, mean_vec, norm_vec)
    if label is not None:
        hand_axis.set_title(label)

    for ax in axes[:2]:
        ax.xaxis.set_visible(False)
        ax.yaxis.set_visible(False)
        ax.set_axis_off()

def _plot_hand_orientation(ax, mean, norm):
    """
    Plots an arrow (a vector) from the 'mean' position
    in the direction of the 'norm' vector.
    """
    ax.arrow(mean[0], mean[1], dx=norm[0] - mean[0], dy=norm[1] - mean[1],
             color='orange', head_length=5, shape='full', head_width=4, zorder=1000,
             width=1)

def cmap_subset(cmap, min, max):
    """ Create a subset of a cmap. """
    return colors.LinearSegmentedColormap.from_list(
        'trunc({n},{a:.2f},{b:.2f})'.format(n=cmap.name, a=min, b=max),
        cmap(np.linspace(min, max, 256)))

def _plot_joint_errors(fig, hand_axis, bar_axis, joints, joint_errors):
    """
    Plots a colorbar in the given axis, displaying the
    range of Joint Relation Errors.
    """
    plt.rcParams.update({"font.size": 18})
    min_err, max_err = joint_errors.min(), joint_errors.max()
    min_s, max_s = min_err * 4, max_err * 4  # 20., 200.
    scaled_errors = joint_errors / (max_err / (max_s - min_s)) + min_s

    cmap = cmap_subset(cm.get_cmap('Reds'), 0.4, 1.0)
    hand_axis.scatter(joints[..., 0], joints[..., 1],
                      c=joint_errors, cmap=cmap, s=scaled_errors, alpha=1, zorder=100)
    cbar = fig.colorbar(cm.ScalarMappable(cmap=cmap), cax=bar_axis, format='%.2f')
    colorbar_tick_labels = np.linspace(min_err, max_err, 5, dtype=float)
    colorbar_tick_labels = np.round(colorbar_tick_labels, 1)
    cbar.set_ticks(np.linspace(0, 1, 5))
    cbar.set_ticklabels(colorbar_tick_labels)
    cbar.ax.set_ylabel('Joint relation error [mm]', labelpad=15, fontsize=18)

def _plot_hand_skeleton(ax, joints, wrist_index=0, s=20, alpha=1, marker='o', scatter=True,
                        finger_colors='rmcgb', linewidth=2, linestyle='solid'):
    """
    Plots the hand's skeleton from joints in image coordinates.
    """
    joints = np.squeeze(joints)  # get rid of surplus dimensions
    if joints.ndim != 2:
        raise ValueError(F"joints.ndim should be 2, but is {joints.ndim}")
    fingers_bases = np.arange(wrist_index + 1, wrist_index + 20, 4)
    wrist_joint = joints[wrist_index]
    if scatter:
        ax.scatter(wrist_joint[..., 0], wrist_joint[..., 1], c=finger_colors[-1], marker=marker, s=s, alpha=alpha)
    for i, finger_base in enumerate(fingers_bases):
        finger_joints = joints[finger_base:finger_base + 4]
        if scatter:
            ax.scatter(finger_joints[..., 0], finger_joints[..., 1], c=finger_colors[i], marker=marker, s=s,
                       alpha=alpha)
        xs = np.concatenate([wrist_joint[0:1], finger_joints[:, 0]])
        ys = np.concatenate([wrist_joint[1:2], finger_joints[:, 1]])
        ax.plot(xs, ys, c=finger_colors[i], linewidth=linewidth, linestyle=linestyle)

def _plot_depth_image(ax, image):
    """
    Plot a depth image in an existing axis.
    """
    ax.imshow(image, cmap=depth_image_cmap)

@plotlive
def _plot_depth_image_live(ax, image):
    """
    Plot a depth image in an existing axis
    by replacing its previous content.
    """
    ax.imshow(image, cmap=depth_image_cmap)

def plot_image_with_skeleton(image, joints2d, show_fig=True, fig_location=None, figsize=(4, 3)):
    """
    Plots the depth image and the skeleton in a new figure.
    """
    fig, ax = plt.subplots(1, figsize=figsize)
    _plot_image_with_skeleton(fig, ax, image, joints2d)
    save_show_fig(fig, fig_location, show_fig)

@plotlive
def plot_image_with_skeleton_live(fig, ax, image, joints2d):
    """
    Plots the depth image and the skeleton in an existing figure
    by replacing its content.
    """
    _plot_image_with_skeleton(fig, ax, image, joints2d)

def _plot_image_with_skeleton(fig, ax, image, joints2d):
    """
    Plots the depth image and the skeleton in an existing figure.
    """
    _plot_depth_image(ax, image)
    _plot_hand_skeleton(ax, joints2d)
    ax.set_axis_off()
    fig.tight_layout()

def plot_joints_with_annotations(image, joints_pred, joints_true, show_fig=True, fig_location=None, figsize=(4, 3)):
    """
    Plots a depth image with predicted and ground truth skeletons.
    """
    fig, ax = plt.subplots(1, figsize=figsize)
    _plot_depth_image(ax, image)
    _plot_hand_skeleton(ax, joints_true, scatter=False, linewidth=2, linestyle=(0, (2, 2)))
    _plot_hand_skeleton(ax, joints_pred, scatter=False, linewidth=2)
    ax.set_axis_off()
    fig.tight_layout()
    save_show_fig(fig, fig_location, show_fig)

@plotlive
def plot_skeleton_with_label_live(fig, ax, image, joints2d, label):
    """
    Plots a depth image with skeleton and
    a label above the axis.
    """
    _plot_depth_image(ax, image)
    _plot_hand_skeleton(ax, joints2d)
    ax.set_title(label)
    ax.set_axis_off()
    fig.tight_layout()

def plot_proportion_below_threshold(proportions, show_figure=True, fig_location=None):
    """
    Plots the Proportion of joints below threshold metric.
    """
    if np.max(proportions) <= 1:
        proportions *= 100
    fig, ax = plt.subplots(1, 1)
    ax.plot(proportions)
    ax.set_ylim(0, 100)
    ax.set_ylabel('Proportion of frames within distnace (%)')
    ax.set_xlabel('Max joint error threshold (mm)')
    fig.tight_layout()
    save_show_fig(fig, fig_location, show_figure)

def plot_depth_image_histogram(image, show_fig=True, fig_location=None):
    """
    Plots a histogram of depth values in the image.
    """
    plt.rcParams.update({"font.size": 24})
    fig = plt.figure(figsize=(10, 7))
    min, max = np.min(image), np.max(image)
    plt.hist(image, bins=np.arange(min, max + 1, step=1), histtype='stepfilled')
    plt.xlabel('Depth [mm]', labelpad=20)
    plt.ylabel('Frequency', labelpad=20)
    plt.margins(x=0, y=0)
    plt.tick_params(axis='x', pad=15)
    plt.tick_params(axis='y', pad=15)
    plt.tight_layout()
    save_show_fig(fig, fig_location, show_fig)

def plot_image_comparison(original, filtered, filter_name):
    """
    Plots two depth images next to each other.
    """
    fig, (ax1, ax2) = plt.subplots(ncols=2, figsize=(8, 4), sharex=True,
                                   sharey=True)
    _plot_depth_image(ax1, original)
    ax1.set_title('original')
    ax1.axis('off')
    _plot_depth_image(ax2, filtered)
    ax2.set_title(filter_name)
    ax2.axis('off')
    fig.show()

def plot_bounding_cube(image, bcube, cam: Camera, fig_location=None, show_fig=True):
    """
    Projects a bounding cube in 3d coordinates
    onto the given depth image.
    """

    def get_four_points(P, dx, dy):
        P = np.array(P)
        P1 = P.copy()
        P2 = P.copy()
        P3 = P.copy()
        P4 = P.copy()
        P2[0] += dx
        P3[1] += dy
        P4[:2] += [dx, dy]
        Ps_xyz = np.array([P1, P2, P4, P3])
        Ps_uv = cam.world_to_pixel(Ps_xyz)[:, :2]
        return Ps_uv

    Au, Av, Az, Bu, Bv, Bz = bcube
    A = np.array([Au, Av, Az])
    B = np.array([Bu, Bv, Bz])
    A = cam.pixel_to_world(A[np.newaxis, ...])[0]
    B = cam.pixel_to_world(B[np.newaxis, ...])[0]
    dx = B[0] - A[0]
    dy = B[1] - A[1]
    As = get_four_points(A, dx, dy)
    Bs = get_four_points(B, -dx, -dy)
    rect1 = np.concatenate([As, [As[0]]])
    rect2 = np.concatenate([Bs, [Bs[0]]])

    fig, ax = plt.subplots(1, figsize=(3, 3))
    _plot_depth_image(ax, image)
    color = '#c61732'
    ax.plot(rect1[:, 0], rect1[:, 1], c=color)
    ax.plot(rect2[:, 0], rect2[:, 1], c=color)
    ax.plot([As[0, 0], Bs[2, 0]], [As[0, 1], Bs[2, 1]], c=color)
    ax.plot([As[1, 0], Bs[3, 0]], [As[1, 1], Bs[3, 1]], c=color)
    ax.plot([As[2, 0], Bs[0, 0]], [As[2, 1], Bs[0, 1]], c=color)
    ax.plot([As[3, 0], Bs[1, 0]], [As[3, 1], Bs[1, 1]], c=color)
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    fig.subplots_adjust(left=0.01, right=0.99, bottom=0.01, top=0.99)
    save_show_fig(fig, fig_location, show_fig)

def plot_scores(x, y, labels=None, fig_location=None, show_fig=True):
    """
    Plots evaluation metrics dependent on a threshold.
    """
    figsize = (8, 6)
    plt.style.use('seaborn-whitegrid')
    plt.rcParams.update({"font.size": 24})

    fig = plt.figure(figsize=figsize)
    for y_score in y:
        plt.plot(x, y_score)
    if labels is not None:
        if len(labels) != len(y):
            raise ValueError('y shape is not the same as labels shape')
        plt.legend(labels=labels)
    plt.xlabel('Threshold', labelpad=20)
    plt.ylabel('Score', labelpad=20)
    plt.margins(x=0, y=0)
    plt.tick_params(axis='x', pad=15)
    plt.tick_params(axis='y', pad=15)
    plt.tight_layout()
    save_show_fig(fig, fig_location, show_fig)

def precision_recall_curve(precision, recall, fig_location=None, show_fig=True):
    plt.style.use('seaborn-whitegrid')
    plt.rcParams.update({"font.size": 24})
    figsize = (8, 6)
    fig = plt.figure(figsize=figsize)
    plt.plot(recall, precision)
    plt.xlabel('Recall', labelpad=20)
    plt.ylabel('Precision', labelpad=20)
    plt.margins(x=0, y=0)
    plt.tick_params(axis='x', pad=15)
    plt.tick_params(axis='y', pad=15)
    plt.tight_layout()
    save_show_fig(fig, fig_location, show_fig)

def roc_curve(fnr, fpr, fig_location=None, show_fig=True):
    plt.style.use('seaborn-whitegrid')
    plt.rcParams.update({"font.size": 24})
    figsize = (8, 6)
    fig = plt.figure(figsize=figsize)
    plt.plot(fpr, fnr)
    plt.xlabel('False positive rate', labelpad=20)
    plt.ylabel('False negative rate', labelpad=20)
    plt.margins(x=0, y=0)
    plt.tick_params(axis='x', pad=15)
    plt.tick_params(axis='y', pad=15)
    plt.tight_layout()
    save_show_fig(fig, fig_location, show_fig)

def histogram(y, label, fig_location=None, show_fig=True, **kwargs):
    plt.style.use('seaborn-whitegrid')
    plt.rcParams.update({"font.size": 24})
    fig = plt.figure(figsize=(8, 6))
    plt.hist(y, bins=50, density=True, **kwargs)
    _set_pad_and_labels(label, 'Density')
    save_show_fig(fig, fig_location, show_fig)

def _set_pad_and_labels(xlabel, ylabel):
    plt.xlabel(xlabel, labelpad=20)
    plt.ylabel(ylabel, labelpad=20)
    plt.margins(x=0, y=0)
    plt.tick_params(axis='x', pad=15)
    plt.tick_params(axis='y', pad=15)
    plt.tight_layout()

def save_show_fig(fig, fig_location, show_figure):
    if fig_location:
        fig.savefig(fig_location)
    if show_figure:
        fig.show()

max_stars_repo_name: LadaOndris/IBT

max_issues_repo_name: LadaOndris/IBT

max_forks_repo_name: LadaOndris/IBT


text:
"""Boosting as described by David et al. (2007).

Versions
--------
Stored in ``algorithm_version`` attribute

 -1. Prior to storing version
 0. Normalize ``x`` after applying basis

Profiling
---------
ds = datasets._get_continuous()
y = ds['y']
x1 = ds['x1']
x2 = ds['x2']

%prun -s cumulative res = boosting(y, x1, 0, 1)

"""
from __future__ import annotations

from dataclasses import dataclass, field, fields
from functools import cached_property, reduce
import inspect
from itertools import chain, product, repeat
from math import ceil
from multiprocessing.sharedctypes import RawArray
from operator import mul
import os
import time
from threading import Event, Thread
from typing import Any, Callable, List, Literal, Union, Tuple, Sequence
import warnings

import numpy as np

from .._config import CONFIG, mpc
from .._data_obj import Case, Dataset, Dimension, SourceSpaceBase, NDVar, CategorialArg, NDVarArg, dataobj_repr
from .._exceptions import OldVersionError
from .._ndvar import _concatenate_values, convolve_jit, parallel_convolve, set_connectivity, set_parc
from .._utils import PickleableDataClass, user_activity
from .._utils.notebooks import tqdm
from ._boosting_opt import l1, l2, generate_options, update_error
from .shared import PredictorData, DeconvolutionData, Split, Splits, merge_segments
from ._fit_metrics import get_evaluators

# process messages
JOB_TERMINATE = -1

# error functions
ERROR_FUNC = {'l2': l2, 'l1': l1}
DELTA_ERROR_FUNC = {'l2': 2, 'l1': 1}

@dataclass(eq=False)
class BoostingResult(PickleableDataClass):
    """Result from boosting a temporal response function

    Attributes
    ----------
    h : NDVar | tuple of NDVar
        The temporal response function (the average of all TRFs from the
        different runs/partitions).
        Whether ``h`` is an :class:`NDVar` or a :class:`tuple` of :class:`NDVar`
        depends on whether the ``x`` parameter to :func:`boosting` was an
        :class:`NDVar` or a sequence of :class:`NDVar`.
    h_scaled : NDVar | tuple of NDVar
        ``h`` scaled such that it applies to the original input ``y`` and ``x``.
        If boosting was done with ``scale_data=False``, ``h_scaled`` is the same
        as ``h``.
    h_source : NDVar | tuple of NDVar
        If ``h`` was constructed using a basis, ``h_source`` represents the
        source of ``h`` before being convolved with the basis.
    h_time : UTS
        Time dimension of the kernel.
    r : float | NDVar
        Correlation between the measured response ``y`` and the predicted
        response ``h * x``. When using cross-validation (calling
        :func:`boosting` with ``test=True``), each partition of ``y`` is
        predicted using the ``h`` estimated from the corresponding training
        partitions. Otherwise, all of ``y`` is estimated using the average ``h``.
        For vector data, measured and predicted responses are normalized, and ``r``
        is computed as the average dot product over time.
        The type of ``r`` depends on the ``y`` parameter to :func:`boosting`:
        If ``y`` is one-dimensional, ``r`` is scalar, otherwise it is a :class:`NDVar`.
    r_rank : float | NDVar
        As ``r``, the Spearman rank correlation.
    t_run : float
        Time it took to run the boosting algorithm (in seconds).
    error : str
        The error evaluation method used.
    residual : float | NDVar
        The residual of the final result

        - ``error='l1'``: the sum of the absolute differences between ``y`` and
          ``h * x``.
        - ``error='l2'``: the sum of the squared differences between ``y`` and
          ``h * x``.

        For vector ``y``, the error is defined based on the distance in space
        for each data point.
    delta : scalar
        Kernel modification step used.
    mindelta : None | scalar
        Mindelta parameter used.
    n_samples : int
        Number of samples in the input data time axis.
    scale_data : bool
        Scale_data parameter used.
    y_mean : NDVar | scalar
        Mean that was subtracted from ``y``.
    y_scale : NDVar | scalar
        Scale by which ``y`` was divided.
    x_mean : NDVar | scalar | tuple
        Mean that was subtracted from ``x``.
    x_scale : NDVar | scalar | tuple
        Scale by which ``x`` was divided.
    splits : Splits
        Data splits used for cross-validation.
        Use :meth:`.splits.plot` to visualize the cross-validation scheme.
    partition_results : list of BoostingResuls
        If :func:`boosting` is called with ``partition_results=True``, this
        attribute contains the results for the individual test paritions.
    algorithm_version : int
        Version of the algorithm with which the model was estimated; ``-1`` for
        results from before this attribute was added.
    """
    # basic parameters
    y: str
    x: Union[str, Tuple[str]]
    tstart: Union[float, Tuple[float, ...]]
    tstop: Union[float, Tuple[float, ...]]
    scale_data: bool
    delta: float
    mindelta: float
    error: str
    selective_stopping: int
    # data properties
    y_mean: NDVar
    y_scale: NDVar
    x_mean: Union[NDVar, Tuple[NDVar, ...]]
    x_scale: Union[NDVar, Tuple[NDVar, ...]]
    # results
    _h: Union[NDVar, Tuple[NDVar, ...]]
    _isnan: np.ndarray
    t_run: float
    # advanced parameters
    basis: float
    basis_window: str
    splits: Splits = None
    # advanced data properties
    n_samples: int = None
    _y_info: dict = field(default_factory=dict)
    _y_dims: Tuple[Dimension, ...] = None
    # fit metrics
    i_test: int = None  # test partition for fit metrics
    residual: Union[float, NDVar] = None
    r: Union[float, NDVar] = None
    r_rank: Union[float, NDVar] = None
    r_l1: NDVar = None
    partition_results: List[BoostingResult] = None
    # store the version of the boosting algorithm with which model was fit
    version: int = 13  # file format (updates when re-saving)
    algorithm_version: int = -1  # does not change when re'saving
    # debug parameters
    y_pred: NDVar = None
    fit: Any = None  # scanpydoc can't handle undocumented 'Boosting'
    # legacy attributes
    prefit: str = None

    def __post_init__(self):
        if self.splits is None:
            self.partitions = None
        else:
            self.partitions = self.splits.n_partitions

    def __setstate__(self, state: dict):
        # backwards compatibility
        version = state.pop('version')
        if version < self.version:
            if version == 7:
                state['partitions'] = state.pop('n_partitions')
                state['partitions_arg'] = state.pop('n_partitions_arg')
            if version < 9:
                state['residual'] = state.pop('fit_error')
            if version < 11:
                for key in ['partitions_arg', 'h', 'isnan', 'y_info']:
                    state[f'_{key}'] = state.pop(key, None)
                state['r_rank'] = state.pop('spearmanr')
            if version < 12:
                # Vector residuals are averaged
                if state['r_rank'] is None:
                    state['residual'] *= state['n_samples']
            if version < 13:
                state['splits'] = Splits(None, state.pop('_partitions_arg'), state.pop('partitions'), state.pop('validate', 1), state.pop('test', 0), state.pop('model'))
        PickleableDataClass.__setstate__(self, state)

    def __repr__(self):
        items = []
        if isinstance(self.tstart, tuple):
            x = ' + '.join(f'{x} ({t1:g} - {t2:g})' for x, t1, t2 in zip(self.x, self.tstart, self.tstop))
        else:
            if self.x is None or isinstance(self.x, str):
                x = self.x
            else:
                x = ' + '.join(map(str, self.x))
            items.append(f'{self.tstart:g} - {self.tstop:g}')
        items.insert(0, f'boosting {self.y} ~ {x}')
        for name, param in inspect.signature(boosting).parameters.items():
            if param.default is inspect.Signature.empty or name == 'ds':
                continue
            elif name == 'debug':
                continue
            elif name == 'partition_results':
                value = bool(self.partition_results)
            elif name == 'partitions':
                value = None if self.splits is None else self.splits.partitions_arg
            elif name == 'model':
                if self.splits is None or self.splits.model is None:
                    continue
                value = dataobj_repr(self.splits.model)
            elif name == 'validate':
                value = None if self.splits is None else self.splits.n_validate
            elif name == 'test':
                value = None if self.splits is None else self.splits.n_test
            else:
                value = getattr(self, name)
            if value is not None and value != param.default:
                items.append(f'{name}={value}')
        return f"<{', '.join(items)}>"

    @cached_property
    def h(self):
        if not self.basis:
            return self._h
        elif isinstance(self._h, tuple):
            return tuple(h.smooth('time', self.basis, self.basis_window, 'full') for h in self._h)
        else:
            return self._h.smooth('time', self.basis, self.basis_window, 'full')

    @cached_property
    def h_scaled(self):
        if self.y_scale is None:
            return self.h
        elif isinstance(self.h, NDVar):
            out = self.h * (self.y_scale / self.x_scale)
            out.info = self._y_info.copy()
            return out
        else:
            out = []
            for h, sx in zip(self.h, self.x_scale):
                h = h * (self.y_scale / sx)
                h.info = self._y_info.copy()
                out.append(h)
            return tuple(out)

    @cached_property
    def h_source(self):
        return self._h

    @cached_property
    def h_time(self):
        if isinstance(self.h, NDVar):
            return self.h.time
        else:
            return self.h[0].time

    @cached_property
    def _variability(self):
        # variability in the data
        if self.y_scale is None:
            raise NotImplementedError("Not implemented for scale_data=False")
        elif self.n_samples is None:
            raise OldVersionError("This is an older result object which did not store some necessary information; refit the model to use this attribute")
        else:
            # Due to the normalization:
            return self.n_samples

    def cross_predict(
            self,
            x: Union[NDVarArg, Sequence[NDVarArg]] = None,
            ds: Dataset = None,
            name: str = None,
    ) -> NDVar:
        """Predict responses to ``x`` using complementary training data

        Parameters
        ----------
        x
            Predictors used in the original model fit, or a subset thereof.
            In order for cross-prediction to be accurate, ``x`` needs to match the ``x``
            used in the original fit exactly in cases and time.
        ds
            Dataset with predictors. If ``ds`` is specified, ``x`` can be omitted.
        name
            Name for the output :class:`NDVar`.

        See Also
        --------
        convolve : Simple prediction of linear model

        Notes
        -----
        This function does not adjust the mean across time of predicted
        responses; subtract the mean in order to compute explained variance.
        """
        # predictors
        x_ = self.x if x is None else x
        x_data = PredictorData(x_, ds)
        if not self.partition_results:
            raise ValueError("BoostingResult does not contain partition-specific models; fit with partition_results=True")
        # check predictors match h
        if x_data.x_name == self.x:
            x_use = None
        else:
            if isinstance(self.x, str):
                raise ValueError(f'{x=} for {self}')
            elif x_data.multiple_x:
                x_use = x_data.x_name
            else:
                x_use = [x_data.x_name]
            missing = set(x_use).difference(self.x)
            if missing:
                raise ValueError(f"{x=}: has variables not in {self}:\nx: {', '.join(missing)}")
        # prepare output array
        if self._y_dims is None:  # only possible in results from dev version
            y_dims = self.y_mean.dims
        else:
            y_dims = self._y_dims
        y_dimnames = [dim.name for dim in y_dims]
        n_y = sum(len(dim) for dim in y_dims) or 1
        y_pred = np.zeros((1, n_y, x_data.n_times_flat))
        # prepare x:  (n_x_only, n_shared, n_x_times)
        x_array = x_data.data[np.newaxis, :, :]
        # prepare h
        h_i_start = int(round(self.h_time.tmin / self.h_time.tstep))
        h_i_stop = h_i_start + len(self.h_time)
        # iterate through partitions
        for result in self.partition_results:
            # find segments
            for split in self.splits.splits:
                if split.i_test == result.i_test:
                    segments = split.test
                    break
            else:
                raise RuntimeError(f"Split missing for test segment {result.i_test}")
            # h to flat array: (n_h_only == in y, n_shared == in x, n_h_times)
            hs = [result.h_scaled] if isinstance(result.h_scaled, NDVar) else result.h_scaled
            if x_use:
                hs = {h.name: h for h in hs}
                hs = [hs[x] for x in x_use]
            h_array = []
            for h, (name, xdims, index) in zip(hs, x_data.x_meta):
                dimnames = [*y_dimnames, *[dim.name for dim in xdims], 'time']
                h_data = h.get_data(dimnames)
                h_data = h_data.reshape((n_y, -1, h_data.shape[-1]))
                h_array.append(h_data)
            h_array = np.concatenate(h_array, 1)
            # convolution
            for start, stop in segments:
                parallel_convolve(h_array, x_array[:, :, start:stop], y_pred[:, :, start:stop], h_i_start, h_i_stop)
        # package output
        dims = [*y_dims, x_data.time_dim]
        shape = [len(dim) for dim in dims]
        if x_data.case_to_segments:
            dims = (Case, *dims)
            shape.insert(-1, x_data.n_cases)
        y_pred = y_pred.reshape(shape)
        if x_data.case_to_segments:
            y_pred = np.rollaxis(y_pred, -2)
        if name is None:
            name = self.y
        return NDVar(y_pred, dims, name, self._y_info)

    def partition_result_data(self) -> Dataset:
        """Results from the different test partitions in a :class:`Dataset`"""
        h_is_list = isinstance(self._h, tuple)
        rows = []
        for res in self.partition_results:
            hs = res.h if h_is_list else [res.h]
            rows.append([res.i_test, res.r, res.proportion_explained, *hs])
        if self.x in (None, (None,)):
            xs = ['x']
        elif isinstance(self.x, str):
            xs = [self.x]
        else:
            xs = [f'x_{i}' if x is None else x for i, x in enumerate(self.x)]
        return Dataset.from_caselist(['i_test', 'r', 'det', *xs], rows)

    @cached_property
    def proportion_explained(self):
        return 1 - (self.residual / self._variability)

    def _apply_ndvar_transform(self, func: Callable):
        "Apply func to all NDVars in-place (only for source space transformation)"
        def sub_func(obj):
            if obj is None:
                return None
            elif isinstance(obj, tuple):
                return tuple(sub_func(obj_) for obj_ in obj)
            return func(obj)

        # NDVars
        for attr in ('_h', 'r', 'r_rank', 'residual', 'y_mean', 'y_scale'):
            setattr(self, attr, sub_func(getattr(self, attr)))

        # List of Dimension
        if self._y_dims is not None:
            self._y_dims = tuple([sub_func(dim) if isinstance(dim, SourceSpaceBase) else dim for dim in self._y_dims])

        if self.partition_results:
            for res in self.partition_results:
                res._apply_ndvar_transform(func)

    def _morph(self, to_subject: str):
        "Morph source space"
        from .._mne import morph_source_space

        def func(obj: NDVar):
            return morph_source_space(obj, to_subject)
        self._apply_ndvar_transform(func)

    def _set_connectivity(self, dim, connectivity):
        def func(obj: NDVar):
            return set_connectivity(obj, dim, connectivity)
        self._apply_ndvar_transform(func)

    def _set_parc(self, parc: str):
        """Change the parcellation of source-space result

        Notes
        -----
        No warning for missing sources!
        """
        def func(obj: NDVar):
            return set_parc(obj, parc, mask=True)
        self._apply_ndvar_transform(func)

    @classmethod
    def _eelbrain_concatenate(
            cls,
            results: Sequence[BoostingResult],
            dim: str = 'source',
    ):
        "Combine multiple complementary source-space BoostingResult objects"
        # result = results[0]
        out = {}
        for field in fields(cls):
            if field.name == 'version':
                continue
            elif field.name == '_isnan':
                out['_isnan'] = None
                continue
            values = [getattr(result, field.name) for result in results]
            if field.name == 't_run':
                out['t_run'] = None if any(v is None for v in values) else sum(values)
                continue
            if field.name == 'partition_results' and any(v is not None for v in values):
                if not all(v is not None for v in values):
                    raise ValueError(f'partition_results avaiable for some but not all part-results')
                new_value = [cls._eelbrain_concatenate(p_results) for p_results in zip(*values)]
            elif any(v is None for v in values):
                new_value = None
            else:
                new_value = _concatenate_values(values, dim, field.name)
            out[field.name] = new_value
        return cls(**out)

class SplitResult:
    __slots__ = ('split', 'h', 'h_failed')

    def __init__(self, split: Split, n_y: int, n_x: int, n_times_h: int):
        self.split = split
        self.h = np.empty((n_y, n_x, n_times_h), np.float64)
        self.h_failed = np.zeros(n_y, bool)

    def add_h(self, i_y: int, h: Union[np.ndarray, None]):
        if h is None:
            self.h_failed[i_y] = True
            self.h[i_y] = 0
        else:
            self.h[i_y] = h

    def h_with_nan(self):
        "Set failed TRFs to NaN"
        if np.any(self.h_failed):
            h = self.h.copy()
            h[self.h_failed] = np.nan
            return h
        return self.h

class Boosting:
    """Object-oriented API for boosting

    Examples
    --------
    Standard usage of the object-oriented API for a model with
    cross-validation, comparable to the :func:`boosting` function::

        data = DeconvolutionData(y, x, ds)
        data.apply_basis(0.05, 'hamming'')
        data.normalize('l1')
        data.initialize_cross_validation(5, test=1)
        model = Boosting(data)
        model.fit(0, 0.500, selective_stopping=1, error='l1')
        result = model.evaluate_fit()
    """
    # fit parameters
    tstart = None
    tstart_h = None
    tstop = None
    selective_stopping = None
    error = None
    delta = None
    mindelta = None
    # timing
    t_fit_start = None
    t_fit_done = None
    # fit result
    _i_start = None
    split_results = None
    n_skip = 0
    # eval result
    y_pred = None

    def __init__(
            self,
            data: DeconvolutionData,
    ):
        self.data = data

    def fit(
            self,
            tstart: Union[float, Sequence[float]],
            tstop: Union[float, Sequence[float]],
            selective_stopping: int = 1,
            error: str = 'l1',
            delta: float = 0.005,  # coordinate search step
            mindelta: float = None,  # narrow search by reducing delta until reaching mindelta
    ):
        self.data._check_data()
        assert error in ERROR_FUNC
        mindelta_ = delta if mindelta is None else mindelta
        self.selective_stopping = selective_stopping
        self.error = error
        self.delta = delta
        self.mindelta = mindelta
        n_y = len(self.data.y)
        n_x = len(self.data.x)
        # find TRF start/stop for each x
        if isinstance(tstart, (tuple, list, np.ndarray)):
            if len(tstart) != len(self.data.x_name):
                raise ValueError(f'tstart={tstart!r}: len(tstart) ({len(tstart)}) is different from len(x) ({len(self.data.x_name)}')
            elif len(tstart) != len(tstop):
                raise ValueError(f'tstop={tstop!r}: mismatched len(tstart) = {len(tstart)}, len(tstop) = {len(tstop)}')
            self.tstart = tuple(tstart)
            self.tstart_h = min(self.tstart)
            self.tstop = tuple(tstop)
            n_xs = [reduce(mul, map(len, xdims), 1) for _, xdims, _ in self.data._x_meta]
            tstart = [t for t, n in zip(tstart, n_xs) for _ in range(n)]
            tstop = [t for t, n in zip(tstop, n_xs) for _ in range(n)]
        else:
            self.tstart = self.tstart_h = tstart
            self.tstop = tstop
            tstart = [tstart] * n_x
            tstop = [tstop] * n_x

        # TRF extent in indices
        tstep = self.data.time.tstep
        i_start_by_x = np.asarray([int(round(t / tstep)) for t in tstart], np.int64)
        i_stop_by_x = np.asarray([int(ceil(t / tstep)) for t in tstop], np.int64)
        self._i_start = i_start = np.min(i_start_by_x)
        i_stop = np.max(i_stop_by_x)
        h_n_times = i_stop - i_start

        if len(self.data.segments) == 1:
            self.n_skip = h_n_times - 1

        # progress bar
        n_splits = len(self.data.splits.splits)
        pbar = tqdm(desc=f"Fitting models", total=n_y * n_splits, disable=CONFIG['tqdm'], leave=False)
        self.t_fit_start = time.time()

        # boosting
        split_results = [SplitResult(split, n_y, n_x, h_n_times) for split in self.data.splits.splits]
        if CONFIG['n_workers']:
            # Make sure cross-validations are added in the same order, otherwise
            # slight numerical differences can occur
            job_queue, result_queue = setup_workers(self.data, i_start_by_x, i_stop_by_x, delta, mindelta_, error, selective_stopping)
            stop_jobs = Event()
            thread = Thread(target=put_jobs, args=(job_queue, n_y, n_splits, stop_jobs))
            thread.start()
            # collect results
            try:
                for _ in range(n_y * n_splits):
                    i_y, i_split, h = result_queue.get()
                    split_results[i_split].add_h(i_y, h)
                    pbar.update()
            except KeyboardInterrupt:
                stop_jobs.set()
                raise
        else:
            for i_y, y_i in enumerate(self.data.y):
                for split in split_results:
                    h = boost(y_i, self.data.x, self.data.x_pads, split.split, i_start_by_x, i_stop_by_x, delta, mindelta_, error, selective_stopping)
                    split.add_h(i_y, h)
                    pbar.update()
        self.split_results = split_results
        pbar.close()
        self.t_fit_done = time.time()

    def _get_i_tests(self):
        assert self.data.splits.n_test
        return sorted({split.split.i_test for split in self.split_results})

    def _get_h(
            self,
            skip_failed: bool = False,
            i_test: int = None,  # test partition
    ):  # kernel and sgements on which to evaluate
        if i_test is None:
            split_results = self.split_results
            segments = self.data.segments
        else:
            split_results = [split for split in self.split_results if split.split.i_test == i_test]
            segments = split_results[0].split.test

        if skip_failed:
            with warnings.catch_warnings():
                warnings.filterwarnings('ignore', 'Mean of empty slice', RuntimeWarning)
                h = np.nanmean([split.h_with_nan() for split in split_results], 0)
            is_nan = np.isnan(h[:, :, 0])
            h[is_nan] = 0
        else:
            h = np.mean([split.h for split in split_results])
        return h, segments

    def _get_h_failed(self, i_test: int = None) -> np.ndarray:
        if i_test is None:
            split_results = self.split_results
        else:
            split_results = [split for split in self.split_results if split.split.i_test == i_test]
        out = np.all([split.h_failed for split in split_results], 0)
        if self.data.vector_dim:
            out = np.all(out.reshape((len(self.data.vector_dim), -1)), 0)
        return out

    def evaluate_fit(
            self,
            i_test: int = None,
            metrics: Sequence[str] = None,
            cross_fit: bool = None,
            partition_results: bool = False,
            debug: bool = False,
    ):
        """Compute average TRF and fit metrics

        Parameters
        ----------
        i_test
            Test partition index (only applies for models fit with
            cross-validation). By default, ``y`` in each test segment is
            predicted from the averaged model from the corresponding training
            sets, and all test segments are pooled for computing fit metrics.
            Set ``i_test`` with an integer to compute fit metrics for a single
            test partition.
        metrics
            Which model fit metrics to compute (default depends on the data).
            If no metrics are needed, ``metrics=()`` is faster.
        cross_fit
            Compute fit metrics from cross-validation (only applies to model
            with cross-validation; default ``True``).
        partition_results
            Keep results (TRFs and model evaluation) for each test-partition.
        debug
            Add additional attributes to the returned result.
        """
        if cross_fit is None:
            cross_fit = bool(self.data.splits.n_test)
        elif cross_fit and not self.data.splits.n_test:
            raise ValueError(f"cross_fit={cross_fit!r} for model without cross-validation")
        if partition_results and not cross_fit:
            raise ValueError(f"partition_results={partition_results!r} with cross_fit={cross_fit!r}")

        # fit evaluation
        if metrics is None:
            if self.data.vector_dim:
                metrics = [f'vec-{self.error}', f'vec-corr']
                if self.error == 'l1':
                    metrics.append('vec-corr-l1')
            else:
                metrics = [self.error, 'r', 'r_rank']

        # test sets to use
        if cross_fit:
            if i_test is None:
                i_tests = self._get_i_tests()
            else:
                i_tests = [i_test]
        elif i_test is not None:
            raise ValueError(f"i_test={i_test!r} without cross_fit")
        else:
            i_tests = None

        # hs: [(h, test_segments), ...]
        if cross_fit:
            hs = [self._get_h(True, i) for i in i_tests]
            all_segments = np.sort(np.vstack([segments for _, segments in hs]), 0)
            eval_segments = [merge_segments(all_segments, True)]
            if partition_results:
                for _, test_segments in hs:
                    eval_segments.append(merge_segments(test_segments, True))
        else:
            hs = [self._get_h(True)]
            eval_segments = merge_segments(self.data.segments, True)
            if self.n_skip:
                eval_segments[:, 0] += self.n_skip
                # check for invalid segments (negative duration)
                valid = eval_segments[:, 1] - eval_segments[:, 0] > 0
                if not np.all(valid):
                    eval_segments = eval_segments[valid]
            eval_segments = [eval_segments]

        if metrics:
            # y dimensions
            n_y = len(self.data.y)
            if self.data.vector_dim:
                n_vec = len(self.data.vector_dim)
                n_vecs = n_y // n_vec
            else:
                n_vec = n_vecs = 0

            # predicted y
            if debug:
                y_pred_iter = y_pred = np.empty(self.data.y.shape)
            elif n_vecs:
                y_pred = np.empty((n_vec, *self.data.y.shape[1:]))
                y_pred_iter = chain.from_iterable(repeat(tuple(y_pred), n_vecs))
            else:
                y_pred = np.empty(self.data.y.shape[1:])
                y_pred_iter = repeat(y_pred, n_y)

            all_evaluators, evaluators_s, evaluators_v = get_evaluators(metrics, self.data, eval_segments)

            # fit and evaluate each y
            for i_y, y_pred_i in enumerate(y_pred_iter):
                # for cross-validation, different segments are predicted by different h:
                for h, segments in hs:
                    convolve(h[i_y], self.data.x, self.data.x_pads, self._i_start, segments, y_pred_i)

                if evaluators_s:
                    for e in evaluators_s:
                        e.add_y(i_y, self.data.y[i_y], y_pred_i)

                if evaluators_v and i_y % n_vec == n_vec - 1:
                    i_vec = i_y // n_vec
                    i_y_vec = slice(i_y-n_vec+1, i_y+1)
                    y_pred_i_vec = y_pred[i_y_vec] if debug else y_pred
                    for e in evaluators_v:
                        e.add_y(i_vec, self.data.y[i_y_vec], y_pred_i_vec)

            # Package evaluators
            evaluations = {e.attr: e.get() for e in all_evaluators}
            if debug:
                evaluations['y_pred'] = self.data.package_y_like(y_pred, 'y-pred')
            if partition_results:
                partition_evaluations = {i: {e.attr: e.get(i) for e in all_evaluators} for i in i_tests}
            else:
                partition_evaluations = None
        else:
            evaluations = {}
            partition_evaluations = {i: {} for i in i_tests}

        # package h
        h_xs = [h for h, _ in hs]
        h_x = h_xs[0] if len(h_xs) == 1 else np.mean(h_xs, 0)
        h = self.data.package_kernel(h_x, self.tstart_h)
        # package model parameters
        y_mean, y_scale, x_mean, x_scale = self.data.data_scale_ndvars()
        if debug:
            evaluations['fit'] = self
        t_run = self.t_fit_done - self.t_fit_start
        # partition-specific results
        if partition_results:
            partition_results_list = []
            for i in i_tests:
                h_i = self.data.package_kernel(h_xs[i], self.tstart_h)
                evaluations_i = partition_evaluations[i]
                result = BoostingResult(
                    self.data.y_name, self.data.x_name, self.tstart, self.tstop, bool(self.data.scale_data), self.delta, self.mindelta, self.error, self.selective_stopping,
                    y_mean, y_scale, x_mean, x_scale,
                    h_i, self._get_h_failed(i), 0,
                    self.data.basis, self.data.basis_window, None,
                    self.data.y.shape[1], self.data.y_info, self.data.ydims,
                    algorithm_version=0, i_test=i, **evaluations_i)
                partition_results_list.append(result)
        else:
            partition_results_list = None
        return BoostingResult(
            # basic parameters
            self.data.y_name, self.data.x_name, self.tstart, self.tstop, bool(self.data.scale_data), self.delta, self.mindelta, self.error, self.selective_stopping,
            # data properties
            y_mean, y_scale, x_mean, x_scale,
            # results
            h, self._get_h_failed(), t_run,
            # advanced parameters
            self.data.basis, self.data.basis_window, self.data.splits,
            # advanced data properties
            self.data.y.shape[1], self.data.y_info, self.data.ydims,
            partition_results=partition_results_list,
            algorithm_version=0,
            i_test=i_test, **evaluations)

@user_activity
def boosting(
        y: NDVarArg,
        x: Union[NDVarArg, Sequence[NDVarArg]],
        tstart: Union[float, Sequence[float]],
        tstop: Union[float, Sequence[float]],
        scale_data: Union[bool, str] = True,  # normalize y and x; can be 'inplace'
        delta: float = 0.005,  # coordinate search step
        mindelta: float = None,  # narrow search by reducing delta until reaching mindelta
        error: Literal['l1', 'l2'] = 'l2',
        basis: float = 0,
        basis_window: str = 'hamming',
        partitions: int = None,  # Number of partitionings for cross-validation
        model: CategorialArg = None,
        validate: int = 1,  # Number of segments in validation set
        test: int = 0,  # Number of segments in test set
        ds: Dataset = None,
        selective_stopping: int = 0,
        partition_results: bool = False,
        debug: bool = False,
) -> BoostingResult:
    """Estimate a linear filter with coordinate descent

    Parameters
    ----------
    y : NDVar
        Signal to predict.
    x : NDVar | sequence of NDVar
        Signal to use to predict ``y``. Can be sequence of NDVars to include
        multiple predictors. Time dimension must correspond to ``y``.
    tstart : scalar | sequence of scalar
        Start of the TRF in seconds. A list can be used to specify different
        values for each item in ``x``.
    tstop : scalar | sequence of scalar
        Stop of the TRF in seconds. Format must match ``tstart``.
    scale_data : bool | 'inplace'
        Scale ``y`` and ``x`` before boosting: subtract the mean and divide by
        the standard deviation (when ``error='l2'``) or the mean absolute
        value (when ``error='l1'``). Use ``'inplace'`` to save memory by scaling
        the original objects specified as ``y`` and ``x`` instead of making a
        copy.
    delta
        Step for changes in the kernel.
    mindelta
        If the error for the training data can't be reduced, divide ``delta``
        in half until ``delta < mindelta``. The default is ``mindelta = delta``,
        i.e. ``delta`` is constant.
    error
        Error function to use (default is ``l2``).

        - ``error='l1'``: the sum of the absolute differences between ``y`` and
          ``h * x``.
        - ``error='l2'``: the sum of the squared differences between ``y`` and
          ``h * x``.

        For vector ``y``, the error is defined based on the distance in space
        for each data point.
    basis
        Use a basis of windows with this length for the kernel (by default,
        impulses are used).
    basis_window : str | scalar | tuple
        Basis window (see :func:`scipy.signal.get_window` for options; default
        is ``'hamming'``).
    partitions
        Divide the data into this many ``partitions`` for cross-validation-based
        early stopping. In each partition, ``n - 1`` segments are used for
        training, and the remaining segment is used for validation.
        If data is continuous, data are divided into contiguous segments of
        equal length (default 10).
        If data has cases, cases are divided with ``[::partitions]`` slices
        (default ``min(n_cases, 10)``; if ``model`` is specified, ``n_cases``
        is the lowest number of cases in any cell of the model).
        See :ref:`exa-data_split` example.
    model
        If data has cases, divide cases into different categories (division
        for crossvalidation is done separately for each cell).
    ds
        If provided, other parameters can be specified as string for items in
        ``ds``.
    validate
        Number of segments in validation dataset (currently has to be 1).
    test
        By default (``test=0``), the boosting algorithm uses all available data
        to estimate the kernel. Set ``test=1`` to perform *k*-fold cross-
        validation instead (with *k* = ``partitions``):
        Each partition is used as test dataset in turn, while the remaining
        ``k-1`` partitions are used to estimate the kernel. The resulting model
        fit metrics reflect the re-combination of all partitions, each one
        predicted from the corresponding, independent training set.
    selective_stopping
        By default, the boosting algorithm stops when the testing error stops
        decreasing. With ``selective_stopping=True``, boosting continues but
        excludes the predictor (one time-series in ``x``) that caused the
        increase in testing error, and continues until all predictors are
        stopped. The integer value of ``selective_stopping`` determines after
        how many steps with error increases each predictor is excluded.
    partition_results
        Keep results (TRFs and model evaluation) for each test-partition.
        This is disabled by default to reduce file size when saving results.
    debug
        Add additional attributes to the returned result.

    See Also
    --------
    plot.preview_partitions : preview data partitions for cross-validation

    Notes
    -----
    The boosting algorithm is described in [1]_.

    In order to predict data, use the :func:`convolve` function::

    >>> ds = datasets.get_uts()
    >>> ds['a1'] = epoch_impulse_predictor('uts', 'A=="a1"', ds=ds)
    >>> ds['a0'] = epoch_impulse_predictor('uts', 'A=="a0"', ds=ds)
    >>> res = boosting('uts', ['a0', 'a1'], 0, 0.5, partitions=10, model='A', ds=ds)
    >>> y_pred = convolve(res.h_scaled, ['a0', 'a1'], ds=ds)

    References
    ----------
    .. [1] David, S. V., Mesgarani, N., & Shamma, S. A. (2007). Estimating
        sparse spectro-temporal receptive fields with natural stimuli. Network:
        Computation in Neural Systems, 18(3), 191-212.
        `10.1080/09548980701609235 <https://doi.org/10.1080/09548980701609235>`_.
    """
    # scale_data
    if isinstance(scale_data, bool):
        scale_in_place = False
    elif isinstance(scale_data, str):
        if scale_data == 'inplace':
            scale_in_place = True
        else:
            raise ValueError(f"{scale_data=}")
    else:
        raise TypeError(f"{scale_data=}, need bool or str")
    # selective_stopping
    selective_stopping = int(selective_stopping)
    if selective_stopping < 0:
        raise ValueError(f"{selective_stopping=}")

    data = DeconvolutionData(y, x, ds, scale_in_place)
    data.apply_basis(basis, basis_window)
    if scale_data:
        data.normalize(error)
    data.initialize_cross_validation(partitions, model, ds, validate, test)

    fit = Boosting(data)
    fit.fit(tstart, tstop, selective_stopping, error, delta, mindelta)
    return fit.evaluate_fit(debug=debug, partition_results=partition_results)

class BoostingStep:
    __slots__ = ('i_stim', 'i_time', 'delta', 'e_train', 'e_test')

    def __init__(self, i_stim, i_time, delta_signed, e_test, e_train):
        self.i_stim = i_stim
        self.i_time = i_time
        self.delta = delta_signed
        self.e_train = e_train
        self.e_test = e_test

def boost(y, x, x_pads, split, i_start_by_x, i_stop_by_x, delta, mindelta, error, selective_stopping=0, return_history=False):
    """Estimate one filter with boosting

    Parameters
    ----------
    y : array (n_times,)
        Dependent signal, time series to predict.
    x : array (n_stims, n_times)
        Stimulus.
    x_pads : array (n_stims,)
        Padding for x.
    split : Split
        Training/validation data split.
    i_start_by_x : ndarray
        Array of i_start for trfs.
    i_stop_by_x : ndarray
        Array of i_stop for TRF.
    delta : scalar
        Step of the adjustment.
    mindelta : scalar
        Smallest delta to use. If no improvement can be found in an iteration,
        the first step is to divide delta in half, but stop if delta becomes
        smaller than ``mindelta``.
    error : str
        Error function to use.
    selective_stopping : int
        Selective stopping.
    return_history : bool
        Return error history as second return value.

    Returns
    -------
    history[best_iter] : None | array
        Winning kernel, or None if 0 is the best kernel.
    test_sse_history : list (only if ``return_history==True``)
        SSE for test data at each iteration.
    """
    delta_error_func = DELTA_ERROR_FUNC[error]
    error = ERROR_FUNC[error]
    n_stims, n_times = x.shape
    assert y.shape == (n_times,)
    i_start = np.min(i_start_by_x)
    n_times_trf = np.max(i_stop_by_x) - i_start
    h = np.zeros((n_stims, n_times_trf))
    # buffers
    y_error = y.copy()
    new_error = np.empty(h.shape)
    new_error.fill(np.inf)  # ignore values outside TRF
    new_sign = np.empty(h.shape, np.int8)
    x_active = np.ones(n_stims, dtype=np.int8)

    # history
    best_test_error = np.inf
    history = []
    i_stim = i_time = delta_signed = None
    best_iteration = 0
    # pre-assign iterators
    for i_boost in range(999999):
        # evaluate current h
        e_train = error(y_error, split.train)
        e_test = error(y_error, split.validate)
        step = BoostingStep(i_stim, i_time, delta_signed, e_test, e_train)
        history.append(step)

        # evaluate stopping conditions
        if e_test < best_test_error:
            best_test_error = e_test
            best_iteration = i_boost
        elif i_boost >= 2 and e_test > history[-2].e_test:
            if selective_stopping:
                if selective_stopping > 1:
                    n_bad = selective_stopping - 1
                    # only stop if the predictor overfits twice without intermittent improvement
                    undo = 0
                    for i in range(-2, -len(history), -1):
                        step = history[i]
                        if step.e_test > e_test:
                            break  # the error improved
                        elif step.i_stim == i_stim:
                            if step.e_test > history[i - 1].e_test:
                                # the same stimulus caused an error increase
                                if n_bad == 1:
                                    undo = i
                                    break
                                n_bad -= 1
                            else:
                                break
                else:
                    undo = -1

                if undo:
                    # revert changes
                    for i in range(-undo):
                        step = history.pop(-1)
                        h[step.i_stim, step.i_time] -= step.delta
                        update_error(y_error, x[step.i_stim], x_pads[step.i_stim], split.train_and_validate, -step.delta, step.i_time + i_start)
                    step = history[-1]
                    # disable predictor
                    x_active[i_stim] = False
                    if not np.any(x_active):
                        break
                    new_error[i_stim, :] = np.inf
            # Basic
            # -----
            # stop the iteration if all the following requirements are met
            # 1. more than 10 iterations are done
            # 2. The testing error in the latest iteration is higher than that in
            #    the previous two iterations
            elif i_boost > 10 and e_test > history[-3].e_test:
                # print("error(test) not improving in 2 steps")
                break

        # generate possible movements -> training error
        generate_options(y_error, x, x_pads, x_active, split.train, i_start, i_start_by_x, i_stop_by_x, delta_error_func, delta, new_error, new_sign)
        i_stim, i_time = np.unravel_index(np.argmin(new_error), h.shape)
        new_train_error = new_error[i_stim, i_time]
        delta_signed = new_sign[i_stim, i_time] * delta

        # If no improvements can be found reduce delta
        if new_train_error > step.e_train:
            delta *= 0.5
            if delta >= mindelta:
                i_stim = i_time = delta_signed = None
                # print("new delta: %s" % delta)
                continue
            else:
                # print("No improvement possible for training data")
                break

        # abort if we're moving in circles
        if step.delta and i_stim == step.i_stim and i_time == step.i_time and delta_signed == -step.delta:
            break

        # update h with best movement
        h[i_stim, i_time] += delta_signed
        update_error(y_error, x[i_stim], x_pads[i_stim], split.train_and_validate, delta_signed, i_time + i_start)
    else:
        raise RuntimeError("Maximum number of iterations exceeded")
    # print('  (%i iterations)' % (i_boost + 1))

    # reverse changes after best iteration
    if best_iteration:
        for step in history[-1: best_iteration: -1]:
            if step.delta:
                h[step.i_stim, step.i_time] -= step.delta
    else:
        h = None

    if return_history:
        return h, [step.e_test for step in history]
    else:
        return h

def setup_workers(data, i_start, i_stop, delta, mindelta, error, selective_stopping):
    n_y, n_times = data.y.shape
    n_x, _ = data.x.shape

    y_buffer = RawArray('d', n_y * n_times)
    y_buffer[:] = data.y.ravel()
    x_buffer = RawArray('d', n_x * n_times)
    x_buffer[:] = data.x.ravel()
    x_pads_buffer = RawArray('d', n_x)
    x_pads_buffer[:] = data.x_pads

    job_queue = mpc.Queue(200)
    result_queue = mpc.Queue(200)

    args = (y_buffer, x_buffer, x_pads_buffer, n_y, n_times, n_x, data.splits.splits, i_start, i_stop, delta, mindelta, error, selective_stopping, job_queue, result_queue)
    for _ in range(CONFIG['n_workers']):
        process = mpc.Process(target=boosting_worker, args=args)
        process.start()

    return job_queue, result_queue

def boosting_worker(y_buffer, x_buffer, x_pads_buffer, n_y, n_times, n_x, splits, i_start, i_stop, delta, mindelta, error, selective_stopping, job_queue, result_queue):
    if CONFIG['nice']:
        os.nice(CONFIG['nice'])

    y = np.frombuffer(y_buffer, np.float64, n_y * n_times).reshape((n_y, n_times))
    x = np.frombuffer(x_buffer, np.float64, n_x * n_times).reshape((n_x, n_times))
    x_pads = np.frombuffer(x_pads_buffer, np.float64, n_x)

    while True:
        i_y, i_split = job_queue.get()
        if i_y == JOB_TERMINATE:
            return
        h = boost(y[i_y], x, x_pads, splits[i_split], i_start, i_stop, delta, mindelta, error, selective_stopping)
        result_queue.put((i_y, i_split, h))

def put_jobs(queue, n_y, n_splits, stop):
    "Feed boosting jobs into a Queue"
    for job in product(range(n_y), range(n_splits)):
        queue.put(job)
        if stop.isSet():
            while not queue.empty():
                queue.get()
            break
    for _ in range(CONFIG['n_workers']):
        queue.put((JOB_TERMINATE, None))

def convolve(
        h: np.ndarray,
        x: np.ndarray,
        x_pads: np.ndarray,
        h_i_start: int,
        segments: np.ndarray = None,
        out: np.ndarray = None,
):
    """h * x with time axis matching x

    Parameters
    ----------
    h : array, (n_stims, h_n_samples)
        H.
    x : array, (n_stims, n_samples)
        X.
    x_pads : array (n_stims,)
        Padding for x.
    h_i_start : int
        Time shift of the first sample of ``h``.
    segments : array (n_segments, 2)
        Data segments.
    out : array
        Buffer for predicted ``y``.
    """
    n_x, n_times = x.shape
    h_n_times = h.shape[1]
    if segments is None:
        segments = np.array(((0, n_times),), np.int64)
    if out is None:
        out = np.zeros(n_times)
    else:
        for a, b in segments:
            out[a:b] = 0
    h_i_stop = h_i_start + h_n_times

    # padding
    h_pad = np.sum(h * x_pads.reshape((-1, 1)), 0)
    # padding for pre-
    pad_head_n_times = max(0, h_n_times + h_i_start)
    if pad_head_n_times:
        pad_head = np.zeros(pad_head_n_times)
        for i in range(min(pad_head_n_times, h_n_times)):
            pad_head[:pad_head_n_times - i] += h_pad[- i - 1]
    # padding for post-
    pad_tail_n_times = -min(0, h_i_start)
    if pad_tail_n_times:
        pad_tail = np.zeros(pad_tail_n_times)
        for i in range(pad_tail_n_times):
            pad_tail[i:] += h_pad[i]

    for start, stop in segments:
        if pad_head_n_times:
            out[start: start + pad_head_n_times] += pad_head
        if pad_tail_n_times:
            out[stop - pad_tail_n_times: stop] += pad_tail
        convolve_jit(h, x[:, start:stop], out[start:stop], h_i_start, h_i_stop)
    return out

max_stars_repo_name: christianbrodbeck/Eelbrain

max_issues_repo_name: christianbrodbeck/Eelbrain

max_forks_repo_name: christianbrodbeck/Eelbrain


text:
# @Time   : 2020/11/14
# @Author : Gaole He
# @Email  : hegaole@ruc.edu.cn

# UPDATE:
# @Time   : 2020/12/3
# @Author : Tianyi Tang
# @Email  : steventang@ruc.edu.cn

# UPDATE
# @Time   : 2021/4/12
# @Author : Lai Xu
# @Email  : tsui_lai@163.com

"""
textbox.evaluator.bleu_evaluator
#######################################
"""

import numpy as np
from fast_bleu import BLEU
from textbox.evaluator.sentence_bleu import sentence_bleu, SmoothingFunction
from textbox.evaluator.abstract_evaluator import AbstractEvaluator

class BleuEvaluator(AbstractEvaluator):
    r"""Bleu Evaluator. Now, we support metrics `'bleu'`
    """

    def __init__(self, task_type):
        self.n_grams = [1, 2, 3, 4]
        self.task_type = task_type
        self.weights = self._generate_weights()

    def _generate_weights(self):
        weight = [0] * max(self.n_grams)
        weights = {}
        for n_gram in self.n_grams:
            weight[n_gram - 1] = 1.0
            weights['bleu-{}'.format(n_gram)] = tuple(weight)
            weight[n_gram - 1] = 0.0
            avg_weight = [1. / n_gram] * n_gram
            avg_weight.extend([0. for index in range(max(self.n_grams) - n_gram)])
            weights['bleu-{}-avg'.format(n_gram)] = tuple(avg_weight)
        return weights

    def _calc_fast_bleu(self, generate_corpus, reference_corpus):
        r""" Calculate the BLEU metrics of the generated corpus in referenced corpus.

        Args:
            generate_corpus (List[List[str]]): the generated corpus
            reference_corpus (List[List[str]]): the referenced corpus
            n_grams (List): the n-gram metric to be calculated

        Returns:
            list: the BLEU results and average BLEU scores
        """

        bleu = BLEU(reference_corpus, self.weights)
        scores = bleu.get_score(generate_corpus)
        return scores

    def _calc_metrics_info(self, generate_corpus, reference_corpus):
        r"""get metrics result

        Args:
            generate_corpus: the generated corpus
            reference_corpus: the referenced corpus

        Returns:
            dict: a dict of metrics <metric> which record the results according to self.n_grams
        """

        bleu_dict = {}
        for n_gram in self.n_grams:
            bleu_dict['bleu-{}'.format(n_gram)] = []
        for n_gram in self.n_grams:
            bleu_dict['bleu-{}-avg'.format(n_gram)] = []

        if self.task_type:
            results = self._calc_fast_bleu(generate_corpus=generate_corpus, reference_corpus=reference_corpus)
            for n_gram in self.n_grams:
                bleu_dict['bleu-{}'.format(n_gram)].append(np.array(results['bleu-{}'.format(n_gram)]).mean())
                bleu_dict['bleu-{}-avg'.format(n_gram)].append(np.array(results['bleu-{}-avg'.format(n_gram)]).mean())
        else:
            for i in range(len(generate_corpus)):
                pred_sent = generate_corpus[i]
                gold_sent = reference_corpus[i]
                results = sentence_bleu(
                    hypothesis=pred_sent,
                    references=[gold_sent],
                    weights=self.weights,
                    smoothing_function=SmoothingFunction().method1
                )
                for n_gram in self.n_grams:
                    bleu_dict['bleu-{}'.format(n_gram)].append(np.array(results['bleu-{}'.format(n_gram)]).mean())
                    bleu_dict['bleu-{}-avg'.format(n_gram)].append(
                        np.array(results['bleu-{}-avg'.format(n_gram)]).mean()
                    )
        return bleu_dict


text:
#coding:utf-8

# A class of iir butterworth filter

# Check version
#  Python 3.6.4 on win32 (Windows 10)
#  numpy 1.16.3
#  scipy 1.4.1
#  matplotlib  2.1.1

import matplotlib.pyplot as plt
import numpy as np
from scipy import signal

class Class_IIR1(object):
    def __init__(self, fc=2205.0, btype='low', n_order=2, sampling_rate=44100):
        # design iir butterworth filter
        # initalize
        self.fc= fc # cut off frequency of High Pass Filter by unit is [Hz]
        self.sr= sampling_rate
        self.n_order= n_order
        self.btype= btype
        self.b, self.a= signal.butter(self.n_order, (self.fc / (self.sr/2.0)) , btype=self.btype, analog=False, output='ba') # b=Numerator(bunsi), a=denominator(bunbo)

    def filtering(self, xin):
        return signal.lfilter(self.b, self.a, xin)

    def f_show(self, worN=1024):
        # show frequency response
        wlist, fres = signal.freqz(self.b, self.a, worN=worN)

        fig = plt.figure()
        ax1 = fig.add_subplot(111)
        flist = wlist / ((2.0 * np.pi) / self.sr)
        plt.title('frequency response')
        ax1 = fig.add_subplot(111)

        plt.semilogx(flist, 20 * np.log10(abs(fres)), 'b')  # plt.plot(flist, 20 * np.log10(abs(fres)), 'b')
        plt.ylabel('Amplitude [dB]', color='b')
        plt.xlabel('Frequency [Hz]')

        ax2 = ax1.twinx()
        angles = np.unwrap(np.angle(fres))
        angles = angles / ((2.0 * np.pi) / 360.0)
        plt.semilogx(flist, angles, 'g')  # plt.plot(flist, angles, 'g')
        plt.ylabel('Angle(deg)', color='g')
        plt.grid()
        plt.axis('tight')
        plt.show()

    def N(self,):
        # if moving average is approximated as a LPF
        # n is span of moving average
        n=np.sqrt(np.power( 0.443 * self.sr / self.fc, 2.0) +1.0)
        print (n)

if __name__ == '__main__':

    # instance
    sampling_rate=44100
    fc=sampling_rate / 20  # re-sample sampling_rate / 10 ?
    lpf=Class_IIR1(fc=fc, sampling_rate=sampling_rate)
    lpf.f_show()

    lpf.N()


text:
import os
import numpy as np
from keras.callbacks import ModelCheckpoint
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers.core import Dense, Dropout
from keras.utils import np_utils

batch_size = 128
num_epochs = 20
model_dir = "./models"
(X_train, y_train), (X_test, y_test) = mnist.load_data()
X_train = X_train.reshape(60000, 784).astype('float32') / 255
X_test = X_test.reshape(10000, 784).astype('float32') / 255
y_train = np_utils.to_categorical(y_train, 10)
y_test = np_utils.to_categorical(y_test, 10)
print(X_train.shape, X_test.shape, y_train.shape, y_test.shape)

model = Sequential()
model.add(Dense(512, input_shape=(784,), activation='relu'))
model.add(Dropout(0.2))
model.add(Dense(512, activation='relu'))
model.add(Dropout(0.2))
model.add(Dense(10, activation='softmax'))

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

# save best model
checkpoint = ModelCheckpoint(filepath=os.path.join(model_dir, 'model-{epoch:02d}.h5'), save_best_only=True)
model.fit(X_train, y_train, batch_size=batch_size, nb_epoch=num_epochs,
          validation_split=0.1, callbacks=[checkpoint])

max_stars_repo_name: Bingwen-Hu/hackaway

max_issues_repo_name: Bingwen-Hu/hackaway

max_forks_repo_name: Bingwen-Hu/hackaway


text:
import unittest
import pandas as pd
import numpy as np
import tempfile
from dataset import split_dataset, get_emos_vectors, combine_all_together
from dataset import SynthDataset, EmosVectorsDataset
from dataset import _drop_duplicates_by_filename, _drop_rows_with_undefined_true_range, _mk_full_path_to_java_files

class TestDataset(unittest.TestCase):

    def _mock_synth_dataset(self) -> SynthDataset:
        columns = [
            'output_filename',
            'insertion_start',
            'insertion_end',
            'class_name',
            'target_method',
            'target_method_start_line',
            'project_id',
            'filename',
            'true_inline_range'
        ]
        return pd.DataFrame(
            [
                ('/path/to/file_1.java', '40', '59', 'Foo', 'getItem', '34', 'apache/netbeans', 'file_1.java', '[40, 60]'),
                ('/path/to/file_2.java', '3', '6', 'Boo', 'subscribe', '2', 'apache/netbeans', 'file_2.java', '[3, 10]')
            ],
            columns=columns
        )

    def _mock_emos_vectors_dataset(self) -> EmosVectorsDataset:
        columns = ['emo_uid', 'filename_origin', 'range', 'vector_str']
        return pd.DataFrame(
            [
                ('/emo/uid/1', 'file_1.java', '[40, 60]', '-0.1 0.3 -0.9'),
                ('/emo/uid/2', 'file_1.java', '[40, 50]', '-0.2 0.3 -0.9'),
                ('/emo/uid/3', 'file_2.java', '[5, 7]', '-0.1 0.3 -0.8'),
                ('/emo/uid/4', 'file_2.java', '[4, 7]', '-0.1 0.2 -0.9'),
            ],
            columns=columns
        )

    def test_split_dataset(self):
        df = pd.DataFrame(
            [('A', i) for i in range(1, 101)] + [('B', i) for i in range(101, 105)],
            columns=['project', 'id']
        )
        train, test = split_dataset(df)

        self.assertEqual(len(df.query("project == 'A'")), 100)
        self.assertEqual(len(df.query("project == 'B'")), 4)
        self.assertEqual(len(test.query("project == 'A'")), 25)
        self.assertEqual(len(test.query("project == 'B'")), 1)

    def test_get_emos_vectors(self):
        with tempfile.NamedTemporaryFile(mode='w+') as temp_csv:
            temp_csv.write(','.join(['/path/to/p1_p2_p3_p4_p5_3_8.java', '-0.1 0.3 -0.9']))
            temp_csv.write(','.join(['/path/to/p1_p2_p3_p4_p5_3_8.java', '-0.1 0.3 -0.9']))
            temp_csv.write(','.join(['/path/to/p1_p2_p3_p4_p5_5_7.java', '-0.3 -0.3 0.2']))
            temp_csv.seek(0)
            emos_vectors = get_emos_vectors(temp_csv.name)
            self.assertIsInstance(emos_vectors, pd.DataFrame)
            self.assertIn('vector_str', emos_vectors.columns)
            self.assertEqual(len(emos_vectors), 1)
            self.assertEqual(emos_vectors.range[0], '[5, 7]')
            self.assertEqual(emos_vectors.filename_origin[0], 'p1_p2_p3_p4.java')
            self.assertEqual(emos_vectors.emo_uid[0], 'p1_p2_p3_p4_p5_5_7.java')

    def test_combine_all_together(self):
        emos_vectors: EmosVectorsDataset = self._mock_emos_vectors_dataset()
        synth: SynthDataset = self._mock_synth_dataset()
        dataset = combine_all_together(emos_vectors, synth)

        columns = [
            'filename', 'emo_uid', 'vector', 'project', 'is_true_emo', 'range_len',
            'target_method', 'target_method_start_line', 'range', 'true_range'
        ]
        for col_name in columns:
            self.assertIn(col_name, dataset.columns)

        self.assertEqual(dataset.dtypes.target_method_start_line, np.int32)
        self.assertIsInstance(dataset, pd.DataFrame)
        self.assertEqual(len(dataset), 4)
        self.assertEqual(len(dataset.query('filename == "/path/to/file_1.java" and is_true_emo == True')), 1)
        self.assertEqual(dataset.query('filename == "/path/to/file_1.java" and is_true_emo == True').true_range[0], '[40, 60]')
        self.assertEqual(len(dataset.query('filename == "/path/to/file_2.java" and is_true_emo == True')), 0)

    def test_drop_duplicates_by_filename(self):
        df = pd.DataFrame({'output_filename': ['fn_1', 'fn_2', 'fn_2', 'fn_3']})
        df = _drop_duplicates_by_filename(df)
        self.assertEqual(len(df), 2)
        self.assertEqual(set(df.output_filename.values), set(df.output_filename.unique()))

    def test_drop_rows_with_undefined_true_range(self):
        df = pd.DataFrame({'true_inline_range': [None, '[3, 4]']})
        df = _drop_rows_with_undefined_true_range(df)
        self.assertEqual(len(df), 1)
        self.assertIn('[3, 4]', df.true_inline_range.values)
        self.assertNotIn(None, df.true_inline_range.values)

    def test_mk_full_path_to_java_files(self):
        df = pd.DataFrame({'output_filename': ['fn_1', 'fn_2']})
        df = _mk_full_path_to_java_files(df, '/path/to/dir')
        self.assertEqual(len(df), 2)
        for fn in df.output_filename.values:
            self.assertIn('/path/to/dir', fn)
            self.assertNotIn('/path/to/other_dir', fn)