text:
import numpy as np
from copy import deepcopy
from sklearn.metrics import mean_squared_error
from models.multiple_solution.root_multiple import RootAlgo

class BaseBFO(RootAlgo):
    """
    Basic version of Bacterial Foraging Optimization Algorithm: Taken from here
    http://www.cleveralgorithms.com/nature-inspired/swarm/bfoa.html
    """
    ID_VECTOR = 0
    ID_COST = 1
    ID_INTER = 2
    ID_FITNESS = 3
    ID_SUM_NUTRIENTS = 4

    def __init__(self, root_algo_paras=None, bfo_paras = None):
        RootAlgo.__init__(self, root_algo_paras)
        # algorithm configuration
        self.pop_size = bfo_paras["pop_size"]
        self.step_size = bfo_paras["Ci"]         # Ci
        self.p_eliminate = bfo_paras["Ped"]     # Ped
        self.swim_length = bfo_paras["Ns"]     # Ns

        self.elim_disp_steps = bfo_paras["Ned"]  # Ned
        self.repro_steps = bfo_paras["Nre"]     # Nre
        self.chem_steps = bfo_paras["Nc"]       # Nc

        self.d_attr = bfo_paras["attract_repel"][0]
        self.w_attr = bfo_paras["attract_repel"][1]
        self.h_rep = bfo_paras["attract_repel"][2]
        self.w_rep = bfo_paras["attract_repel"][3]

    def _create_solution__(self, minmax=None):
        vector = np.random.uniform(self.domain_range[0], self.domain_range[1], self.problem_size)
        cost = 0.0
        interaction = 0.0
        fitness = 0.0
        sum_nutrients = 0.0
        return [vector, cost, interaction, fitness, sum_nutrients]

    def _compute_cell_interaction__(self, cell, cells, d, w):
        sum_inter = 0.0
        for other in cells:
            diff = self.problem_size * mean_squared_error(cell[self.ID_VECTOR], other[self.ID_VECTOR])
            sum_inter += d * np.exp(w * diff)
        return sum_inter

    def _attract_repel__(self, cell, cells):
        attract = self._compute_cell_interaction__(cell, cells, -self.d_attr, -self.w_attr)
        repel = self._compute_cell_interaction__(cell, cells, self.h_rep, -self.w_rep)
        return attract + repel

    def _evaluate__(self, cell, cells):
        cell[self.ID_COST] = self._fitness_model__(cell[self.ID_VECTOR])
        cell[self.ID_INTER] = self._attract_repel__(cell, cells)
        cell[self.ID_FITNESS] = cell[self.ID_COST] + cell[self.ID_INTER]

    def _tumble_cell__(self, cell, step_size):
        delta_i = np.random.uniform(self.domain_range[0], self.domain_range[1], self.problem_size)
        unit_vector = delta_i / np.sqrt(np.dot(delta_i, delta_i.T))
        vector = cell[self.ID_VECTOR] + step_size * unit_vector
        return [vector, 0.0, 0.0, 0.0, 0.0]

    def _chemotaxis__(self, l=None, k = None, cells=None):
        current_best = None
        for j in range(0, self.chem_steps):
            moved_cells = []                            # New generation
            for i, cell in enumerate(cells):
                sum_nutrients = 0.0
                self._evaluate__(cell, cells)
                if (current_best is None) or cell[self.ID_COST] < current_best[self.ID_COST]:
                    current_best = deepcopy(cell)
                sum_nutrients += cell[self.ID_FITNESS]

                for m in range(0, self.swim_length):
                    new_cell = self._tumble_cell__(cell, self.step_size)
                    self._evaluate__(new_cell, cells)
                    if current_best[self.ID_COST] > new_cell[self.ID_COST]:
                        current_best = deepcopy(new_cell)
                    if new_cell[self.ID_FITNESS] > cell[self.ID_FITNESS]:
                        break
                    cell = deepcopy(new_cell)
                    sum_nutrients += cell[self.ID_FITNESS]

                cell[self.ID_SUM_NUTRIENTS] = sum_nutrients
                moved_cells.append(deepcopy(cell))
            cells = deepcopy(moved_cells)
            self.loss_train.append(current_best[self.ID_COST])
            if self.print_train:
                print("elim = %d, repro = %d, chemo = %d >> best fitness=%.8f" %(l + 1, k + 1, j + 1, current_best[self.ID_COST]))
        return current_best, cells

    def _train__(self):
        cells = [self._create_solution__(minmax=0) for _ in range(0, self.pop_size)]
        half_pop_size = int(self.pop_size / 2)
        best = None
        for l in range(0, self.elim_disp_steps):
            for k in range(0, self.repro_steps):
                current_best, cells = self._chemotaxis__(l, k, cells)
                if (best is None) or current_best[self.ID_COST] < best[self.ID_COST]:
                    best = current_best
                cells = sorted(cells, key=lambda cell: cell[self.ID_SUM_NUTRIENTS])
                cells = deepcopy(cells[0:half_pop_size]) + deepcopy(cells[0:half_pop_size])
            for idc in range(self.pop_size):
                if np.random.uniform() < self.p_eliminate:
                    cells[idc] = self._create_solution__(minmax=0)
        return best[self.ID_VECTOR], self.loss_train

class ABFOLS(RootAlgo):
    """
    ### This is the best improvement version of BFO
    ## Paper: An Adaptive Bacterial Foraging Optimization Algorithm with Lifecycle and Social Learning
    """
    ID_VECTOR = 0
    ID_FITNESS = 1
    ID_NUTRIENT = 2
    ID_PERSONAL_BEST = 3

    NUMBER_CONTROL_RATE = 2

    def __init__(self, root_algo_paras=None, abfols_paras=None):
        RootAlgo.__init__(self, root_algo_paras)
        # algorithm configuration
        self.epoch = abfols_paras["epoch"]
        self.pop_size = abfols_paras["pop_size"]
        self.step_size = abfols_paras["Ci"]                     # Ci
        self.p_eliminate = abfols_paras["Ped"]                 # Ped
        self.swim_length = abfols_paras["Ns"]                 # Ns

        self.N_adapt = abfols_paras["N_minmax"][0]  # Dead threshold value
        self.N_split = abfols_paras["N_minmax"][1]                          # split threshold value
        self.C_s = self.step_size[0] * (self.domain_range[1] - self.domain_range[0])
        self.C_e = self.step_size[1] * (self.domain_range[1] - self.domain_range[0])

    def _create_solution__(self, minmax=None):
        vector = np.random.uniform(self.domain_range[0], self.domain_range[1], self.problem_size)
        fitness = self._fitness_model__(vector, self.ID_MAX_PROBLEM)          # Current position
        nutrient = 0          # total nutrient gained by the bacterium in its whole searching process.(int number)
        p_best = deepcopy(vector)
        return [vector, fitness, nutrient, p_best]

    def _tumble_cell__(self, cell=None, step_size=None):
        delta_i = (self.global_best[self.ID_VECTOR] - cell[self.ID_VECTOR]) + (cell[self.ID_PERSONAL_BEST] - cell[self.ID_VECTOR])
        if np.all(delta_i == 0):
            delta_i = np.random.uniform(self.domain_range[0], self.domain_range[1], self.problem_size)
        unit_vector = delta_i / np.sqrt(np.dot(delta_i, delta_i.T))

        #unit_vector = np.random.uniform() * 1.2 * (self.global_best[self.ID_VECTOR] - cell[self.ID_VECTOR]) + np.random.uniform() *1.2* (cell[self.ID_PERSONAL_BEST] - cell[self.ID_VECTOR])
        vec = cell[self.ID_VECTOR] + step_size * unit_vector
        fit = self._fitness_model__(vec, self.ID_MAX_PROBLEM)
        if fit> cell[self.ID_FITNESS]:
            cell[self.ID_NUTRIENT] += 1
        else:
            cell[self.ID_NUTRIENT] -= 1

        if fit > cell[self.ID_FITNESS]:             # Update personal best
            cell[self.ID_PERSONAL_BEST] = deepcopy(vec)
        cell[self.ID_VECTOR] = deepcopy(vec)
        cell[self.ID_FITNESS] = fit

        # Update global best
        self.global_best = deepcopy(cell) if self.global_best[self.ID_FITNESS] < cell[self.ID_FITNESS] else self.global_best
        return cell

    def _update_step_size__(self, cells=None, id=None):
        total_fitness = sum(temp[self.ID_FITNESS] for temp in cells)
        step_size = self.C_s - (self.C_s - self.C_e) * cells[id][self.ID_FITNESS]/ total_fitness
        step_size = step_size / cells[id][self.ID_NUTRIENT] if cells[id][self.ID_NUTRIENT] > 0 else step_size
        return step_size

    def _train__(self):
        cells = [self._create_solution__(minmax=0) for _ in range(0, self.pop_size)]
        self.global_best = self._get_global_best__(cells, self.ID_FITNESS, self.ID_MAX_PROBLEM)

        for loop in range(self.epoch):
            i = 0
            while i < len(cells):
                step_size = self._update_step_size__(cells, i)
                JLast = cells[i][self.ID_FITNESS]
                cells[i] = self._tumble_cell__(cell=cells[i], step_size=step_size)
                m = 0
                while m < self.swim_length:             # Ns
                    if cells[i][self.ID_FITNESS] < JLast:
                        step_size = self._update_step_size__(cells, i)
                        JLast = cells[i][self.ID_FITNESS]
                        cells[i] = self._tumble_cell__(cell=cells[i], step_size=step_size)
                        m += 1
                    else:
                        m = self.swim_length

                S_current = len(cells)
                #print("======= Current Nutrient: {}".format(cells[i][self.ID_NUTRIENT]))

                if cells[i][self.ID_NUTRIENT] > max(self.N_split, self.N_split + (S_current - self.pop_size) / self.N_adapt):
                    new_cell = deepcopy(cells[i])
                    new_cell[self.ID_NUTRIENT] = 0
                    cells[i][self.ID_NUTRIENT] = 0
                    cells.append(new_cell)
                    break

                if cells[i][self.ID_NUTRIENT] < min(self.NUMBER_CONTROL_RATE, self.NUMBER_CONTROL_RATE + (S_current - self.pop_size) / self.N_adapt):
                    cells.pop(i)
                    i -= 1
                    break

                if cells[i][self.ID_NUTRIENT] < self.NUMBER_CONTROL_RATE and np.random.uniform() < self.p_eliminate:
                    temp = self._create_solution__(minmax=0)
                    self.global_best = deepcopy(temp) if temp[self.ID_FITNESS] > self.global_best[self.ID_FITNESS] else self.global_best
                    cells[i] = temp
                i += 1
            self.loss_train.append([1.0 / self.global_best[self.ID_FITNESS], 1.0 / self.global_best[self.ID_FITNESS]])
            if self.print_train:
                print("Epoch = {}, Pop_size = {}, >> Best fitness = {}".format(loop + 1, len(cells), 1.0 / self.global_best[self.ID_FITNESS]))

        return self.global_best[self.ID_VECTOR], self.loss_train, self.global_best[self.ID_FITNESS]

max_issues_repo_name: luanedge/metaheuristics

max_forks_repo_name: luanedge/metaheuristics


text:
# source: https://github.com/keras-team/keras/issues/3611

import numpy as np
from keras import backend as K

def dice(y_true, y_pred, smooth=1):
    y_true_f = K.flatten(y_true)
    y_pred_f = K.flatten(y_pred)
    intersection = K.sum(y_true_f * y_pred_f)
    return (2. * intersection + smooth) / (K.sum(y_true_f) + K.sum(y_pred_f) + smooth)

def dice_loss(y_true, y_pred):
    return 1-dice(y_true, y_pred)

def categorical_crossentropy(y_true, y_pred):
    return K.categorical_crossentropy(y_true, y_pred)

def binary_crossentropy(y_true, y_pred):
    return K.mean(K.binary_crossentropy(y_true, y_pred), axis=-1)

def dice_crossentropy(y_true, y_pred):
    return categorical_crossentropy(y_true, y_pred) * dice_loss(y_true, y_pred)

ext: py

max_stars_repo_name: chriamue/Cnet

max_issues_repo_name: chriamue/Cnet

max_forks_repo_name: chriamue/Cnet

reason: import numpy


text:
from __future__ import print_function, division, absolute_import

import numpy as np

def linear_interp_resampler(wave_new, wave, flux, err=None):
    """
    Call interpolation, repackage new spectra

    Parameters
    ----------
    orig_spectrum : `~specutils.Spectrum1D`
        The original 1D spectrum.
    fin_spec_axis : ndarray
        The desired spectral axis array.

    Returns
    -------
    resample_spectrum : `~specutils.Spectrum1D`
        An output spectrum containing the resampled `~specutils.Spectrum1D`
    """

    fill_val = np.nan
    flux_new = np.interp(wave_new, wave, flux, left=fill_val, right=fill_val)

    new_unc = None
    if err is not None:
        err_new = np.interp(
            wave_new,
            wave,
            err,
            left=fill_val,
            right=fill_val,
        )
        return flux_new, err_new

    return flux_new

"""
Vectorized implementation of the Flux Conserving resampler.
The implementation in `desispec` is not vectorized along the sample axis.
Obtained from: https://github.com/ACCarnall/SpectRes/blob/master/spectres/spectral_resampling.py
Implements the equations from: https://ui.adsabs.harvard.edu/abs/2017arXiv170505165C/abstract
Coed still involves a bunch of loops which can be eliminated using more vectorization or JIT
Look at the code from specutils for inspiration
"""

def make_bins(wavs):
    """Given a series of wavelength points, find the edges and widths
    of corresponding wavelength bins."""
    edges = np.zeros(wavs.shape[0] + 1)
    widths = np.zeros(wavs.shape[0])
    edges[0] = wavs[0] - (wavs[1] - wavs[0]) / 2
    widths[-1] = wavs[-1] - wavs[-2]
    edges[-1] = wavs[-1] + (wavs[-1] - wavs[-2]) / 2
    edges[1:-1] = (wavs[1:] + wavs[:-1]) / 2
    widths[:-1] = edges[1:-1] - edges[:-2]

    return edges, widths

def _spectres(
    new_wavs, spec_wavs, spec_fluxes, spec_errs=None, fill=None, verbose=True
):

    """
    Function for resampling spectra (and optionally associated
    uncertainties) onto a new wavelength basis.

    Parameters
    ----------

    new_wavs : numpy.ndarray
        Array containing the new wavelength sampling desired for the
        spectrum or spectra.

    spec_wavs : numpy.ndarray
        1D array containing the current wavelength sampling of the
        spectrum or spectra.

    spec_fluxes : numpy.ndarray
        Array containing spectral fluxes at the wavelengths specified in
        spec_wavs, last dimension must correspond to the shape of
        spec_wavs. Extra dimensions before this may be used to include
        multiple spectra.

    spec_errs : numpy.ndarray (optional)
        Array of the same shape as spec_fluxes containing uncertainties
        associated with each spectral flux value.

    fill : float (optional)
        Where new_wavs extends outside the wavelength range in spec_wavs
        this value will be used as a filler in new_fluxes and new_errs.

    verbose : bool (optional)
        Setting verbose to False will suppress the default warning about
        new_wavs extending outside spec_wavs and "fill" being used.

    Returns
    -------

    new_fluxes : numpy.ndarray
        Array of resampled flux values, first dimension is the same
        length as new_wavs, other dimensions are the same as
        spec_fluxes.

    new_errs : numpy.ndarray
        Array of uncertainties associated with fluxes in new_fluxes.
        Only returned if spec_errs was specified.
    """

    # Rename the input variables for clarity within the function.
    old_wavs = spec_wavs
    old_fluxes = spec_fluxes
    old_errs = spec_errs

    # Make arrays of edge positions and widths for the old and new bins

    old_edges, old_widths = make_bins(old_wavs)
    new_edges, new_widths = make_bins(new_wavs)

    # Generate output arrays to be populated
    new_fluxes = np.zeros(old_fluxes[..., 0].shape + new_wavs.shape)

    if old_errs is not None:
        if old_errs.shape != old_fluxes.shape:
            raise ValueError(
                "If specified, spec_errs must be the same shape " "as spec_fluxes."
            )
        else:
            new_errs = np.copy(new_fluxes)

    start = 0
    stop = 0
    warned = False

    # Calculate new flux and uncertainty values, looping over new bins
    for j in range(new_wavs.shape[0]):

        # Add filler values if new_wavs extends outside of spec_wavs
        if (new_edges[j] < old_edges[0]) or (new_edges[j + 1] > old_edges[-1]):
            new_fluxes[..., j] = fill

            if spec_errs is not None:
                new_errs[..., j] = fill

            if (j == 0 or j == new_wavs.shape[0] - 1) and verbose and not warned:
                warned = True
                print(
                    "\nSpectres: new_wavs contains values outside the range "
                    "in spec_wavs, new_fluxes and new_errs will be filled "
                    "with the value set in the 'fill' keyword argument. \n"
                )
            continue

        # Find first old bin which is partially covered by the new bin
        while old_edges[start + 1] <= new_edges[j]:
            start += 1

        # Find last old bin which is partially covered by the new bin
        while old_edges[stop + 1] < new_edges[j + 1]:
            stop += 1

        # If new bin is fully inside an old bin start and stop are equal
        if stop == start:
            new_fluxes[..., j] = old_fluxes[..., start]
            if old_errs is not None:
                new_errs[..., j] = old_errs[..., start]

        # Otherwise multiply the first and last old bin widths by P_ij
        else:
            start_factor = (old_edges[start + 1] - new_edges[j]) / (
                old_edges[start + 1] - old_edges[start]
            )

            end_factor = (new_edges[j + 1] - old_edges[stop]) / (
                old_edges[stop + 1] - old_edges[stop]
            )

            old_widths[start] *= start_factor
            old_widths[stop] *= end_factor

            # Populate new_fluxes spectrum and uncertainty arrays
            f_widths = old_widths[start : stop + 1] * old_fluxes[..., start : stop + 1]
            new_fluxes[..., j] = np.sum(f_widths, axis=-1)
            new_fluxes[..., j] /= np.sum(old_widths[start : stop + 1])

            if old_errs is not None:
                e_wid = old_widths[start : stop + 1] * old_errs[..., start : stop + 1]

                new_errs[..., j] = np.sqrt(np.sum(e_wid ** 2, axis=-1))
                new_errs[..., j] /= np.sum(old_widths[start : stop + 1])

            # Put back the old bin widths to their initial values
            old_widths[start] /= start_factor
            old_widths[stop] /= end_factor

    # If errors were supplied return both new_fluxes and new_errs.
    if old_errs is not None:
        return new_fluxes, new_errs

    # Otherwise just return the new_fluxes spectrum array
    else:
        return new_fluxes

def resample(
    wave_new, wave, flux, ivar=None, fill=None, method="linear", verbose=False
):

    """
    Function for resampling spectra (and optionally associated
    uncertainties) onto a new wavelength basis.

    Parameters
    ----------

    wave_new : numpy.ndarray
        Array containing the new wavelength sampling desired for the
        spectrum or spectra.

    wave : numpy.ndarray
        array containing the current wavelength sampling of the
        spectrum or spectra.

    flux : numpy.ndarray
        Array containing spectral fluxes at the wavelengths specified in
        spec_wavs, last dimension must correspond to the shape of
        spec_wavs. Extra dimensions before this may be used to include
        multiple spectra.

    ivar : numpy.ndarray (optional)
        Array of the same shape as spec_fluxes containing uncertainties
        associated with each spectral flux value.

    fill : float (optional)
        Where new_wavs extends outside the wavelength range in spec_wavs
        this value will be used as a filler in new_fluxes and new_errs.
    method : str (optional)
        choose from `linear` or `flux_cons`.
        Use a linear resampled spectra or flux conserving resampler.
    verbose : bool (optional)
        Setting verbose to False will suppress the default warning about
        new_wavs extending outside spec_wavs and "fill" being used.

    Returns
    -------

    flux_new : numpy.ndarray
        Array of resampled flux values, first dimension is the same
        length as new_wavs, other dimensions are the same as
        spec_fluxes.

    ivar_new : numpy.ndarray
        Array of uncertainties associated with fluxes in new_fluxes.
        Only returned if spec_errs was specified.
    New bins which include missing data are denoted by nan.
    """
    num_spec = len(flux)
    # TODO depending on num_spec choose whether to JIT or not
    if wave_new.ndim == 1:
        wave_new = np.repeat(wave_new[np.newaxis, :], num_spec, axis=0)
    if wave.ndim == 1:
        wave = np.repeat(wave[np.newaxis, :], num_spec, axis=0)

    if ivar is not None:
        ivar = ivar.copy()
        flux = flux.copy()
        flux[ivar == 0] = np.nan
        ivar[ivar == 0] = np.nan
        flux_new = np.zeros_like(wave_new)
        err_new = np.zeros_like(wave_new)
        err = np.sqrt(1 / ivar)
        # for i in range(num_spec):
        #     flux_new[i], err_new[i] = _spectres(
        #         wave_new[i], wave[i], flux[i], err[i], fill=fill, verbose=verbose
        #     )
        for i in range(num_spec):
            flux_new[i], err_new[i] = linear_interp_resampler(
                wave_new[i], wave[i], flux[i], err[i]
            )
        ivar_new = 1 / err_new ** 2
        # ivar_new[np.isnan(ivar_new)] == 0
        return flux_new, ivar_new

    # Otherwise just return the new_fluxes spectrum array
    else:
        flux_new = np.zeros_like(wave_new)
        for i in range(num_spec):
            flux_new[i] = _spectres(
                wave_new[i],
                wave[i],
                flux[i],
                spec_errs=None,
                fill=fill,
                verbose=verbose,
            )
        return flux_new

max_stars_repo_name: biprateep/DESI-stack

max_issues_repo_name: biprateep/DESI-stack

max_forks_repo_name: biprateep/DESI-stack


text:
import numpy as np
import seaborn as sns
import matplotlib.pylab as plt
import sys

N = sys.argv[1]
data = np.random.rand(N)
ax = sns.heatmap(data, linewidth=0.5)
plt.show()


text:
# -*- coding: utf-8 -*-

"""
@author: Quoc-Tuan Truong <tuantq.vnu@gmail.com>
"""

import numpy as np

def fallback_feature(func):
    """Decorator to fallback to `batch_feature` in FeatureModule
    """
    def wrapper(self, *args, **kwargs):
        if self.features is not None:
            ids = args[0] if len(args) > 0 else kwargs['batch_ids']
            return FeatureModule.batch_feature(self, batch_ids=ids)
        else:
            return func(self, *args, **kwargs)

    return wrapper

class Module:
    """Module
    """

    def __init__(self, **kwargs):
        pass

class FeatureModule(Module):
    """FeatureModule

    Parameters
    ----------
    features: numpy.ndarray or scipy.sparse.csr_matrix, default = None
        Numpy 2d-array that the row indices are aligned with user/item in `ids`.

    ids: List, default = None
        List of user/item ids that the indices are aligned with `corpus`.
        If None, the indices of provided `features` will be used as `ids`.
    """

    def __init__(self, features=None, ids=None, normalized=False, **kwargs):
        super().__init__(**kwargs)
        self.features = features
        self.ids = ids
        self._normalized = normalized

    @property
    def features(self):
        """Return the whole feature matrix
        """
        return self.__features

    @features.setter
    def features(self, input_features):
        if input_features is not None:
            assert len(input_features.shape) == 2
        self.__features = input_features

    @property
    def feature_dim(self):
        """Return the dimensionality of the feature vectors
        """
        return self.features.shape[1]

    def _swap_feature(self, id_map):
        if self.ids is None:
            self.ids = np.arange(self.features.shape[0])

        for old_idx, raw_id in enumerate(self.ids):
            new_idx = id_map.get(raw_id, None)
            if new_idx is None:
                continue
            assert new_idx < self.features.shape[0]
            self.features[[new_idx, old_idx]] = self.features[[old_idx, new_idx]]

    def build(self, id_map=None):
        """Build the feature matrix.
        Features will be swapped if the id_map is provided
        """
        if self.features is None:
            return

        if id_map is not None:
            self._swap_feature(id_map)

        if self._normalized:
            self.features = self.features - np.min(self.features)
            self.features = self.features / (np.max(self.features) + 1e-10)

    def batch_feature(self, batch_ids):
        """Return a matrix (batch of feature vectors) corresponding to provided batch_ids
        """
        assert self.features is not None
        return self.features[batch_ids]

max_stars_repo_name: GuoJingyao/cornac

max_issues_repo_name: GuoJingyao/cornac

max_forks_repo_name: GuoJingyao/cornac


text:
import re
from collections import OrderedDict
from functools import partial
import itertools

from six import string_types
from tabulate import tabulate
import numpy as np
from artemis.experiments.experiment_record import NoSavedResultError, ExpInfoFields, ExperimentRecord, \
    load_experiment_record, is_matplotlib_imported, UnPicklableArg
from artemis.general.display import deepstr, truncate_string, hold_numpy_printoptions, side_by_side, \
    surround_with_header, section_with_header, dict_to_str
from artemis.general.duck import Duck
from artemis.general.nested_structures import flatten_struct, PRIMATIVE_TYPES
from artemis.general.should_be_builtins import separate_common_items, bad_value, izip_equal, \
    remove_duplicates, get_unique_name, entries_to_table
from artemis.general.tables import build_table
import os

from artemis.plotting.parallel_coords_plots import plot_hyperparameter_search_parallel_coords
from artemis.plotting.pyplot_plus import get_color_cycle_map

def get_record_result_string(record, func='deep', truncate_to = None, array_print_threshold=8, array_float_format='.3g', oneline=False, default_one_liner_func=str):
    """
    Get a string representing the result of the experiment.
    :param record:
    :param func:
    :return:
    """
    with hold_numpy_printoptions(threshold = array_print_threshold, formatter={'float': lambda x: '{{:{}}}'.format(array_float_format).format(x)}):
        if isinstance(func, string_types):
            func = {
                'deep': deepstr,
                'str': str,
                }[func]
        else:
            assert callable(func), 'func must be callable'
        try:
            result = record.get_result()
        except NoSavedResultError:
            return '<No result has been saved>'
        string = func(result)
        if not isinstance(string, string_types):
            string = default_one_liner_func(string)

    if truncate_to is not None:
        string = truncate_string(string, truncation=truncate_to, message = '...<truncated>')
    if oneline:
        string = string.replace('\n', ', ')
    return string

def _get_record_info_section(record, header_width):
    return section_with_header('Info', record.info.get_text(), width=header_width)

def _get_record_log_section(record, truncate_logs = None, header_width=64):
    log = record.get_log()
    if truncate_logs is not None and len(log)>truncate_logs:
        log = log[:truncate_logs-100] + '\n\n ... LOG TRUNCATED TO {} CHARACTERS ... \n\n'.format(truncate_logs) + log[-100:]
    return section_with_header('Logs', log, width=header_width)

def _get_record_error_trace_section(record, header_width):
    error_trace = record.get_error_trace()
    if error_trace is None:
        return ''
    else:
        return section_with_header('Error Trace', record.get_error_trace(), width=header_width)

def _get_result_section(record, truncate_result, show_result, header_width):
    assert show_result in (False, 'full', 'deep')
    result_str = get_record_result_string(record, truncate_to=truncate_result, func=show_result)
    return section_with_header('Result', result_str, width=header_width)

def get_record_full_string(record, show_info = True, show_logs = True, truncate_logs = None, show_result ='deep',
        show_exceptions=True, truncate_result = None, include_bottom_border = True, header_width=64, return_list = False):
    """
    Get a human-readable string containing info about the experiment record.

    :param show_info: Show info about experiment (name, id, runtime, etc)
    :param show_logs: Show logs (True, False, or an integer character count to truncate logs at)
    :param show_result: Show the result.  Options for display are:
        'short': Print a one-liner (note: sometimes prints multiple lines)
        'long': Directly call str
        'deep': Use the deepstr function for a compact nested printout.
    :return: A string to print.
    """

    parts = [surround_with_header(record.get_id(), width=header_width, char='=')]

    if show_info:
        parts.append(section_with_header('Info', record.info.get_text(), width=header_width))

    if show_logs:
        log = record.get_log()
        if truncate_logs is not None and len(log)>truncate_logs:
            log = log[:truncate_logs-100] + '\n\n ... LOG TRUNCATED TO {} CHARACTERS ... \n\n'.format(truncate_logs) + log[-100:]
        # return section_with_header('Logs', log, width=header_width)
        parts.append(section_with_header('Logs', log, width=header_width))

    if show_exceptions:
        error_trace = record.get_error_trace()
        error_trace_text = '' if error_trace is None else section_with_header('Error Trace', record.get_error_trace(), width=header_width)
        parts.append(error_trace_text)

    if show_result:
        assert show_result in (False, 'full', 'deep')
        result_str = get_record_result_string(record, truncate_to=truncate_result, func=show_result)
        parts.append(section_with_header('Result', result_str, width=header_width))

    if return_list:
        return parts
    else:
        return '\n'.join(parts)

def get_record_invalid_arg_string(record, recursive=False, ignore_valid_keys=(), note_version = 'full'):
    """
    Return a string identifying ig the arguments for this experiment are still valid.
    :return:
    """
    from artemis.experiments.experiments import is_experiment_loadable
    assert note_version in ('full', 'short')
    experiment_id = record.get_experiment_id()
    if is_experiment_loadable(experiment_id):
        if record.info.has_field(ExpInfoFields.ARGS):
            last_run_args = OrderedDict([(k,v) for k,v in record.get_args().items() if k not in ignore_valid_keys])
            current_args = OrderedDict([(k,v) for k,v in record.get_experiment().get_args().items() if k not in ignore_valid_keys])
            if recursive:
                last_run_args = OrderedDict(flatten_struct(last_run_args, first_dict_is_namespace=True, primatives = PRIMATIVE_TYPES + (UnPicklableArg, )))
                last_run_args = OrderedDict([(k, v) for k, v in last_run_args.items() if k not in ignore_valid_keys])
                current_args = OrderedDict(flatten_struct(current_args, first_dict_is_namespace=True))
                current_args = OrderedDict([(k, v) for k, v in current_args.items() if k not in ignore_valid_keys])

            validity = record.args_valid(last_run_args=last_run_args, current_args=current_args)
            if validity is False:
                common, (old_args, new_args) = separate_common_items([list(last_run_args.items()), list(current_args.items())])
                if len(old_args)+len(new_args)==0:
                    raise Exception('Error displaying different args.  Bug Peter.')

                all_changed_arg_names = remove_duplicates(list(name for name, _ in old_args)+list(name for name, _ in new_args))
                changestr = ', '.join("{}:{}->{}".format(k, last_run_args[k] if k in last_run_args else '<N/A>', current_args[k] if k in current_args else '<N/A>') for k in all_changed_arg_names)
                notes = ("Change: " if note_version=='full' else "") + changestr
            elif validity is None:
                notes = "Cannot Determine: Unhashable Args" if note_version=='full' else '<Unhashable Args>'
            else:
                notes = "<No Change>"
        else:
            notes = "Cannot Determine: Inconsistent Experiment Record" if note_version == 'full' else '<Inconsistent Record>'
    else:
        notes = "Cannot Determine: Experiment Not Imported" if note_version=='full' else '<Not Imported>'
    return notes

def get_oneline_result_string(record, truncate_to=None, array_float_format='.3g', array_print_threshold=8, default_one_liner_func=dict_to_str):
    """
    Get a string that describes the result of the record in one line.  This can optionally be specified by
    experiment.one_liner_function.

    :param record: An ExperimentRecord.
    :param truncate_to:
    :param array_float_format:
    :param array_print_threshold:
    :return: A string with no newlines briefly describing the result of the record.
    """
    from artemis.experiments.experiments import is_experiment_loadable
    if isinstance(record, string_types):
        record = load_experiment_record(record)
    if not is_experiment_loadable(record.get_experiment_id()):
        one_liner_function=default_one_liner_func
    else:
        one_liner_function = record.get_experiment().one_liner_function
        if one_liner_function is None:
            one_liner_function = default_one_liner_func
    return get_record_result_string(record, func=one_liner_function, truncate_to=truncate_to, array_print_threshold=array_print_threshold,
        array_float_format=array_float_format, oneline=True, default_one_liner_func=default_one_liner_func)

def print_experiment_record_argtable(records):
    """
    Print a table comparing experiment arguments and their results.
    """
    funtion_names = [record.info.get_field(ExpInfoFields.FUNCTION) for record in records]
    args = [record.get_args() for record in records]
    common_args, different_args = separate_common_items(args)
    record_ids = [record.get_id() for record in records]
    def lookup_fcn(record_id, column):
        index = record_ids.index(record_id)
        if column=='Function':
            return funtion_names[index]
        elif column=='Run Time':
            return records[index].info.get_field_text(ExpInfoFields.RUNTIME)
        elif column=='Common Args':
            return ', '.join('{}={}'.format(k, v) for k, v in common_args.items())
        elif column=='Different Args':
            return ', '.join('{}={}'.format(k, v) for k, v in different_args[index].items())
        elif column=='Result':
            return get_oneline_result_string(records[index])
        else:
            bad_value(column)

    rows = build_table(lookup_fcn,
        row_categories=record_ids,
        column_categories=['Function', 'Run Time', 'Common Args', 'Different Args', 'Result'],
        prettify_labels=False
        )

    print(tabulate(rows))

def get_column_change_ordering(tabular_data):
    """
    Get the order in which to rearrange the columns so that the fastest-changing data comes last.

    :param tabular_data: A list of equal-length lists
    :return: A set of permutation indices for the columns.
    """
    n_rows, n_columns = len(tabular_data), len(tabular_data[0])
    deltas = [sum(row_prev[i]!=row[i] for row_prev, row in zip(tabular_data[:-1], tabular_data[1:])) for i in range(n_columns)]
    return np.argsort(deltas)

def get_different_args(args, no_arg_filler = 'N/A', arrange_by_deltas=False):
    """
    Get a table of different args between records.
    :param Sequence[List[Tuple[str, Any]]] args: A list of lists of argument (name, value) pairs.
    :param no_arg_filler: The filler value to use if a record does not have a particular argument (possibly due to an argument being added to the code after the record was made)
    :param arrange_by_deltas: If true, order arguments so that the fastest-changing ones are in the last column
    :return Tuple[List[str], List[List[Any]]]: (arg_names, arg_values) where:
        arg_names is a list of arguments that differ between records
        arg_values is a len(records)-list of len(arg_names) lists of values of the arguments for each record.
    """
    args = list(args)
    common_args, different_args = separate_common_items(args)
    all_different_args = list(remove_duplicates((k for dargs in different_args for k in dargs.keys())))
    values = [[record_args[argname] if argname in record_args else no_arg_filler for argname in all_different_args] for record_args in args]
    if arrange_by_deltas:
        col_shuf_ixs = get_column_change_ordering(values)
        all_different_args = [all_different_args[i] for i in col_shuf_ixs]
        values = [[row[i] for i in col_shuf_ixs] for row in values]
    return all_different_args, values

def get_exportiment_record_arg_result_table(records, result_parser = None, fill_value='N/A', arg_rename_dict = None):
    """
    Given a list of ExperimentRecords, make a table containing the arguments that differ between them, and their results.
    :param Sequence[ExperimentRecord] records:
    :param Optional[Callable] result_parser: Takes the result and returns either:
        - a List[Tuple[str, Any]], containing the (name, value) pairs of results which will form the rightmost columns of the table
        - Anything else, in which case the header of the last column is taken to be "Result" and the value is put in the table
    :param fill_value: Value to fill in when the experiment does not have a particular argument.
    :return Tuple[List[str], List[List[Any]]]: headers, results
    """
    if arg_rename_dict is not None:
        arg_processor = lambda args: OrderedDict((arg_rename_dict[name] if name in arg_rename_dict else name, val) for name, val in args.items() if name not in arg_rename_dict or arg_rename_dict[name] is not None)
    else:
        arg_processor = lambda args: args
    record_ids = [record.get_id() for record in records]
    all_different_args, arg_values = get_different_args([arg_processor(r.get_args()) for r in records], no_arg_filler=fill_value)

    parsed_results = [(result_parser or record.get_experiment().result_parser)(record.get_result()) if record.has_result() else [('Result', 'N/A')] for record in records]
    result_fields, result_data = entries_to_table(parsed_results, fill_value = fill_value)
    result_fields = [get_unique_name(rf, all_different_args) for rf in result_fields]  # Just avoid name collisions

    # result_column_name = get_unique_name('Results', taken_names=all_different_args)

    def lookup_fcn(record_id, arg_or_result_name):
        row_index = record_ids.index(record_id)
        if arg_or_result_name in result_fields:
            return result_data[row_index][result_fields.index(arg_or_result_name)]
        else:
            column_index = all_different_args.index(arg_or_result_name)
            return arg_values[row_index][column_index]

    rows = build_table(lookup_fcn,
        row_categories=record_ids,
        column_categories=all_different_args + result_fields,
        prettify_labels=False,
        include_row_category=False,
        )

    return rows[0], rows[1:]

def show_record(record, show_logs=True, truncate_logs=None, truncate_result=10000, header_width=100, show_result ='deep', hang=True):
    """
    Show the results of an experiment record.
    :param record:
    :param show_logs:
    :param truncate_logs:
    :param truncate_result:
    :param header_width:
    :param show_result:
    :param hang:
    :return:
    """
    string = get_record_full_string(record, show_logs=show_logs, show_result=show_result, truncate_logs=truncate_logs,
        truncate_result=truncate_result, header_width=header_width, include_bottom_border=False)

    has_matplotlib_figures = any(loc.endswith('.pkl') for loc in record.get_figure_locs())
    if has_matplotlib_figures:
        record.show_figures(hang=hang)
    print(string)

def show_multiple_records(records, func = None):

    if func is None:
        func = lambda rec: rec.get_experiment().show(rec)

    if is_matplotlib_imported():
        from artemis.plotting.manage_plotting import delay_show
        with delay_show():
            for rec in records:

                func(rec)
    else:
        for rec in records:
            func(rec)

def compare_experiment_records(records, parallel_text=None, show_logs=True, truncate_logs=None,
        truncate_result=10000, header_width=100, max_linewidth=128, show_result ='deep'):
    """
    Show the console logs, figures, and results of a collection of experiments.

    :param records:
    :param parallel_text:
    :param hang_notice:
    :return:
    """
    if isinstance(records, ExperimentRecord):
        records = [records]
    if parallel_text is None:
        parallel_text = len(records)>1
    if len(records)==0:
        print('... No records to show ...')
        return
    else:
        records_sections = [get_record_full_string(rec, show_logs=show_logs, show_result=show_result, truncate_logs=truncate_logs,
                    truncate_result=truncate_result, header_width=header_width, include_bottom_border=False, return_list=True) for rec in records]

    if parallel_text:
        full_string = '\n'.join(side_by_side(records_section, max_linewidth=max_linewidth) for records_section in zip(*records_sections))
    else:
        full_string = '\n'.join('\n'.join(record_sections) for record_sections in records_sections)

    print(full_string)

    has_matplotlib_figures = any(loc.endswith('.pkl') for rec in records for loc in rec.get_figure_locs())
    if has_matplotlib_figures:
        from artemis.plotting.manage_plotting import delay_show
        with delay_show():
            for rec in records:
                rec.show_figures()

    return has_matplotlib_figures

def make_record_comparison_duck(records, only_different_args = False, results_extractor = None):
    """
    Make a data structure containing arguments and results of the experiment.
    :param Sequence[ExperimentRecord] records:
    :param Optional[Callable] results_extractor:
    :return Duck: A Duck with one entry per record.  Each entry has keys ['args', 'result']
    """
    duck = Duck()

    if only_different_args:
        common, diff = separate_common_args(records)
    else:
        common = None

    for rec in records:
        duck[next, 'args', :] = rec.get_args() if common is None else OrderedDict((k, v) for k, v in rec.get_args().items() if k not in common)
        result = rec.get_result()
        if results_extractor is not None:
            result = results_extractor(result)
        duck[-1, 'exp_id'] = rec.get_experiment_id()
        duck[-1, 'id'] = rec.get_id()
        duck[-1, 'result', ...] = result
    return duck

def make_record_comparison_table(records, args_to_show=None, results_extractor = None, print_table = False, tablefmt='simple', reorder_by_args=False):
    """
    Make a table comparing the arguments and results of different experiment records.  You can use the output
    of this function with the tabulate package to make a nice readable table.

    :param records: A list of records whose results to compare
    :param args_to_show: A list of arguments to show.  If none, it will just show all arguments
        that differ between experiments.
    :param results_extractor: A dict<str->callable> where the callables take the result of the
        experiment as an argument and return an entry in the table.
    :param print_table: Optionally, import tabulate and print the table here and now.
    :return: headers, rows
        headers is a list of of headers for the top of the table
        rows is a list of lists filling in the information.

    example usage:

        headers, rows = make_record_comparison_table(
            record_ids = [experiment_id_to_latest_record_id(eid) for eid in [
                'demo_fast_weight_mlp.multilayer_baseline.1epoch.version=mlp',
                'demo_fast_weight_mlp.multilayer_baseline.1epoch.full-gd.n_steps=1',
                'demo_fast_weight_mlp.multilayer_baseline.1epoch.full-gd.n_steps=20',
                ]],
            results_extractor={
                'Test': lambda result: result.get_best('test').score.get_score('test'),
                'Train': lambda result: result.get_best('test').score.get_score('train'),
                }
             )
        import tabulate
        print tabulate.tabulate(rows, headers=headers, tablefmt=tablefmt)
    """

    args = [rec.get_args(ExpInfoFields.ARGS) for rec in records]
    if args_to_show is None:
        common, separate = separate_common_items(args)
        args_to_show = [k for k, v in separate[0]]

    if results_extractor is None:
        results_extractor = {'Result': str}
    elif callable(results_extractor):
        results_extractor = {'Result': results_extractor}
    else:
        assert isinstance(results_extractor, dict)

    headers = args_to_show + results_extractor.keys()

    rows = []
    for record, record_args in izip_equal(records, args):
        arg_dict = dict(record_args)
        args_vals = [arg_dict[k] for k in args_to_show]
        results = record.get_result()
        rows.append(args_vals+[f(results) for f in results_extractor.values()])

    if reorder_by_args:
        rows = sorted(rows)

    if print_table:
        import tabulate
        print(tabulate.tabulate(rows, headers=headers, tablefmt=tablefmt))
    return headers, rows

def separate_common_args(records, as_dicts=False, return_dict = False, only_shared_argdiffs = False):
    """

    :param records: A List of records
    :param return_dict: Return the different args as a dict<ExperimentRecord: args>
    :return Tuple[OrderedDict[str, Any], List[OrderedDict[str, Any]]: (common, different)
        Where common is an OrderedDict of common args
        different is a list (the same lengths of records) of OrderedDicts containing args that are not the same in all records.
    """
    common, argdiff = separate_common_items([list(rec.get_args().items()) for rec in records])
    if only_shared_argdiffs:
        args_that_they_all_have = set.intersection(*(set(k for k, v in different) for different in argdiff))
        argdiff = [[(k, v) for k, v in ad if k in args_that_they_all_have] for ad in argdiff]

    if as_dicts:
        common = OrderedDict(common)
        argdiff = [OrderedDict(ad) for ad in argdiff]

    if return_dict:
        argdiff = {rec.get_id(): args for rec, args in zip(records, argdiff)}
    return common, argdiff

def compare_timeseries_records(records, yfield, xfield = None, hang=True, ax=None):
    """
    :param Sequence[ExperimentRecord] records: A list of records containing results of the form
        Sequence[Dict[str, number]]
    :param yfield: The name of the field for the x-axis
    :param xfield: The name of the field(s) for the y-axis
    """
    from matplotlib import pyplot as plt
    results = [rec.get_result() for rec in records]
    all_different_args, values = get_different_args([r.get_args() for r in records])

    if not isinstance(yfield, (list, tuple)):
        yfield = [yfield]

    ax = ax if ax is not None else plt.figure().add_subplot(1, 1, 1)
    for result, argvals in izip_equal(results, values):
        xvals = [r[xfield] for r in result] if xfield is not None else list(range(len(result)))
        # yvals = [r[yfield[0]] for r in result]
        h, = ax.plot(xvals, [r[yfield[0]] for r in result], label=(yfield[0]+': ' if len(yfield)>1 else '')+', '.join('{}={}'.format(argname, argval) for argname, argval in izip_equal(all_different_args, argvals)))
        for yf, linestyle in zip(yfield[1:], itertools.cycle(['--', ':', '-.'])):
            ax.plot(xvals, [r[yf] for r in result], linestyle=linestyle, color=h.get_color(), label=yf+': '+', '.join('{}={}'.format(argname, argval) for argname, argval in izip_equal(all_different_args, argvals)))

    ax.grid(True)
    if xfield is not None:
        ax.set_xlabel(xfield)
    if len(yfield)==1:
        ax.set_ylabel(yfield[0])
    plt.legend()
    if hang:
        plt.show()

def get_timeseries_record_comparison_function(yfield, xfield = None, hang=True, ax=None):
    """
    :param yfield: The name of the field for the x-axis
    :param xfield: The name of the field(s) for the y-axis
    """
    return lambda records: compare_timeseries_records(records, yfield, xfield = xfield, hang=hang, ax=ax)

def timeseries_oneliner_function(result, fields, show_len, show = 'last'):
    assert show=='last', 'Only support showing last element now'
    return ('{} items.  '.format(len(result)) if show_len else '')+', '.join('{}: {:.3g}'.format(k, result[-1][k]) if isinstance(result[-1][k], float) else '{}: {}'.format(k, result[-1][k]) for k in fields)

def get_timeseries_oneliner_function(fields, show_len=False, show='last'):
    return partial(timeseries_oneliner_function, fields=fields, show_len=show_len, show=show)

def browse_record_figs(record):
    """
    Browse through the figures associated with an experiment record
    :param ExperimentRecord record: An experiment record
    """
    # TODO: Generalize this to just browse through the figures in a directory.

    from artemis.plotting.saving_plots import interactive_matplotlib_context
    import pickle
    from matplotlib import pyplot as plt
    from artemis.plotting.drawing_plots import redraw_figure
    fig_locs = record.get_figure_locs()

    class nonlocals:
        this_fig = None
        figno = 0

    def show_figure(ix):
        path = fig_locs[ix]
        dir, name = os.path.split(path)
        if nonlocals.this_fig is not None:
            plt.close(nonlocals.this_fig)
        plt.close(plt.gcf())
        with open(path, "rb") as f:
            fig = pickle.load(f)
            fig.canvas.set_window_title(record.get_id()+': ' +name+': (Figure {}/{})'.format(ix+1, len(fig_locs)))
            fig.canvas.mpl_connect('key_press_event', changefig)
        print('Showing {}: Figure {}/{}.  Full path: {}'.format(name, ix+1, len(fig_locs), path))
        plt.show()
        nonlocals.this_fig = plt.gcf()

    def changefig(keyevent):
        if keyevent.key=='right':
            nonlocals.figno = (nonlocals.figno+1)%len(fig_locs)
        elif keyevent.key=='left':
            nonlocals.figno = (nonlocals.figno-1)%len(fig_locs)
        elif keyevent.key=='up':
            nonlocals.figno = (nonlocals.figno-10)%len(fig_locs)
        elif keyevent.key=='down':
            nonlocals.figno = (nonlocals.figno+10)%len(fig_locs)

        elif keyevent.key==' ':
            nonlocals.figno = queryfig()
        else:
            print("No handler for key: {}.  Changing Nothing".format(keyevent.key))
        show_figure(nonlocals.figno)

    def queryfig():
        user_input = input('Which Figure (of 1-{})?  >>'.format(len(fig_locs)))
        try:
            nonlocals.figno = int(user_input)-1
        except:
            if user_input=='q':
                raise Exception('Quit')
            else:
                print("No handler for input '{}'".format(user_input))
        return nonlocals.figno

    print('Use Left/Right arrows to navigate, ')
    show_figure(nonlocals.figno)

def plot_hyperparameter_search(record, relabel = None, assert_all_relabels_used = True, **hypersearch_parallel_kwargs):
    """
    Create a parallel coordinates plot representing a hyperparameter search experiment record.
    :param ExperimentRecord record: An experiment record object
    :param hypersearch_parallel_kwargs: See plot_hyperparameter_search_parallel_coords
    :return: A bunch of plot handels
    """
    result = record.get_result()
    assert {'names', 'x_iters', 'func_vals', 'x'}.issubset(result.keys()), "Record {} does not appear to be from a Parameter Search experiment!".format(record)
    names, x_iters, func_vals, x = result['names'], result['x_iters'], result['func_vals'], result['x']
    if relabel is not None:
        if assert_all_relabels_used:
            assert set(relabel.keys()).issubset(names), 'Not all relabeling keys {} were found in names {}'.format(list(relabel.keys()), list(names))
        names = [relabel[n] if n in relabel else n for n in names]
    return plot_hyperparameter_search_parallel_coords(field_names=list(names), param_sequence=x_iters, func_vals=func_vals, final_params=x, **hypersearch_parallel_kwargs)

max_stars_repo_name: peteroconnor-bc/artemis

max_issues_repo_name: peteroconnor-bc/artemis

max_forks_repo_name: peteroconnor-bc/artemis