text:
import numpy as np
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
import pandas as pd
from copy import deepcopy
from PIL import Image
from stepbystep.v2 import StepByStep
from torchvision.transforms import ToPILImage
from sklearn.linear_model import LinearRegression
from torch.optim.lr_scheduler import StepLR, ReduceLROnPlateau, MultiStepLR, CyclicLR, LambdaLR

def EWMA(past_value, current_value, alpha):
    return (1 - alpha) * past_value + alpha * current_value

def calc_ewma(values, period):
    alpha = 2 / (period + 1)
    result = []
    for v in values:
        try:
            prev_value = result[-1]
        except IndexError:
            prev_value = 0

        new_value = EWMA(prev_value, v, alpha)
        result.append(new_value)
    return np.array(result)

def correction(averaged_value, beta, steps):
    return averaged_value / (1 - (beta ** steps))

def figure1(folder='rps'):
    paper = Image.open(f'{folder}/paper/paper02-089.png')
    rock = Image.open(f'{folder}/rock/rock06ck02-100.png')
    scissors = Image.open(f'{folder}/scissors/testscissors02-006.png')

    images = [rock, paper, scissors]
    titles = ['Rock', 'Paper', 'Scissors']

    fig, axs = plt.subplots(1, 3, figsize=(12, 5))
    for ax, image, title in zip(axs, images, titles):
        ax.imshow(image)
        ax.set_xticks([])
        ax.set_yticks([])
        ax.set_title(title)

    return fig

def calc_corrected_ewma(values, period):
    ewma = calc_ewma(values, period)

    alpha = 2 / (period + 1)
    beta = 1 - alpha

    result = []
    for step, v in enumerate(ewma):
        adj_value = correction(v, beta, step + 1)
        result.append(adj_value)

    return np.array(result)

def figure2(first_images, first_labels):
    fig, axs = plt.subplots(1, 6, figsize=(12, 4))
    titles = ['Paper', 'Rock', 'Scissors']
    for i in range(6):
        image, label = ToPILImage()(first_images[i]), first_labels[i]
        axs[i].imshow(image)
        axs[i].set_xticks([])
        axs[i].set_yticks([])
        axs[i].set_title(titles[label], fontsize=12)
    fig.tight_layout()
    return fig

def plot_dist(ax, distrib_outputs, p):
    ax.hist(distrib_outputs, bins=np.linspace(0, 20, 21))
    ax.set_xlabel('Sum of Adjusted Outputs')
    ax.set_ylabel('# of Scenarios')
    ax.set_title('p = {:.2f}'.format(p))
    ax.set_ylim([0, 500])
    mean_value = distrib_outputs.mean()
    ax.plot([mean_value, mean_value], [0, 500], c='r', linestyle='--', label='Mean = {:.2f}'.format(mean_value))
    ax.legend()

def figure7(p, distrib_outputs):
    fig, ax = plt.subplots(1, 1, figsize=(6, 4))
    plot_dist(ax, distrib_outputs, p)
    fig.tight_layout()
    return fig

def figure8(ps=(0.1, 0.3, 0.5, 0.9)):
    spaced_points = torch.linspace(.1, 1.1, 11)
    fig, axs = plt.subplots(1, 4, figsize=(15, 4))
    for ax, p in zip(axs.flat, ps):
        torch.manual_seed(17)
        distrib_outputs = torch.tensor([F.linear(F.dropout(spaced_points, p=p),
                                                 weight=torch.ones(11), bias=torch.tensor(0))
                                        for _ in range(1000)])
        plot_dist(ax, distrib_outputs, p)
        ax.label_outer()
    fig.tight_layout()
    return fig

def figure9(first_images, seed=17, p=.33):
    torch.manual_seed(seed)
    fig, axs = plt.subplots(1, 3, figsize=(12, 4))
    axs[0].imshow(ToPILImage()(first_images[0]))
    axs[0].set_title('Original Image')
    axs[0].grid(False)
    axs[0].set_xticks([])
    axs[0].set_yticks([])
    axs[1].imshow(ToPILImage()(F.dropout(first_images[:1], p=p)[0]))
    axs[1].set_title('Regular Dropout')
    axs[1].grid(False)
    axs[1].set_xticks([])
    axs[1].set_yticks([])
    axs[2].imshow(ToPILImage()(F.dropout2d(first_images[:1], p=p)[0]))
    axs[2].set_title('Two-Dimensional Dropout')
    axs[2].grid(False)
    axs[2].set_xticks([])
    axs[2].set_yticks([])
    fig.tight_layout()
    return fig

def figure11(losses, val_losses, losses_nodrop, val_losses_nodrop):
    fig, axs = plt.subplots(1, 1, figsize=(10, 5))
    axs.plot(losses, 'b', label='Training Losses - Dropout')
    axs.plot(val_losses, 'r', label='Validation Losses - Dropout')
    axs.plot(losses_nodrop, 'b--', label='Training Losses - No Dropout')
    axs.plot(val_losses_nodrop, 'r--', label='Validation Losses - No Dropout')
    plt.yscale('log')
    plt.xlabel('Epochs')
    plt.ylabel('Loss')
    plt.title('Regularizing Effect')
    fig.legend(loc='lower left')
    fig.tight_layout()
    return fig

def figure15(alpha=1/3, periods=5, steps=10):
    t = np.arange(1, steps+1)
    fig, ax = plt.subplots(1, 1, figsize=(6, 4))
    ax.bar(t-1, alpha*(1-alpha)**(t-1), label='EWMA')
    ax.bar(t-1, [1/periods]*periods + [0]*(10-periods), color='r', alpha=.3, label='MA')
    ax.set_xticks(t-1)
    ax.grid(False)
    ax.set_xlabel('Lag')
    ax.set_ylabel('Weight')
    ax.set_title(r'$EWMA\ \alpha=\frac{1}{3}$ vs MA (5 periods)')
    ax.legend()
    fig.tight_layout()
    return fig

def ma_vs_ewma(values, periods=19):
    ma19 = pd.Series(values).rolling(min_periods=0, window=periods).mean()
    fig, ax = plt.subplots(1, 1, figsize=(6, 4))
    ax.plot(values, c='k', label='Temperatures')
    ax.plot(ma19, c='k', linestyle='--', label='MA')
    ax.plot(calc_ewma(values, periods), c='r', linestyle='--', label='EWMA')
    ax.plot(calc_corrected_ewma(values, periods), c='r', linestyle='-', label='Bias-corrected EWMA')
    ax.set_title('MA vs EWMA')
    ax.set_ylabel('Temperature')
    ax.set_xlabel('Days')
    ax.legend(fontsize=12)
    fig.tight_layout()
    return fig

def figure17(gradients, corrected_gradients, corrected_sq_gradients, adapted_gradients):
    fig, axs = plt.subplots(1, 3, figsize=(15, 5))
    ax = axs[0]
    ax.plot(gradients, c='k', label=r'$Gradients$')
    ax.plot(corrected_gradients, c='r', linestyle='-', label=r'$Bias-corrected\ EWMA(grad)$')
    ax.set_title('EWMA for Smoothing')
    ax.set_ylabel('Gradient')
    ax.set_xlabel('Mini-batches')
    ax.set_ylim([-1.5, 1.5])
    ax.legend(fontsize=12)

    ax = axs[1]
    ax.plot(1/(np.sqrt(corrected_sq_gradients)+1e-8), c='b', linestyle='-', label=r'$\frac{1}{\sqrt{Bias-corrected\ EWMA(grad^2)}}$')
    ax.set_title('EWMA for Scaling')
    ax.set_ylabel('Factor')
    ax.set_xlabel('Mini-batches')
    ax.set_ylim([0, 5])
    ax.legend(fontsize=12)

    ax = axs[2]
    ax.plot(gradients, c='k', label='Gradients')
    ax.plot(adapted_gradients, c='g', label='Adapted Gradients')
    ax.set_title('Gradients')
    ax.set_ylabel('Gradient')
    ax.set_xlabel('Mini-batches')
    ax.set_ylim([-1.5, 1.5])
    ax.legend(fontsize=12)
    fig.tight_layout()
    return fig

def contour_data(x_tensor, y_tensor):
    linr = LinearRegression()
    linr.fit(x_tensor, y_tensor)
    b, w = linr.intercept_, linr.coef_[0]

    # we have to split the ranges in 100 evenly spaced intervals each
    b_range = np.linspace(.7, 2.3, 101)
    w_range = np.linspace(.7, 2.3, 101)
    # meshgrid is a handy function that generates a grid of b and w
    # values for all combinations
    bs, ws = np.meshgrid(b_range, w_range)
    all_predictions = np.apply_along_axis(
        func1d=lambda x: bs + ws * x,
        axis=1,
        arr=x_tensor.numpy()
    )
    all_labels = y_tensor.numpy().reshape(-1, 1, 1)
    all_errors = (all_predictions - all_labels)
    all_losses = (all_errors ** 2).mean(axis=0)
    return b, w, bs, ws, all_losses

def compare_optimizers(model, loss_fn, optimizers, train_loader,
                       val_loader=None, layers_to_hook='', n_epochs=50):
    results = {}
    model_state = model.state_dict()

    for desc, opt in optimizers.items():
        model.load_state_dict(model_state)
        optimizer = opt['class'](model.parameters(), **opt['parms'])

        sbs = StepByStep(model, loss_fn, optimizer)
        sbs.set_loaders(train_loader, val_loader)
        sbs.capture_parameters(layers_to_hook)
        sbs.capture_gradients(layers_to_hook)
        sbs.train(n_epochs)
        sbs.remove_hooks()

        results.update({desc: {'parms': deepcopy(sbs._parameters),
                               'grads': deepcopy(sbs._gradients),
                               'losses': np.array(sbs.losses),
                               'val_losses': np.array(sbs.val_losses),
                               'state': optimizer.state_dict()}})

    return results

def plot_paths(results, b, w, bs, ws, all_losses, axs=None):
    if axs is None:
        fig, axs = plt.subplots(1, len(results), figsize=(5 * len(results), 5))
    axs = np.atleast_2d(axs)
    axs = [ax for row in axs for ax in row]
    for i, (ax, desc) in enumerate(zip(axs, results.keys())):
        biases = np.array(results[desc]['parms']['']['linear.bias']).squeeze()
        weights = np.array(results[desc]['parms']['']['linear.weight']).squeeze()
        ax.plot(biases, weights, '-o', linewidth=1, zorder=1, c='k', markersize=4)
        # Loss surface
        CS = ax.contour(bs[0, :], ws[:, 0], all_losses, cmap=plt.cm.jet, levels=12)
        ax.clabel(CS, inline=1, fontsize=10)
        ax.scatter(b, w, c='r', zorder=2, s=40)
        ax.set_xlim([.7, 2.3])
        ax.set_ylim([.7, 2.3])
        ax.set_xlabel('Bias')
        ax.set_ylabel('Weight')
        ax.set_title(desc)
        ax.label_outer()
    fig = ax.get_figure()
    fig.tight_layout()
    return fig

def plot_losses(results, axs=None):
    n = len(results.keys())
    if axs is None:
        fig, axs = plt.subplots(1, n, figsize=(5*n, 4))
    else:
        fig = axs[0].get_figure()
    for ax, k in zip(axs, results.keys()):
        ax.plot(results[k]['losses'], label='Training Loss', c='b')
        ax.plot(results[k]['val_losses'], label='Validation Loss', c='r')
        ax.set_yscale('log')
        ax.set_xlabel('Epochs')
        ax.set_ylabel('Loss')
        ax.set_ylim([1e-3, 1])
        ax.set_title(k)
        ax.legend()
    fig.tight_layout()
    return fig

def momentum(past_value, current_value, beta):
    return beta * past_value + current_value

def calc_momentum(values, beta):
    result = []
    for v in values:
        try:
            prev_value = result[-1]
        except IndexError:
            prev_value = 0

        new_value = momentum(prev_value, v, beta)
        result.append(new_value)
    return np.array(result)

def calc_nesterov(values, beta):
    result = calc_momentum(values, beta)
    return beta * result + values

def figure21(results):
    parm = 'linear.weight'

    fig, axs = plt.subplots(1, 3, figsize=(15, 5))

    for i, ax in enumerate(axs):
        desc = list(results.keys())[i]
        gradients = np.array(results[desc]['grads'][''][parm]).squeeze()
        momentums = calc_momentum(gradients, 0.9)
        nesterovs = calc_nesterov(gradients, 0.9)
        ax.plot(gradients, c='k', label='Gradients')
        if i > 0:
            ax.plot(momentums, c='r', label='Momentums')
        if i > 1:
            ax.plot(nesterovs, c='b', label='Nesterov Momentums')
        ax.set_title(desc)
        ax.set_ylabel('Gradient')
        ax.set_xlabel('Mini-batches')
        ax.set_ylim([-2, 1.5])
        ax.legend(fontsize=12)

    fig.tight_layout()
    return fig

def plot_scheduler(dummy_optimizer, dummy_scheduler, logscale=True, ax=None):
    learning_rates = []
    for i in range(12):
        current_lr = list(map(lambda d: d['lr'], dummy_scheduler.optimizer.state_dict()['param_groups']))
        learning_rates.append(current_lr)
        dummy_optimizer.step()
        if isinstance(dummy_scheduler, ReduceLROnPlateau):
            dummy_loss = 0.1
            dummy_scheduler.step(dummy_loss)
        else:
            dummy_scheduler.step()

    if ax is None:
        fig, ax = plt.subplots(1, 1, figsize=(5, 4))

    ax.plot(learning_rates)
    if logscale:
        ax.set_yscale('log')
    ax.set_xlabel('Steps')
    ax.set_ylabel('Learning Rate')
    ax.set_title(type(dummy_scheduler).__name__)
    fig = ax.get_figure()
    fig.tight_layout()
    return fig

def figure26(dummy_optimizer, dummy_schedulers):
    fig, axs = plt.subplots(1, 3, figsize=(15, 4))
    fig = plot_scheduler(dummy_optimizer, dummy_schedulers[0], ax=axs[0], logscale=False)
    fig = plot_scheduler(dummy_optimizer, dummy_schedulers[1], ax=axs[1], logscale=False)
    fig = plot_scheduler(dummy_optimizer, dummy_schedulers[2], ax=axs[2], logscale=False)
    axs[0].set_ylim([9e-5, 1e-3])
    axs[1].set_ylim([9e-5, 1e-3])
    axs[2].set_ylim([9e-5, 1e-3])
    axs[0].set_title('CyclicLR - mode=triangular')
    axs[1].set_title('CyclicLR - mode=triangular2')
    axs[2].set_title('CyclicLR - mode=exp_range')
    fig.tight_layout()
    return fig

def compare_optimizers(model, loss_fn, optimizers, train_loader, val_loader=None, schedulers=None, layers_to_hook='', n_epochs=50):
    from stepbystep.v3 import StepByStep
    results = {}
    model_state = model.state_dict()

    for desc, opt in optimizers.items():
        model.load_state_dict(model_state)

        optimizer = opt['class'](model.parameters(), **opt['parms'])

        sbs = StepByStep(model, loss_fn, optimizer)
        sbs.set_loaders(train_loader, val_loader)

        try:
            if schedulers is not None:
                sched = schedulers[desc]
                scheduler = sched['class'](optimizer, **sched['parms'])
                sbs.set_lr_scheduler(scheduler)
        except KeyError:
            pass

        sbs.capture_parameters(layers_to_hook)
        sbs.capture_gradients(layers_to_hook)
        sbs.train(n_epochs)
        sbs.remove_hooks()

        parms = deepcopy(sbs._parameters)
        grads = deepcopy(sbs._gradients)

        lrs = sbs.learning_rates[:]
        if not len(lrs):
            lrs = [list(map(lambda p: p['lr'], optimizer.state_dict()['param_groups']))] * n_epochs

        results.update({desc: {'parms': parms,
                               'grads': grads,
                               'losses': np.array(sbs.losses),
                               'val_losses': np.array(sbs.val_losses),
                               'state': optimizer.state_dict(),
                               'lrs': lrs}})

    return results

def figure28(results, b, w, bs, ws, all_losses):
    axs = []
    fig = plt.figure(figsize=(15, 12))
    for i in range(3):
        axs.append(plt.subplot2grid((5, 3), (0, i), rowspan=2))
    for i in range(3):
        axs.append(plt.subplot2grid((5, 3), (3, i), rowspan=2))
    for i in range(3):
        axs.append(plt.subplot2grid((5, 3), (2, i)))

    lrs = [results[k]['lrs'] for k in ['SGD + Momentum', 'SGD + Momentum + Step', 'SGD + Momentum + Cycle']]
    for ax, l, title in zip(axs[6:], lrs, ['No Scheduler', 'StepLR', 'CyclicLR']):
        ax.plot(l)
        ax.set_title(title)
        if title == 'CyclicLR':
            ax.set_xlabel('Mini-batches')
        else:
            ax.set_xlabel('Epochs')
        ax.set_ylabel('Learning Rate')
        ax.set_ylim([0.0, .11])

    fig = plot_paths(results, b, w, bs, ws, all_losses, axs=axs[:6])
    for ax in axs[:6]:
        ax.set_xlabel('Bias')
    fig.tight_layout()


text:
import argparse
import pathlib
import random

import gin
import matplotlib
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import numpy as np
import pandas as pd
import seaborn as sns
import tensorflow as tf
import tensorflow_datasets as tfds
from tqdm import tqdm

from scipy.signal import filtfilt
from tensorflow.python.ops.gen_array_ops import const
import tensorflow.keras.activations as activations
from sklearn.datasets import make_moons, make_circles, make_blobs
from tensorflow.keras import backend as K
from tensorflow.keras.optimizers import Adam, SGD, RMSprop
from tensorflow.keras.losses import BinaryCrossentropy
from tensorflow.keras.metrics import BinaryAccuracy, PrecisionAtRecall, RecallAtPrecision, Recall, Precision
from deel.lip.callbacks import CondenseCallback
import models

@tf.function
def logloss_support(y):
    ones = tf.ones(y.shape)
    loss = tf.nn.sigmoid_cross_entropy_with_logits(ones, y)
    return tf.reduce_mean(loss)

@tf.function
def logloss_border(y):
    border = tf.ones(y.shape) / 2.
    loss = tf.nn.sigmoid_cross_entropy_with_logits(border, y)
    return tf.reduce_mean(loss)

@tf.function
def logloss_ood(y):
    zeros = tf.zeros(y.shape)
    loss = tf.nn.sigmoid_cross_entropy_with_logits(zeros, y)
    return tf.reduce_mean(loss)

@tf.function
def logloss_3parts(y_tp, y_fp, y_tn, lbda, temperature):
    tp_to_tp = logloss_support(y_tp * temperature)
    fp_to_tn = logloss_border(y_fp * temperature)
    tn_to_tn = logloss_ood(y_tn * temperature)
    to_tn = lbda*fp_to_tn + (1-lbda)*tn_to_tn
    loss = tp_to_tp + to_tn
    return loss, tp_to_tp

@tf.function
def border_hinge(y):  # null-margin, on the other side of the frontiere
    return tf.reduce_mean(tf.nn.relu(y))

@tf.function
def border_manifold(y):  # null-margin, exactly on the frontiere
    return tf.reduce_mean(tf.abs(y))

@tf.function
def hinge(y, margin):
    return tf.reduce_mean(tf.nn.relu(margin - y))

@tf.function
def hinge_3parts(y_tp, y_fp, y_tn, lbda, alpha, margin):
    tp_to_tp = -tf.reduce_mean(y_tp) + alpha*hinge(y_tp, margin)
    to_tn = tf.reduce_mean(y_tn) + alpha*border_hinge(y_fp)
    loss = tp_to_tp + to_tn
    return loss, tf.reduce_mean(y_tp)

@tf.function
def triplet_loss(y_tp, y_fp, y_tn, alpha, margin):
    pa = y_tp - y_fp  # positive
    an = y_fp - y_tn  # negative
    delta = an - pa
    triplet = hinge(delta, margin)
    wass = tf.reduce_mean(y_tn) - tf.reduce_mean(y_tp)
    loss = wass + alpha*triplet
    return loss, tf.reduce_mean(delta)

class ManifoldForward():
    def __init__(self, model, n_stop):
        self.model = model
        self.n_stop = n_stop
    def random_feature_map(self, x):
        half = x.shape[0]//2
        if len(x.shape) == 2:
            features = x.shape[1:]
            noise = tf.random.normal(shape=(half,)+tuple(features))
            noise = tf.concat([noise, tf.zeros((half,)+tuple(features))], axis=0)
            z_act = tf.reduce_sum(x*noise, axis=list(range(1, len(noise.shape))), keepdims=True)
            return z_act
        assert len(x.shape) == 4
        features = x.shape[-1]
        idx = np.randint(features, size=half)
        idx = np.concatenate([idx, np.zeros(half)], axis=0)
        idx = np.stack([np.arange(len(idx)), idx], axis=1)
        x = tf.transpose(x, perm=[0, 3, 1, 2])  # swap last dim
        z_act = tf.gather_nd(x, tf.constant(idx))
        z_act = tf.reduce_sum(z_act, axis=[-1, -2])
        return z_act
    def sample(self, x_0):
        if self.n_stop == 0:
            return self.model(x_0, training=False), None
        x = x_0
        for layer in self.model.layers[:-self.n_stop]:
            x = layer(x, training=False)
        z_act = self.random_feature_map(x)
        for layer in self.model.layers[-self.n_stop:]:
            x = layer(x, training=False)  # yield logits in the end
        return x, z_act
    def representation_goal(self, z_act):
        return tf.reduce_mean(z_act)  # average of activations over batch

def renormalize_grads(grads):  # side-effect on parameters: mitigate vanishing gradient
    return [tf.math.l2_normalize(grad, axis=list(range(1,tf.rank(grad)))) for grad in grads]

def add_noise(grads, step_size):  # noisy gradient step
    return [grad + tf.math.sqrt(2.*step_size)*tf.random.normal(shape=grad.shape) for grad in grads]

@tf.function
def ball_project(delta, x_0, adv_radius):
    delta = tf.clip_by_norm(delta, adv_radius, axes=list(range(1,len(x_0.shape))))
    delta = tf.clip_by_value(x_0 + delta, 0., 1.) - x_0
    return delta

def l1_penalty(x):
    l1 = tf.reduce_sum(tf.abs(x), axis=list(range(1,len(x.shape))), keepdims=True)
    non_batch_dims = int(tf.size(x)) / int(x.shape[0])
    return l1 / non_batch_dims

def goal(training_type, adv_type, y):
    if (training_type, adv_type) == ('wasserstein', 'y_fp'):
        return -tf.reduce_mean(y)
    elif (training_type, adv_type) == ('wasserstein', 'y_tn'):
        return tf.reduce_mean(y)  # border_hinge(y)
    elif (training_type, adv_type) == ('logloss', 'y_fp'):
        return logloss_support(y)
    elif (training_type, adv_type) == ('logloss', 'y_tn'):
        return logloss_ood(y)

@gin.configurable
def generate_adversarial(model, x_0, adv_radius, adv_type, training_type, adv_policy,
    step_size=gin.REQUIRED,
    max_iter=gin.REQUIRED,
    n_stop=gin.REQUIRED,
    l1_regularization=gin.REQUIRED):  # l1 special case of intermdiate activation => merge into ManifoldSampler
    if adv_policy == 'adv':
        noise = tf.random.uniform(shape=x_0.shape, minval=-adv_radius, maxval=adv_radius)
        noise = ball_project(noise, x_0, adv_radius)
        delta = tf.Variable(initial_value=noise, trainable=True)
    elif adv_policy == 'gan':
        noise = tf.random.uniform(shape=x_0.shape, minval=0., maxval=1.)
        x = tf.Variable(initial_value=noise, trainable=True)
    lr = step_size * (adv_radius/max_iter)
    optimizer = SGD(learning_rate=lr)  # sufficient considering loss landscape
    manifold = ManifoldForward(model, n_stop)
    for _ in range(max_iter):
        with tf.GradientTape(watch_accessed_variables=False) as tape:
            if adv_policy == 'adv':
                tape.watch(delta)
                x = x_0 + delta
            elif adv_policy == 'gan':
                tape.watch(x)
            y, z_map = manifold.sample(x)
            loss = goal(training_type, adv_type, y)
            if z_map is not None:
                z_loss = manifold.representation_goal(z_map)
                loss = loss + z_loss
            if l1_regularization is not None:
                l1 = l1_penalty(x) if adv_policy == 'gan' else l1_penalty(delta)
                loss = loss + l1_regularization * l1
        diff_target = x if adv_policy == 'gan' else delta
        grad_f = tape.gradient(loss, [diff_target])
        grad_f = renormalize_grads(grad_f)
        optimizer.apply_gradients(zip(grad_f,[diff_target]))
        if adv_policy == 'adv':
            delta.assign(ball_project(delta, x_0, adv_radius))  # project back to ball in manifold
        elif adv_policy == 'gan':
            x.assign(tf.clip_by_value(x, 0., 1.))  # image set
    if adv_policy == 'adv':
        x = x_0+delta.value()
        x = tf.clip_by_value(x, 0., 1.)
        return x
    elif adv_policy == 'gan':
        return x.value()

def print_tensors(x_tp, x_fp, x_tn, adv_policy, epoch, step, prefix='train', grid_print=False):
    if step != 0:
        return
    if grid_print:
        d_tp_fp = tf.unstack(x_tp - x_fp); d_tp_tn = tf.unstack(x_tp - x_tn)
        x_tps = tf.unstack(x_tp); x_fps = tf.unstack(x_fp); x_tns = tf.unstack(x_tn)
        to_print = [x_tps, x_fps, d_tp_fp, x_tns, d_tp_tn]
        numcols = len(to_print); numrows = len(x_tps); index = 1
        for imgs in to_print:
            for img in imgs:
                plt.subplot(numcols,numrows,index); plt.imshow(img)
                index += 1
    else:
        if adv_policy == 'adv':
            d_tp_fp = tf.abs(x_tp - x_fp)
            d_tp_tn = tf.abs(x_tp - x_tn)
            to_print = [x_tp, x_fp, d_tp_fp, x_tn, d_tp_tn]
        elif adv_policy == 'gan':
            to_print = [x_tp, x_fp, x_tn]
        rows = [tf.concat(tf.unstack(imgs),axis=1) for imgs in to_print]
        img = tf.concat(rows, axis=0)
        plt.imshow(img)
    plt.savefig(f'samples/images/{prefix}_{epoch}.png', bbox_inches='tight')
    plt.clf()

@gin.configurable
def adv_training(model, x_tp, epoch, step,
                 training_type=gin.REQUIRED,
                 lbda=gin.REQUIRED,
                 alpha=gin.REQUIRED,
                 false_positive_radius=gin.REQUIRED,
                 true_negative_radius=gin.REQUIRED,
                 margin=gin.REQUIRED,
                 temperature=gin.REQUIRED,
                 adv_policy='gan'):
    non_batch_dims = int(tf.size(x_tp)) / int(x_tp.shape[0])
    unit_length = non_batch_dims ** 0.5  # proportionnal to the square root of the number of pixels
    false_positive_radius = tf.constant(false_positive_radius * unit_length, dtype=tf.float32)
    true_negative_radius = tf.constant(true_negative_radius * unit_length, dtype=tf.float32)
    margin = tf.constant(margin * unit_length, dtype=tf.float32)
    x_fp = generate_adversarial(model, x_tp, false_positive_radius, 'y_fp', training_type, 'gan')
    x_tn = generate_adversarial(model, x_tp, true_negative_radius, 'y_tn', training_type, 'adv')
    with tf.GradientTape() as tape:
        y_tp = model(x_tp, training=False)
        y_fp = model(x_fp, training=False)
        y_tn = model(x_tn, training=False)
        if training_type == 'wasserstein':
            loss, support_weight = hinge_3parts(y_tp, y_fp, y_tn, lbda, alpha, margin)
            # loss, support_weight = triplet_loss(y_tp, y_fp, y_tn, alpha, margin)
        elif training_type == 'logloss':
            loss, support_weight = logloss_3parts(y_tp, y_fp, y_tn, lbda, temperature)
    grads = tape.gradient(loss, model.trainable_variables)
    print_tensors(x_tp, x_fp, x_tn, adv_policy, epoch, step)
    labels_pred = tf.argmax(tf.concat([-y_tp, y_tp], axis=1), axis=1)
    return loss, support_weight, labels_pred, grads

def exp_avg(data_avg, data):
    m_coef = 0.97
    if data_avg is None:
        return data
    return m_coef*data_avg + (1-m_coef)*data

def from_logits(Metric):
    class FromLogits(Metric):
        def __init__(self, in_labels, **kwargs):
            name = Metric.__name__.lower()
            super(FromLogits, self).__init__(name=name, **kwargs)
            self.in_labels = in_labels
        def update_state(self, y_true, y_pred, sample_weights=None):
            y_true = tf.reduce_any(tf.equal(tf.cast(y_true, dtype=tf.int64), self.in_labels), axis=-1, keepdims=True)
            y_pred = tf.nn.sigmoid(y_pred)
            super(FromLogits, self).update_state(y_true, y_pred, sample_weights)
    return FromLogits

class ExpAvg():
    def __init__(self, m_coef):
        self.m_coef = m_coef
        self.state = None
    def update(self, measure):
        if self.state is None: self.state = measure
        else:
            self.state *= self.m_coef
            self.state += (1.0-self.m_coef)*measure
    def result(self):
        return f"{self.state:>2.5f}"

class Histogram(tf.keras.metrics.Metric):
    def __init__(self, in_labels, start_epoch, **kwargs):
        super(Histogram, self).__init__(name='bestacc', **kwargs)
        self.hist_true = None; self.hist_pred = None
        self.in_labels = in_labels; self.epoch = start_epoch
        import glob
        import os
        files = glob.glob('samples/hists/*')
        for f in files:
            os.remove(f)
        files = glob.glob('samples/images/*')
        for f in files:
            os.remove(f)
    def update_state(self, y_true, y_pred, sample_weight=None):
        if self.hist_pred is None:
            self.hist_pred = tf.squeeze(y_pred)
            self.hist_true = tf.squeeze(y_true)
        else:
            self.hist_pred = tf.concat([self.hist_pred,tf.squeeze(y_pred)],axis=0)
            self.hist_true = tf.concat([self.hist_true,tf.squeeze(y_true)],axis=0)
    def result(self):
        indexes = tf.argsort(self.hist_pred)
        hist_true = tf.cast(tf.reshape(self.hist_true, shape=(-1, 1)), dtype=tf.int64)
        hist_true = tf.reduce_any(tf.equal(hist_true, self.in_labels), axis=-1, keepdims=True)
        hist_true = tf.cast(hist_true, dtype=tf.float32)
        labels = tf.gather(hist_true, indexes)
        cumlabels = tf.cumsum(labels, reverse=True)
        cumrlabels = tf.cumsum(1. - labels)
        ok_pred = cumlabels + cumrlabels
        accs = ok_pred / tf.cast(tf.size(labels), dtype=tf.float32)
        max_acc = tf.reduce_max(accs)
        return max_acc
    def plot_hist(self):
        labels, _ = tf.unique(self.hist_true)
        hists = {}
        palette = dict()
        palette_in, palette_out = plt.get_cmap('Set1'), plt.get_cmap('Set3')
        hue_in, hue_out = [], []
        for idx, label in enumerate(tf.sort(labels)):
            is_label = tf.equal(self.hist_true, label)
            y_pred = tf.boolean_mask(self.hist_pred, is_label)
            label_name = f"{int(label)}"
            hists[label_name] = pd.Series(y_pred.numpy())
            if int(label) in self.in_labels.numpy():
                palette[label_name] = palette_in(idx / len(labels))
                hue_in.append(label_name)
            else:
                palette[label_name] = palette_out(idx / len(labels))
                hue_out.append(label_name)
        hue_order = hue_in + hue_out
        sns.histplot(pd.DataFrame(hists), stat='density', bins=100, palette=palette, hue_order=hue_order)
        plt.savefig(f'samples/hists/hist_{self.epoch}.png', bbox_inches='tight')
        plt.clf()
    def reset_states(self):
        self.plot_hist()
        self.hist_true = None
        self.hist_pred = None
        self.epoch += 1

@gin.configurable
def train(model, ds_train, ds_test, start_epoch,
          in_labels, num_epochs=gin.REQUIRED):
    optimizer = Adam()
    model.compile(optimizer=optimizer, loss=BinaryCrossentropy(from_logits=True),
                  metrics=[from_logits(BinaryAccuracy)(in_labels),
                           from_logits(PrecisionAtRecall)(in_labels, recall=0.95),
                           from_logits(RecallAtPrecision)(in_labels, precision=0.95),
                           from_logits(Recall)(in_labels), from_logits(Precision)(in_labels),
                           Histogram(in_labels, start_epoch)], run_eagerly=True)
    num_batchs = sum(1 for _ in ds_train)
    model.evaluate(ds_test)
    for epoch in range(start_epoch, num_epochs):
        loss_avg, support_weight_avg = ExpAvg(0.97), ExpAvg(0.97)
        print(f"Epoch {epoch}:")
        progress = tqdm(total=num_batchs, ascii=True)
        for step, (x, _) in enumerate(ds_train):
            loss, support_weight, labels_pred, grads = adv_training(model, x, epoch, step)
            optimizer.apply_gradients(zip(grads, model.trainable_variables))
            model.condense()
            loss_avg.update(loss); support_weight_avg.update(support_weight)
            desc = f'Loss={loss_avg.result()} Wsupport={support_weight_avg.result()}'
            progress.set_description(desc=desc)
            progress.update(1)
        progress.close()
        model.evaluate(ds_test)
        model.save_weights(f"checkpoints/model_{epoch}.h5")
        if epoch>=3:
            file_to_rem = pathlib.Path(f"checkpoints/model_{epoch-3}.h5")
            file_to_rem.unlink(missing_ok=True)

def normalize_img(image, label):
  """Normalizes images: `uint8` -> `float32`."""
  return tf.cast(image, tf.float32) / 255., label

def process_dataset(name, batch_size=gin.REQUIRED):
    (ds_train, ds_test), ds_info = tfds.load(
        name,
        split=['train', 'test'],
        shuffle_files=True,
        as_supervised=True,
        with_info=True,
    )
    ds_train = ds_train.map(normalize_img, num_parallel_calls=tf.data.experimental.AUTOTUNE)
    ds_test = ds_test.map(normalize_img, num_parallel_calls=tf.data.experimental.AUTOTUNE)
    return ds_train, ds_test, ds_info

@gin.configurable
def one_class_dataset(ds, num_classes, num_examples, split,
                      batch_size=gin.REQUIRED,
                      in_labels=gin.REQUIRED,
                      out_labels=gin.REQUIRED):
    assert all(0<=y<num_classes for y in in_labels) and all(0<=y<num_classes for y in out_labels)
    in_labels = tf.constant(in_labels, dtype=tf.int64)
    in_out_labels = tf.concat([in_labels, tf.constant(out_labels, dtype=tf.int64)], axis=0)
    if split == 'train':
        ds = ds.filter(lambda _, y: tf.reduce_any(tf.equal(y, in_labels)))
        ds = ds.cache()  # cache before shuffle for different epochs
        ds = ds.shuffle(num_examples)
    elif split == 'test':
        ds = ds.shuffle(num_examples)
        ds = ds.filter(lambda _, y: tf.reduce_any(tf.equal(y, in_out_labels)))
        ds = ds.cache()  # cache after shuffle
    ds = ds.batch(batch_size)
    ds = ds.prefetch(10)
    return ds, in_labels

@gin.configurable
def model_params(in_put_shape, k_lip=gin.REQUIRED, scale=gin.REQUIRED,
                 niter_spectral=gin.REQUIRED, niter_bjorck=gin.REQUIRED, bjorck_forward=gin.REQUIRED):
    tf.keras.backend.set_image_data_format('channels_last')
    return models.get_cnn_baseline(input_shape, k_lip, scale, niter_spectral, niter_bjorck, bjorck_forward)

def dummy_forward(model, ds_test):
    _ = model.predict(next(ds_test.__iter__())[0])  # garbage
    model.condense()  # compute after forward

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-c', '--config', action='store', dest='config', type=str, help='Gin Configuration file')
    parser.add_argument('-d', '--dataset', action='store', dest='dataset', type=str, help='Dataset for Training')
    parser.add_argument('-r', '--resume', action='store', dest='resume', type=str, required=False, default='', help='Resume training at epoch')
    parser.add_argument('-t', '--val_on_trainset', action='store_true', dest='val_on_trainset',
                                                   required=False, default=False,
                                                   help='Use train set for validation (monitore overfiting)')
    args = parser.parse_args()
    gin.parse_config_file(args.config)
    trainset, testset, infos = process_dataset(args.dataset)
    num_classes = infos.features['label'].num_classes
    ds_train, _ = one_class_dataset(trainset, num_classes, infos.splits['train'].num_examples, split='train')
    if args.val_on_trainset:
        num_examples = infos.splits['train'].num_examples
        ds_test, in_labels = one_class_dataset(trainset, num_classes, num_examples, split='test', batch_size=100)
    else:
        num_examples = infos.splits['test'].num_examples
        ds_test, in_labels = one_class_dataset(testset, num_classes, num_examples, split='test', batch_size=100)
    input_shape = infos.features['image'].shape
    model = model_params(input_shape)
    dummy_forward(model, ds_test)  # init variables
    if args.resume == '':
        epoch = 0
    else:
        epoch = int(args.resume.split('_')[-1].split('.')[0]) + 1
        print(f"Load file {args.resume} and resume at epoch {epoch}")
        model.load_weights(args.resume)
    train(model, ds_train, ds_test, epoch, in_labels)

ext: py

max_stars_repo_name: Algue-Rythme/anomaly

max_issues_repo_name: Algue-Rythme/anomaly

max_forks_repo_name: Algue-Rythme/anomaly

reason: import numpy,from scipy


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

import os
import sys
import copy
import logging
import atexit
import pickle
import tempfile
import numpy as np
import subprocess as sp
from typing import Optional, List, Union, Tuple, Dict, Any
from collections.abc import Callable
from datetime import datetime

from lithops import constants
from lithops.future import ResponseFuture
from lithops.invokers import create_invoker
from lithops.storage import InternalStorage
from lithops.wait import wait, ALL_COMPLETED, THREADPOOL_SIZE, WAIT_DUR_SEC
from lithops.job import create_map_job, create_reduce_job
from lithops.config import default_config, \
    extract_localhost_config, extract_standalone_config, \
    extract_serverless_config, get_log_info, extract_storage_config
from lithops.constants import LOCALHOST, CLEANER_DIR, \
    SERVERLESS, STANDALONE
from lithops.utils import is_notebook, setup_lithops_logger, \
    is_lithops_worker, create_executor_id, create_futures_list
from lithops.localhost.localhost import LocalhostHandler
from lithops.standalone.standalone import StandaloneHandler
from lithops.serverless.serverless import ServerlessHandler
from lithops.storage.utils import create_job_key, CloudObject
from lithops.monitor import JobMonitor
from lithops.utils import FuturesList

logger = logging.getLogger(__name__)

class FunctionExecutor:
    """
    Executor abstract class that contains the common logic for the Localhost, Serverless and Standalone executors

    :param mode: Execution mode. One of: localhost, serverless or standalone
    :param config: Settings passed in here will override those in lithops_config
    :param backend: Compute backend to run the functions
    :param storage: Storage backend to store Lithops data
    :param runtime: Name of the runtime to run the functions
    :param runtime_memory: Memory (in MB) to use to run the functions
    :param monitoring: Monitoring system implementation. One of: storage, rabbitmq
    :param workers: Max number of parallel workers
    :param worker_processes: Worker granularity, number of concurrent/parallel processes in each worker
    :param remote_invoker: Spawn a function that will perform the actual job invocation (True/False)
    :param log_level: Log level printing (INFO, DEBUG, ...). Set it to None to hide all logs. If this is param is set, all logging params in config are disabled
    """

    def __init__(self,
                 mode: Optional[str] = None,
                 config: Optional[Dict[str, Any]] = None,
                 backend: Optional[str] = None,
                 storage: Optional[str] = None,
                 runtime: Optional[str] = None,
                 runtime_memory: Optional[int] = None,
                 monitoring: Optional[str] = None,
                 workers: Optional[int] = None,
                 worker_processes: Optional[int] = None,
                 remote_invoker: Optional[bool] = None,
                 log_level: Optional[str] = False):
        self.is_lithops_worker = is_lithops_worker()
        self.executor_id = create_executor_id()
        self.futures = []
        self.cleaned_jobs = set()
        self.total_jobs = 0
        self.last_call = None

        # setup lithops logging
        if not self.is_lithops_worker:
            # if is lithops worker, logging has been set up in entry_point.py
            if log_level:
                setup_lithops_logger(log_level)
            elif log_level is False and logger.getEffectiveLevel() == logging.WARNING:
                # Set default logging from config
                setup_lithops_logger(*get_log_info(config))

        # overwrite user-provided parameters
        config_ow = {'lithops': {}}
        if runtime is not None:
            config_ow['runtime'] = runtime
        if runtime_memory is not None:
            config_ow['runtime_memory'] = int(runtime_memory)
        if remote_invoker is not None:
            config_ow['remote_invoker'] = remote_invoker
        if mode is not None:
            config_ow['lithops']['mode'] = mode
        if backend is not None:
            config_ow['lithops']['backend'] = backend
        if storage is not None:
            config_ow['lithops']['storage'] = storage
        if workers is not None:
            config_ow['lithops']['workers'] = workers
        if monitoring is not None:
            config_ow['lithops']['monitoring'] = monitoring
        if worker_processes is not None:
            config_ow['lithops']['worker_processes'] = worker_processes

        self.config = default_config(copy.deepcopy(config), config_ow)

        self.data_cleaner = self.config['lithops'].get('data_cleaner', True)
        if self.data_cleaner and not self.is_lithops_worker:
            spawn_cleaner = int(self.executor_id.split('-')[1]) == 0
            atexit.register(self.clean, spawn_cleaner=spawn_cleaner,
                            clean_cloudobjects=False)

        storage_config = extract_storage_config(self.config)
        self.internal_storage = InternalStorage(storage_config)
        self.storage = self.internal_storage.storage

        self.backend = self.config['lithops']['backend']
        self.mode = self.config['lithops']['mode']

        if self.mode == LOCALHOST:
            localhost_config = extract_localhost_config(self.config)
            self.compute_handler = LocalhostHandler(localhost_config)
        elif self.mode == SERVERLESS:
            serverless_config = extract_serverless_config(self.config)
            self.compute_handler = ServerlessHandler(serverless_config, self.internal_storage)
        elif self.mode == STANDALONE:
            standalone_config = extract_standalone_config(self.config)
            self.compute_handler = StandaloneHandler(standalone_config)

        # Create the monitoring system
        monitoring_backend = self.config['lithops']['monitoring'].lower()
        self.job_monitor = JobMonitor(
            executor_id=self.executor_id,
            internal_storage=self.internal_storage,
            backend=monitoring_backend,
            config=self.config.get(monitoring_backend)
        )

        # Create the invoker
        self.invoker = create_invoker(
            config=self.config,
            executor_id=self.executor_id,
            internal_storage=self.internal_storage,
            compute_handler=self.compute_handler,
            job_monitor=self.job_monitor
        )

        logger.debug(f'Function executor for {self.backend} created with ID: {self.executor_id}')

        self.log_path = None

    def __enter__(self):
        """ Context manager method """
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """ Context manager method """
        self.job_monitor.stop()
        self.invoker.stop()
        self.compute_handler.clear()

    def _create_job_id(self, call_type):
        job_id = str(self.total_jobs).zfill(3)
        self.total_jobs += 1
        return '{}{}'.format(call_type, job_id)

    def call_async(self,
                   func: Callable,
                   data: Union[List[Any], Tuple[Any, ...], Dict[str, Any]],
                   extra_env: Optional[Dict] = None,
                   runtime_memory: Optional[int] = None,
                   timeout: Optional[int] = None,
                   include_modules: Optional[List] = [],
                   exclude_modules: Optional[List] = []) -> ResponseFuture:
        """
        For running one function execution asynchronously.

        :param func: The function to map over the data.
        :param data: Input data. Arguments can be passed as a list or tuple, or as a dictionary for keyword arguments.
        :param extra_env: Additional env variables for function environment.
        :param runtime_memory: Memory to use to run the function.
        :param timeout: Time that the function has to complete its execution before raising a timeout.
        :param include_modules: Explicitly pickle these dependencies.
        :param exclude_modules: Explicitly keep these modules from pickled dependencies.

        :return: Response future.
        """
        job_id = self._create_job_id('A')
        self.last_call = 'call_async'

        runtime_meta = self.invoker.select_runtime(job_id, runtime_memory)

        job = create_map_job(config=self.config,
                             internal_storage=self.internal_storage,
                             executor_id=self.executor_id,
                             job_id=job_id,
                             map_function=func,
                             iterdata=[data],
                             runtime_meta=runtime_meta,
                             runtime_memory=runtime_memory,
                             extra_env=extra_env,
                             include_modules=include_modules,
                             exclude_modules=exclude_modules,
                             execution_timeout=timeout)

        futures = self.invoker.run_job(job)
        self.futures.extend(futures)

        return futures[0]

    def map(self,
            map_function: Callable,
            map_iterdata: List[Union[List[Any], Tuple[Any, ...], Dict[str, Any]]],
            chunksize: Optional[int] = None,
            extra_args: Optional[Union[List[Any], Tuple[Any, ...], Dict[str, Any]]] = None,
            extra_env: Optional[Dict[str, str]] = None,
            runtime_memory: Optional[int] = None,
            obj_chunk_size: Optional[int] = None,
            obj_chunk_number: Optional[int] = None,
            timeout: Optional[int] = None,
            include_modules: Optional[List[str]] = [],
            exclude_modules: Optional[List[str]] = []) -> FuturesList:
        """
        Spawn multiple function activations based on the items of an input list.

        :param map_function: The function to map over the data
        :param map_iterdata: An iterable of input data (e.g python list).
        :param chunksize: Split map_iteradata in chunks of this size. Lithops spawns 1 worker per resulting chunk
        :param extra_args: Additional arguments to pass to each map_function activation
        :param extra_env: Additional environment variables for function environment
        :param runtime_memory: Memory (in MB) to use to run the functions
        :param obj_chunk_size: Used for data processing. Chunk size to split each object in bytes. Must be >= 1MiB. 'None' for processing the whole file in one function activation
        :param obj_chunk_number: Used for data processing. Number of chunks to split each object. 'None' for processing the whole file in one function activation. chunk_n has prevalence over chunk_size if both parameters are set
        :param timeout: Max time per function activation (seconds)
        :param include_modules: Explicitly pickle these dependencies. All required dependencies are pickled if default empty list. No one dependency is pickled if it is explicitly set to None
        :param exclude_modules: Explicitly keep these modules from pickled dependencies. It is not taken into account if you set include_modules.

        :return: A list with size `len(map_iterdata)` of futures for each job (Futures are also internally stored by Lithops).
        """

        job_id = self._create_job_id('M')
        self.last_call = 'map'

        runtime_meta = self.invoker.select_runtime(job_id, runtime_memory)

        job = create_map_job(config=self.config,
                             internal_storage=self.internal_storage,
                             executor_id=self.executor_id,
                             job_id=job_id,
                             map_function=map_function,
                             iterdata=map_iterdata,
                             chunksize=chunksize,
                             runtime_meta=runtime_meta,
                             runtime_memory=runtime_memory,
                             extra_env=extra_env,
                             include_modules=include_modules,
                             exclude_modules=exclude_modules,
                             execution_timeout=timeout,
                             extra_args=extra_args,
                             obj_chunk_size=obj_chunk_size,
                             obj_chunk_number=obj_chunk_number)

        futures = self.invoker.run_job(job)
        self.futures.extend(futures)

        if isinstance(map_iterdata, FuturesList):
            for fut in map_iterdata:
                fut._produce_output = False

        return create_futures_list(futures, self)

    def map_reduce(self,
                   map_function: Callable,
                   map_iterdata: List[Union[List[Any], Tuple[Any, ...], Dict[str, Any]]],
                   reduce_function: Callable,
                   chunksize: Optional[int] = None,
                   extra_args: Optional[Union[List[Any], Tuple[Any, ...], Dict[str, Any]]] = None,
                   extra_env: Optional[Dict[str, str]] = None,
                   map_runtime_memory: Optional[int] = None,
                   reduce_runtime_memory: Optional[int] = None,
                   obj_chunk_size: Optional[int] = None,
                   obj_chunk_number: Optional[int] = None,
                   timeout: Optional[int] = None,
                   reducer_one_per_object: Optional[bool] = False,
                   reducer_wait_local: Optional[bool] = False,
                   include_modules: Optional[List[str]] = [],
                   exclude_modules: Optional[List[str]] = []) -> FuturesList:
        """
        Map the map_function over the data and apply the reduce_function across all futures.

        :param map_function: The function to map over the data
        :param map_iterdata: An iterable of input data
        :param reduce_function: The function to reduce over the futures
        :param chunksize: Split map_iteradata in chunks of this size. Lithops spawns 1 worker per resulting chunk. Default 1
        :param extra_args: Additional arguments to pass to function activation. Default None
        :param extra_env: Additional environment variables for action environment. Default None
        :param map_runtime_memory: Memory to use to run the map function. Default None (loaded from config)
        :param reduce_runtime_memory: Memory to use to run the reduce function. Default None (loaded from config)
        :param obj_chunk_size: the size of the data chunks to split each object. 'None' for processing the whole file in one function activation
        :param obj_chunk_number: Number of chunks to split each object. 'None' for processing the whole file in one function activation
        :param timeout: Time that the functions have to complete their execution before raising a timeout
        :param reducer_one_per_object: Set one reducer per object after running the partitioner
        :param reducer_wait_local: Wait for results locally
        :param include_modules: Explicitly pickle these dependencies.
        :param exclude_modules: Explicitly keep these modules from pickled dependencies.

        :return: A list with size `len(map_iterdata)` of futures.
        """
        self.last_call = 'map_reduce'
        map_job_id = self._create_job_id('M')

        runtime_meta = self.invoker.select_runtime(map_job_id, map_runtime_memory)

        map_job = create_map_job(config=self.config,
                                 internal_storage=self.internal_storage,
                                 executor_id=self.executor_id,
                                 job_id=map_job_id,
                                 map_function=map_function,
                                 iterdata=map_iterdata,
                                 chunksize=chunksize,
                                 runtime_meta=runtime_meta,
                                 runtime_memory=map_runtime_memory,
                                 extra_args=extra_args,
                                 extra_env=extra_env,
                                 obj_chunk_size=obj_chunk_size,
                                 obj_chunk_number=obj_chunk_number,
                                 include_modules=include_modules,
                                 exclude_modules=exclude_modules,
                                 execution_timeout=timeout)

        map_futures = self.invoker.run_job(map_job)
        self.futures.extend(map_futures)

        if isinstance(map_iterdata, FuturesList):
            for fut in map_iterdata:
                fut._produce_output = False

        if reducer_wait_local:
            self.wait(map_futures)

        reduce_job_id = map_job_id.replace('M', 'R')

        runtime_meta = self.invoker.select_runtime(reduce_job_id, reduce_runtime_memory)

        reduce_job = create_reduce_job(config=self.config,
                                       internal_storage=self.internal_storage,
                                       executor_id=self.executor_id,
                                       reduce_job_id=reduce_job_id,
                                       reduce_function=reduce_function,
                                       map_job=map_job,
                                       map_futures=map_futures,
                                       runtime_meta=runtime_meta,
                                       runtime_memory=reduce_runtime_memory,
                                       reducer_one_per_object=reducer_one_per_object,
                                       extra_env=extra_env,
                                       include_modules=include_modules,
                                       exclude_modules=exclude_modules)

        reduce_futures = self.invoker.run_job(reduce_job)
        self.futures.extend(reduce_futures)

        for f in map_futures:
            f._produce_output = False

        return create_futures_list(map_futures + reduce_futures, self)

    def wait(self,
             fs: Optional[Union[ResponseFuture, FuturesList, List[ResponseFuture]]] = None,
             throw_except: Optional[bool] = True,
             return_when: Optional[Any] = ALL_COMPLETED,
             download_results: Optional[bool] = False,
             timeout: Optional[int] = None,
             threadpool_size: Optional[int] = THREADPOOL_SIZE,
             wait_dur_sec: Optional[int] = WAIT_DUR_SEC) -> Tuple[FuturesList, FuturesList]:
        """
        Wait for the Future instances (possibly created by different Executor instances)
        given by fs to complete. Returns a named 2-tuple of sets. The first set, named done,
        contains the futures that completed (finished or cancelled futures) before the wait
        completed. The second set, named not_done, contains the futures that did not complete
        (pending or running futures). timeout can be used to control the maximum number of
        seconds to wait before returning.

        :param fs: Futures list. Default None
        :param throw_except: Re-raise exception if call raised. Default True
        :param return_when: One of `ALL_COMPLETED`, `ANY_COMPLETED`, `ALWAYS`
        :param download_results: Download results. Default false (Only get statuses)
        :param timeout: Timeout of waiting for results
        :param threadpool_size: Number of threads to use. Default 64
        :param wait_dur_sec: Time interval between each check

        :return: `(fs_done, fs_notdone)` where `fs_done` is a list of futures that have completed and `fs_notdone` is a list of futures that have not completed.
        """
        futures = fs or self.futures
        if type(futures) != list and type(futures) != FuturesList:
            futures = [futures]

        # Start waiting for results
        try:
            wait(fs=futures,
                 internal_storage=self.internal_storage,
                 job_monitor=self.job_monitor,
                 download_results=download_results,
                 throw_except=throw_except,
                 return_when=return_when,
                 timeout=timeout,
                 threadpool_size=threadpool_size,
                 wait_dur_sec=wait_dur_sec)

        except (KeyboardInterrupt, Exception) as e:
            self.invoker.stop()
            self.job_monitor.stop()
            if not fs and is_notebook():
                del self.futures[len(self.futures) - len(futures):]
            if self.data_cleaner and not self.is_lithops_worker:
                self.clean(clean_cloudobjects=False, force=True)
            raise e

        finally:
            present_jobs = {f.job_key for f in futures}
            if self.data_cleaner and not self.is_lithops_worker:
                self.compute_handler.clear(present_jobs)
                self.clean(clean_cloudobjects=False)

        if download_results:
            fs_done = [f for f in futures if f.done]
            fs_notdone = [f for f in futures if not f.done]
        else:
            fs_done = [f for f in futures if f.success or f.done]
            fs_notdone = [f for f in futures if not f.success and not f.done]

        return create_futures_list(fs_done, self), create_futures_list(fs_notdone, self)

    def get_result(self,
                   fs: Optional[Union[ResponseFuture, FuturesList, List[ResponseFuture]]] = None,
                   throw_except: Optional[bool] = True,
                   timeout: Optional[int] = None,
                   threadpool_size: Optional[int] = THREADPOOL_SIZE,
                   wait_dur_sec: Optional[int] = WAIT_DUR_SEC):
        """
        For getting the results from all function activations

        :param fs: Futures list. Default None
        :param throw_except: Reraise exception if call raised. Default True.
        :param timeout: Timeout for waiting for results.
        :param threadpool_size: Number of threads to use. Default 128
        :param wait_dur_sec: Time interval between each check.
        :return: The result of the future/s
        """
        fs_done, _ = self.wait(fs=fs, throw_except=throw_except,
                               timeout=timeout, download_results=True,
                               threadpool_size=threadpool_size,
                               wait_dur_sec=wait_dur_sec)
        result = []
        fs_done = [f for f in fs_done if not f.futures and f._produce_output]
        for f in fs_done:
            if fs:
                # Process futures provided by the user
                result.append(f.result(throw_except=throw_except,
                                       internal_storage=self.internal_storage))
            elif not fs and not f._read:
                # Process internally stored futures
                result.append(f.result(throw_except=throw_except,
                                       internal_storage=self.internal_storage))
                f._read = True

        logger.debug(f'ExecutorID {self.executor_id} - Finished getting results')

        if len(result) == 1 and self.last_call != 'map':
            return result[0]

        return result

    def plot(self,
             fs: Optional[Union[ResponseFuture, List[ResponseFuture]]] = None,
             dst: Optional[str] = None):
        """
        Creates timeline and histogram of the current execution in dst_dir.

        :param fs: list of futures.
        :param dst: destination path to save .png plots.
        """
        ftrs = self.futures if not fs else fs

        if type(ftrs) != list:
            ftrs = [ftrs]

        ftrs_to_plot = [f for f in ftrs if (f.success or f.done) and not f.error]

        if not ftrs_to_plot:
            logger.debug(f'ExecutorID {self.executor_id} - No futures ready to plot')
            return

        logging.getLogger('matplotlib').setLevel(logging.WARNING)
        from lithops.plots import create_timeline, create_histogram

        logger.info(f'ExecutorID {self.executor_id} - Creating execution plots')

        create_timeline(ftrs_to_plot, dst)
        create_histogram(ftrs_to_plot, dst)

    def clean(self,
              fs: Optional[Union[ResponseFuture, List[ResponseFuture]]] = None,
              cs: Optional[List[CloudObject]] = None,
              clean_cloudobjects: Optional[bool] = True,
              spawn_cleaner: Optional[bool] = True,
              force: Optional[bool] = False):
        """
        Deletes all the temp files from storage. These files include the function,
        the data serialization and the function invocation results. It can also clean
        cloudobjects.

        :param fs: List of futures to clean
        :param cs: List of cloudobjects to clean
        :param clean_cloudobjects: Delete all cloudobjects created with this executor
        :param spawn_cleaner: Spawn cleaner background process
        :param force: Clean all future objects even if they have not benn completed
        """
        os.makedirs(CLEANER_DIR, exist_ok=True)

        def save_data_to_clean(data):
            with tempfile.NamedTemporaryFile(dir=CLEANER_DIR, delete=False) as temp:
                pickle.dump(data, temp)

        if cs:
            data = {
                'cos_to_clean': list(cs),
                'storage_config': self.internal_storage.get_storage_config()
            }
            save_data_to_clean(data)
            if not fs:
                return

        futures = fs or self.futures
        futures = [futures] if type(futures) != list else futures
        present_jobs = {create_job_key(f.executor_id, f.job_id) for f in futures
                        if (f.executor_id.count('-') == 1 and f.done) or force}
        jobs_to_clean = present_jobs - self.cleaned_jobs

        if jobs_to_clean:
            logger.info(f'ExecutorID {self.executor_id} - Cleaning temporary data')
            data = {
                'jobs_to_clean': jobs_to_clean,
                'clean_cloudobjects': clean_cloudobjects,
                'storage_config': self.internal_storage.get_storage_config()
            }
            save_data_to_clean(data)
            self.cleaned_jobs.update(jobs_to_clean)

        if (jobs_to_clean or cs) and spawn_cleaner:
            cmdstr = [sys.executable, '-m', 'lithops.scripts.cleaner']
            sp.Popen(' '.join(cmdstr), shell=True)

    def job_summary(self,
                    cloud_objects_n: Optional[int] = 0):
        """
        Logs information of a job executed by the calling function executor.
        currently supports: code_engine, ibm_vpc and ibm_cf.

        :param cloud_objects_n: number of cloud object used in COS, declared by user.
        """
        import pandas as pd

        def init():
            headers = ['Job_ID', 'Function', 'Invocations', 'Memory(MB)', 'AvgRuntime', 'Cost', 'CloudObjects']
            pd.DataFrame([], columns=headers).to_csv(self.log_path, index=False)

        def append(content):
            """ appends job information to log file."""
            pd.DataFrame(content).to_csv(self.log_path, mode='a', header=False, index=False)

        def append_summary():
            """ add a summary row to the log file"""
            df = pd.read_csv(self.log_path)
            total_average = sum(df.AvgRuntime * df.Invocations) / df.Invocations.sum()
            total_row = pd.DataFrame([['Summary', ' ', df.Invocations.sum(), df['Memory(MB)'].sum(),
                                       round(total_average, 10), df.Cost.sum(), cloud_objects_n]])
            total_row.to_csv(self.log_path, mode='a', header=False, index=False)

        def get_object_num():
            """returns cloud objects used up to this point, using this function executor. """
            df = pd.read_csv(self.log_path)
            return float(df.iloc[-1].iloc[-1])

        # Avoid logging info unless chosen computational backend is supported.
        if hasattr(self.compute_handler.backend, 'calc_cost'):

            if self.log_path:  # retrieve cloud_objects_n from last log file
                cloud_objects_n += get_object_num()
            else:
                self.log_path = os.path.join(constants.LOGS_DIR, datetime.now().strftime("%Y-%m-%d_%H:%M:%S.csv"))
            # override current logfile
            init()

            futures = self.futures
            if type(futures) != list:
                futures = [futures]

            memory = []
            runtimes = []
            curr_job_id = futures[0].job_id
            job_func = futures[0].function_name  # each job is conducted on a single function

            for future in futures:
                if curr_job_id != future.job_id:
                    cost = self.compute_handler.backend.calc_cost(runtimes, memory)
                    append([[curr_job_id, job_func, len(runtimes), sum(memory),
                             np.round(np.average(runtimes), 10), cost, ' ']])

                    # updating next iteration's variables:
                    curr_job_id = future.job_id
                    job_func = future.function_name
                    memory.clear()
                    runtimes.clear()

                memory.append(future.runtime_memory)
                runtimes.append(future.stats['worker_exec_time'])

            # appends last Job-ID
            cost = self.compute_handler.backend.calc_cost(runtimes, memory)
            append([[curr_job_id, job_func, len(runtimes), sum(memory),
                     np.round(np.average(runtimes), 10), cost, ' ']])
            # append summary row to end of the dataframe
            append_summary()

        else:  # calc_cost() doesn't exist for chosen computational backend.
            logger.warning("Could not log job: {} backend isn't supported by this function."
                           .format(self.compute_handler.backend.name))
            return
        logger.info("View log file logs at {}".format(self.log_path))

class LocalhostExecutor(FunctionExecutor):
    """
    Initialize a LocalhostExecutor class.

    :param config: Settings passed in here will override those in config file.
    :param runtime: Runtime name to use.
    :param storage: Name of the storage backend to use.
    :param worker_processes: Worker granularity, number of concurrent/parallel processes in each worker
    :param monitoring: monitoring system.
    :param log_level: log level to use during the execution.
    """

    def __init__(self,
                 config: Optional[Dict[str, Any]] = None,
                 runtime: Optional[int] = None,
                 storage: Optional[str] = None,
                 worker_processes: Optional[int] = None,
                 monitoring: Optional[str] = None,
                 log_level: Optional[str] = False):

        super().__init__(backend=LOCALHOST,
                         config=config,
                         runtime=runtime,
                         storage=storage or LOCALHOST,
                         log_level=log_level,
                         monitoring=monitoring,
                         worker_processes=worker_processes)

class ServerlessExecutor(FunctionExecutor):
    """
    Initialize a ServerlessExecutor class.

    :param config: Settings passed in here will override those in config file
    :param runtime: Runtime name to use
    :param runtime_memory: memory to use in the runtime
    :param backend: Name of the serverless compute backend to use
    :param storage: Name of the storage backend to use
    :param workers: Max number of concurrent workers
    :param worker_processes: Worker granularity, number of concurrent/parallel processes in each worker
    :param monitoring: monitoring system
    :param remote_invoker: Spawn a function that will perform the actual job invocation (True/False)
    :param log_level: log level to use during the execution
    """

    def __init__(self,
                 config: Optional[Dict[str, Any]] = None,
                 runtime: Optional[str] = None,
                 runtime_memory: Optional[int] = None,
                 backend: Optional[str] = None,
                 storage: Optional[str] = None,
                 workers: Optional[int] = None,
                 worker_processes: Optional[int] = None,
                 monitoring: Optional[str] = None,
                 remote_invoker: Optional[bool] = None,
                 log_level: Optional[str] = False):

        backend = backend or constants.SERVERLESS_BACKEND_DEFAULT

        super().__init__(config=config,
                         runtime=runtime,
                         runtime_memory=runtime_memory,
                         backend=backend,
                         storage=storage,
                         workers=workers,
                         worker_processes=worker_processes,
                         monitoring=monitoring,
                         log_level=log_level,
                         remote_invoker=remote_invoker)

class StandaloneExecutor(FunctionExecutor):
    """
    Initialize a StandaloneExecutor class.

    :param config: Settings passed in here will override those in config file
    :param runtime: Runtime name to use
    :param backend: Name of the standalone compute backend to use
    :param storage: Name of the storage backend to use
    :param workers: Max number of concurrent workers
    :param worker_processes: Worker granularity, number of concurrent/parallel processes in each worker
    :param monitoring: monitoring system
    :param log_level: log level to use during the execution
    """

    def __init__(self,
                 config: Optional[Dict[str, Any]] = None,
                 runtime: Optional[str] = None,
                 backend: Optional[str] = None,
                 storage: Optional[str] = None,
                 workers: Optional[int] = None,
                 worker_processes: Optional[int] = None,
                 monitoring: Optional[str] = None,
                 log_level: Optional[str] = False):

        backend = backend or constants.STANDALONE_BACKEND_DEFAULT

        super().__init__(config=config,
                         runtime=runtime,
                         backend=backend,
                         storage=storage,
                         workers=workers,
                         worker_processes=worker_processes,
                         monitoring=monitoring,
                         log_level=log_level)

max_stars_repo_name: Cohen-J-Omer/lithops

max_issues_repo_name: Cohen-J-Omer/lithops

reason: import numpy