text:
from sklearn.neighbors import KNeighborsClassifier
from sklearn.metrics import accuracy_score, confusion_matrix, classification_report
import numpy as np
import joblib
import os
from utils import make_train_data, make_test_data, save_txt

EXPERIMENT_NAME = "experiment1"

FEATURES = ["ar_16","ar_24","dwt","dwt_stat","welch_16","welch_32","welch_64"]
K_VALUES = [5,7,11,14,17]

SKIP_COMBINATIONS = set([])

SAVE_RESULTS      = True  # saves results in single file using joblib library
SAVE_RESULTS_TXT  = True  # saves results in .txt file
SAVE_MODEL        = False # saves trained model

def experiment_knn(train_data,train_data_classes,test_data,test_data_classes,k):
	try:
		knn = KNeighborsClassifier(n_neighbors=k)
		knn.fit(train_data,train_data_classes)

		results = knn.predict(test_data)
		score = accuracy_score(test_data_classes,results)
		report = classification_report(test_data_classes,results,digits=4,output_dict=False)
		report_dict = classification_report(test_data_classes,results,output_dict=True)
		cm = confusion_matrix(test_data_classes,results)

		res = {}
		res['results'] = results
		res['accuracy_score'] = score
		res['classification_report'] = report
		res['classification_report_dict'] = report_dict
		res['confusion_matrix'] = cm

		return res,knn

	except Exception as e:
		print(e)
		return None,None

def main():
	if(EXPERIMENT_NAME not in os.listdir()):
		os.mkdir(EXPERIMENT_NAME)

	for feature in FEATURES:
		try:
			data = np.load(feature + "_stats.npy",allow_pickle=True).item()
			pca = joblib.load("pca_" + feature + "_stats")
			train_data, train_data_classes = make_train_data(data,True)
			test_data, test_data_classes = make_test_data(data)
			train_data_pca = np.array(pca.transform(train_data))
			test_data_pca = np.array(pca.transform(test_data))

			for k in K_VALUES:
				if (feature,k) in SKIP_COMBINATIONS:
					print("Skipping " + feature  + " " + str(k) + "-NN")
					continue

				print("Computing " + feature  + " " + str(k) + "-NN")
				res, model = experiment_knn(train_data_pca,train_data_classes,test_data_pca,test_data_classes,k)

				if res != None:
					if SAVE_RESULTS:
						filename = EXPERIMENT_NAME + "_" + feature + "_" + str(k) + "nn_results"
						path = os.path.join(EXPERIMENT_NAME,filename)
						joblib.dump(res,path)

					if SAVE_RESULTS_TXT:
						filename = EXPERIMENT_NAME + "_" + feature + "_" + str(k) + "nn_results.txt"
						path = os.path.join(EXPERIMENT_NAME,filename)
						save_txt(res,path)

					if SAVE_MODEL:
						filename = EXPERIMENT_NAME + "_" + feature + "_" + str(k) + "nn_model"
						path = os.path.join(EXPERIMENT_NAME,filename)
						joblib.dump(model,path)

		except Exception as e:
			print("Error during " + EXPERIMENT_NAME + " " + feature + " " + str(k) + "-NN")
			print(e)
			pass

if __name__ == "__main__":
	main()

ext: py

max_stars_repo_name: jbrowarczyk/jb-masters-thesis

max_issues_repo_name: jbrowarczyk/jb-masters-thesis

max_forks_repo_name: jbrowarczyk/jb-masters-thesis

reason: import numpy


text:
import numpy as np
import scipy.sparse as sp
import scipy.sparse.linalg

class Variable(object):
    def __init__(self, *args):
        if len(args) == 1 and type(args[0]) in [np.int64,int]:
            args = [(args[0],)]
        if type(args[0][0]) is not str:
            args = list([('v%d'%i, a) for i,a in enumerate(args)])
        self._descr = {}
        self._args = args
        self.size = 0
        for name, dim in args:
            self._descr[name] = {
                'name': name,
                'offset': self.size,
                'shape': dim,
                'size': np.prod(dim)
            }
            self.size += np.prod(dim)

    def vars(self, data=None, named=False):
        if data is None:
            if named:
                return self._descr
            else:
                return [self._descr[a[0]] for a in self._args]

        result = {} if named else [None for a in self._args]
        for i,a in enumerate(self._args):
            d = self._descr[a[0]]
            start, end = d['offset'], d['offset']+np.prod(d['shape'])
            idx = d['name'] if named else i
            result[idx] = data[start:end].reshape(d['shape'])
        return result

    def __getitem__(self, idx):
        if type(idx) is str:
            return self._descr[idx]
        else:
            return self._descr[self._args[idx][0]]

    def new(self, dtype=np.float64):
        return np.zeros((self.size,), order='C', dtype=dtype)

class Operator(object):
    """ Representation of a mathematical operator T """
    def __init__(self):
        self.x = None # Variable, input (domain)
        self.y = None # Variable, output (range)

    def prepare_gpu(self, type_t="double"):
        # prepare/compile gpu kernels if necessary
        pass

    def _call_gpu(self, x, y=None, add=False, jacobian=False):
        raise Exception("This operator does not support GPU arrays")

    def _call_cpu(self, x, y=None, add=False, jacobian=False):
        raise Exception("This operator does not support numpy arrays")

    def __call__(self, x, y=None, add=False, jacobian=False):
        """ Evaluate T(x)

        The result is written to x by default.

        Args:
            x : array (either numpy or gpuarray)
            y : if specified, the result is written to this array
            add : if True, add the result to the target array's current value
            jacobian : if True, return the jacobian linear operator of T

        Returns:
            The jacobian of T as LinOp if jacobian==True, else None.
        """
        try:
            from pycuda import gpuarray
            if type(x) is gpuarray.GPUArray:
                return self._call_gpu(x, y=y, add=add, jacobian=jacobian)
        except ImportError:
            pass
        return self._call_cpu(x, y=y, add=add, jacobian=jacobian)

class LinOp(Operator):
    """ (Matrix-free) representation of a linear operator A """
    def __init__(self):
        Operator.__init__(self)
        self.adjoint = None # another LinOp

    def _call_gpu(self, x, y=None, add=False):
        raise Exception("This operator does not support GPU arrays")

    def _call_cpu(self, x, y=None, add=False):
        if hasattr(self, 'spmat'):
            mat = self.spmat
        elif hasattr(self, 'mat'):
            mat = self.mat
        else:
            raise Exception("This operator does not support numpy arrays")

        assert y is not None
        x, y = x.ravel(), y.ravel()
        if not add:
            y[:] = mat.dot(x)
        else:
            y += mat.dot(x)

    def __call__(self, x, y=None, add=False, jacobian=False):
        try:
            from pycuda import gpuarray
            if type(x) is gpuarray.GPUArray:
                self._call_gpu(x, y=y, add=add)
                return self if jacobian else None
        except ImportError:
            pass
        self._call_cpu(x, y=y, add=add)
        # a linear operator is its own jacobian!
        return self if jacobian else None

    def rowwise_lp(self, y, p=1, add=False):
        """ Compute the rowwise l^p norm of A

        y_i = |A_i|_p^p = \sum_j |A_ij|^p

        Args:
            y : numpy array of shape (A.y.size,)
            p : positive float
            add : if True, add the result to the current value of y

        Returns:
            Nothing, the result is written to y.
        """
        if not add:
            y.fill(0.0)
        if hasattr(self, 'spmat'):
            y += sp.linalg.norm(self.spmat, ord=p, axis=1)
        elif hasattr(self, 'mat'):
            y += np.linalg.norm(self.mat, ord=p, axis=1)
        else:
            raise Exception("This operator does not support rowwise l^p norms.")

class Functional(object):
    """ Representation of a mathematical functional F """
    def __init__(self):
        self.x = None # Variable, input (domain)
        self.conj = None # Functional, convex conjugate

    def __call__(self, x, grad=False):
        """ Evaluate F(x)

        Args:
            x : numpy array
            grad : if True, return the gradient of F at x

        Returns:
            (val,infeas) or ((val,infeas), grad) if grad==True
            val : the value of
            infeas : for functionals taking the value infty, this is a measure
                for the distance to the feasible set
            grad : same shape as x
        """
        raise Exception("This functional does not support being called")

    def prox(self, tau):
        """ Proximal operator of F

        In general:
            F.prox(tau)(x1) = argmin(x2)[0.5*|x2-x1|^2 + tau*F(x2)]
            x = F.prox(tau)(x) + tau*F.conj.prox(1/tau)(x/tau)

        Args:
            tau : scalar or numpy array of shape (self.x.size,)

        Returns:
            Instance of opymize.operators.Operator
        """
        raise Exception("This functional does not support taking prox")

reason: import numpy,import scipy,from pycuda


text:
import numpy as np

def get_hop_distance(num_node, edge, max_hop=1):
    A = np.zeros((num_node, num_node))
    for i, j in edge:
        A[j, i] = 1
        A[i, j] = 1

    #print(A)
    # compute hop steps
    hop_dis = np.zeros((num_node, num_node)) + np.inf
    transfer_mat = [np.linalg.matrix_power(A, d) for d in range(max_hop + 1)]
    #print(transfer_mat)
    arrive_mat = (np.stack(transfer_mat) > 0)
    for d in range(max_hop, -1, -1):
        hop_dis[arrive_mat[d]] = d
    return hop_dis

def get_adjacency(hop_dis, strategy):
    valid_hop = range(0, 2, 1)
    adjacency = np.zeros((18, 18))
    for hop in valid_hop:
        adjacency[hop_dis == hop] = 1
    normalize_adjacency = normalize_digraph(adjacency)

    if strategy == 'uniform':
        A = np.zeros((1, num_node, num_node))
        A[0] = normalize_adjacency
    elif strategy == 'distance':
        A = np.zeros((len(valid_hop), num_node, num_node))
        for i, hop in enumerate(valid_hop):
            A[i][hop_dis == hop] = normalize_adjacency[hop_dis == hop]
    return A

def normalize_digraph(A):
    Dl = np.sum(A, 0)
    num_node = A.shape[0]
    Dn = np.zeros((num_node, num_node))
    for i in range(num_node):
        if Dl[i] > 0:
            Dn[i, i] = Dl[i]**(-1)
    AD = np.dot(A, Dn)
    return AD

num_node = 18
self_link = [(i, i) for i in range(num_node)]
neighbor_link = [(4, 3), (3, 2), (7, 6), (6, 5), (13, 12), (12,11),
                 (10, 9), (9, 8), (11, 5), (8, 2), (5, 1), (2, 1),
                 (0, 1), (15, 0), (14, 0), (17, 15), (16, 14)]

edge = self_link + neighbor_link
hop_dis = get_hop_distance(18, edge)
print(hop_dis)
print(get_adjacency(hop_dis, 'distance'))


text:
import tsfm
import torch
import os.path
import numpy as np
import torch.nn as nn
import torch.optim as optim
from torchvision import transforms
from lodgingdataset import LodgingDataset
from models import DeepLodge
from utils import train_model, train_tfl_model
from dataloader import LodgingDataLoader, LodgingImageLoader
from utils import get_transformations, get_stats

###############################################################################
###############################################################################
def run(model_ft, train_dir, test_dir, crop, num_channels, batch_size,
        num_epochs):
    # Get dataset stats
    dataset_mean, dataset_std = get_stats(crop)
    # parameters for GLCM texture feature extraction
    glcm_pars = {'distances': [1, 2, 4, 5],
                 'angles': [0, np.pi/2, 3*np.pi/4, np.pi],
                 'levels': 256,
                 'symmetric': False,
                 'normed': True,
                 'prop': 'contrast'
                 }
    # parameters for LBP texture feature extraction
    lbp_pars = {'radius': 1, 'num_points': 8, 'num_uniform_bins': 10,
                'num_var_bins': 16, 'hist_range': (0, 7000)}
    handcrafted = {'GLCM': glcm_pars, 'LBP': lbp_pars}
    # Set the image file extension
    file_name_pattern='.tif'
    # Define transformations
    train_tsfm, test_tsfm, feature_tsfm = get_transformations(num_channels, crop)
    # Create dataloader

    loader = LodgingDataLoader(train_dir, test_dir, dataset_mean, dataset_std,
                               batch_size, file_name_pattern, handcrafted,
                               train_tsfm, test_tsfm, feature_tsfm)
    dataloaders = loader.dataloaders
    dataset_sizes = loader.dataset_sizes
    # Determine devide type: GPU versus CPU
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    # Building the network and training it
    ###########################################################################
    model_ft = model_ft.to(device)
    criterion = nn.CrossEntropyLoss()
    ###########################################################################
    optimizer_ft = optim.Adam(model_ft.parameters())
    model_ft = train_model(model_ft, dataloaders, criterion, optimizer_ft,
                           num_epochs=num_epochs, device=device,
                           dataset_sizes=dataset_sizes)
    return model_ft, loader

def run_tfl(model_ft, train_dir, test_dir, crop, batch_size, num_epochs):
    # Get dataset stats
    dataset_mean, dataset_std = get_stats(crop)
    file_name_pattern='.tif'
    # Define transformations
    num_channels = 3

    train_tsfm = transforms.Compose([transforms.Resize(256),
                                     transforms.CenterCrop(224),
                                     transforms.RandomHorizontalFlip(),
                                     transforms.RandomVerticalFlip(),
                                     transforms.ToTensor(),
                                     transforms.Normalize(mean=dataset_mean[:num_channels],
                                                          std=dataset_std[:num_channels])])

    test_tsfm = transforms.Compose([transforms.Resize(256),
                                    transforms.CenterCrop(224),
                                    transforms.ToTensor(),
                                    transforms.Normalize(mean=dataset_mean[:num_channels],
                                                         std=dataset_std[:num_channels])])
    # Create dataloader
    loader = LodgingImageLoader(train_dir, test_dir, dataset_mean, dataset_std,
                               batch_size, file_name_pattern, train_tsfm,
                               test_tsfm)
    dataloaders = loader.dataloaders
    dataset_sizes = loader.dataset_sizes
    # Determine devide type: GPU versus CPU
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    # Building the network and training it
    ###########################################################################
    model_ft = model_ft.to(device)
    criterion = nn.CrossEntropyLoss()
    ###########################################################################
    optimizer_ft = optim.Adam(model_ft.parameters())
    model_ft = train_tfl_model(model_ft, dataloaders, criterion, optimizer_ft,
                               num_epochs=num_epochs, device=device,
                               dataset_sizes=dataset_sizes)
    return model_ft, loader

max_stars_repo_name: FarhadMaleki/LodgedNet

max_issues_repo_name: FarhadMaleki/LodgedNet

max_forks_repo_name: FarhadMaleki/LodgedNet


text:
# #############################################################################
# map.py
# ======
# Author : Matthieu Simeoni [matthieu.simeoni@gmail.com]
# #############################################################################

r"""
Abstract classes for multidimensional nonlinear maps.
"""

import numpy as np
from abc import ABC, abstractmethod
from typing import Union, Tuple
from numbers import Number
from pycsou.util.misc import is_range_broadcastable, range_broadcast_shape

class Map(ABC):
    r"""
    Base class for multidimensional maps.

    Any instance/subclass of this class must at least implement the abstract method ``__call__``.
    This class supports the following arithmetic operators ``+``, ``-``, ``*``, ``@``, ``**`` and ``/``, implemented with the
    class methods ``__add__``/``__radd__``, ``__sub__``/``__neg__``, ``__mul__``/``__rmul__``, ``__matmul__``, ``__pow__``, ``__truediv__``.
    Such arithmetic operators can be used to *add*, *substract*, *scale*, *compose*, *exponentiate* or *evaluate* ``Map`` instances.

    Examples
    --------
    .. testsetup::

       import numpy as np
       from pycsou.func.penalty import L1Norm, SquaredL2Norm
       from pycsou.linop.conv import Convolve1D
       from scipy import signal

       x = np.arange(10)
       filter = signal.hann(5)
       filter[filter.size//2:] = 0
       f1 = L1Norm(dim=x.size)
       f2 = SquaredL2Norm(dim=x.size)
       L1 = Convolve1D(size=x.size, filter=filter)
       L2 = Convolve1D(size=x.size, filter=filter/2)

    Consider four maps: two nonlinear functionals :math:`f_1:\mathbb{R}^{10}\to \mathbb{R}`, :math:`f_2:\mathbb{R}^{10}\to \mathbb{R}`
    and two linear operators :math:`L_1:\mathbb{R}^{10}\to \mathbb{R}^{10}`, :math:`L_2:\mathbb{R}^{10}\to \mathbb{R}^{10}`.

    .. doctest::

       >>> print(f1.shape, f2.shape, L1.shape, L2.shape)
       (1, 10) (1, 10) (10, 10) (10, 10)

    We can combine linearly/compose the maps with consistent domains/ranges:

    .. doctest::

       >>> f3 = f1 / 3 + np.pi * f2
       >>> np.allclose(f3(x), f1(x) / 3 + np.pi * f2(x))
       True
       >>> L3 = L1 * 3 - (L2 ** 2) / 6
       >>> np.allclose(L3(x), L1(x) * 3 - (L2(L2(x))) / 6)
       True
       >>> f4 = f3 * L3
       >>> np.allclose(f4(x), f3(L3(x)))
       True

    Note that multiplying a map with an array is the same as evaluating the map at the array.

    .. doctest::

       >>> np.allclose(f3 * x, f3(x))
       True

    The multiplication operator ``@`` can also be used in place of ``*``, in compliance with Numpy's interface:

    .. doctest::

       >>> np.allclose(f3 * x, f3 @ x)
       True
       >>> np.allclose((f1 * L1)(x), (f1 @ L1)(x))
       True

    Finally, maps can be shifted via the method ``shifter``:

    .. doctest::

       >>> f5=f4.shifter(shift=2 * x)
       >>> np.allclose(f5(x), f4(x + 2 * x))
       True

    """

    def __init__(self, shape: Tuple[int, int], is_linear: bool = False, is_differentiable: bool = False):
        r"""
        Parameters
        ----------
        shape: Tuple[int, int]
            Shape of the map.
        is_linear: bool
            Whether the map is linear or not.
        is_differentiable: bool
            Whether the map is differentiable or not.
        """
        if len(shape) > 2:
            raise NotImplementedError(
                'Shapes of map objects must be tuples of length 2 (tensorial maps not supported).')
        self.shape = shape
        self.is_linear = is_linear
        self.is_functional = True if self.shape[0] == 1 else False
        self.is_differentiable = is_differentiable
        super(Map, self).__init__()

    @abstractmethod
    def __call__(self, arg: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
        r"""
        Call self as a function.

        Parameters
        ----------
        arg: Union[Number, np.ndarray]
            Argument of the map.

        Returns
        -------
        Union[Number, np.ndarray]
            Value of ``arg`` through the map.
        """
        pass

    def shifter(self, shift: Union[Number, np.ndarray]) -> 'MapShifted':
        r"""
        Returns a shifted version of the map.

        Parameters
        ----------
        shift: Union[Number, np.ndarray]
            Shift vector.

        Returns
        -------
        :py:class:`~pycsou.core.map.MapShifted`
            Shifted map.

        Notes
        -----
        Let ``A`` be a ``Map`` instance  and ``B=A.shifter(y)`` with ``y`` some vector in the domain of ``A``. Then we have:
        ``B(x)=A(x+y)``.
        """
        return MapShifted(map=self, shift=shift)

    def __add__(self, other: 'Map') -> 'MapSum':
        r"""
        Add a map with another map instance.

        Parameters
        ----------
        other: Union['Map', np.ndarray]
            The other map or array to be added.

        Returns
        -------
        :py:class:`~pycsou.core.map.MapSum`
            Sum of the map with another map.

        Raises
        ------
        NotImplementedError
            If ``other`` is not a map.
        """

        # if isinstance(other, np.ndarray):
        #     return MapBias(self, bias=other)
        if isinstance(other, Map):
            return MapSum(self, other)
        else:
            raise NotImplementedError

    def __radd__(self, other: 'Map') -> 'MapSum':
        r"""
        Add a map with another map instance when the latter is on the right hand side of the summation.

        Parameters
        ----------
        other: Union['Map', np.ndarray]
            The other map or array to be added.

        Returns
        -------
        :py:class:`~pycsou.core.map.MapSum`
            Sum of the map with another map.

        Raises
        ------
        NotImplementedError
            If ``other`` is not a map.

        """

        # if isinstance(other, np.ndarray):
        #     return MapBias(self, bias=other)
        if isinstance(other, Map):
            return MapSum(other, self)
        else:
            raise NotImplementedError

    def __mul__(self, other: Union[Number, 'Map', np.ndarray]) -> Union['MapComp', np.ndarray]:
        r"""
        Multiply a map with another map, a scalar, or an array.

        The behaviour of this method depends on the type of ``other``:

        * If ``other`` is another map, then it returns the composition of the map with ``other``.
        * If ``other`` is an array with compatible shape, then it calls the map on ``other``.
        * If ``other`` is a scalar, then it multiplies the map with this scalar.

        Parameters
        ----------
        other: Union[Number, 'Map', np.ndarray]
            Scalar, map or array with which to multiply.

        Returns
        -------
        :py:class:`~pycsou.core.map.MapComp`, np.ndarray
            Composition of the map with another map, or product with a scalar or array.

        Raises
        ------
        NotImplementedError
            If ``other`` is not a scalar, a map or an array.

        See Also
        --------
        :py:func:`~pycsou.core.map.Map.__matmul__`, :py:func:`~pycsou.core.map.Map.__rmul__`

        """
        if isinstance(other, Number):
            from pycsou.linop.base import HomothetyMap

            other = HomothetyMap(constant=other, size=self.shape[1])

        if isinstance(other, np.ndarray):
            return self(other)
        elif isinstance(other, Map):
            return MapComp(self, other)
        else:
            raise NotImplementedError

    def __rmul__(self, other: Union[Number, 'Map']) -> 'MapComp':
        if isinstance(other, Number):
            from pycsou.linop.base import HomothetyMap

            other = HomothetyMap(constant=other, size=self.shape[0])

        if isinstance(other, Map):
            return MapComp(other, self)
        else:
            raise NotImplementedError

    def __matmul__(self, other: 'Map') -> 'MapComp':
        r"""Alias for :py:func:`~pycsou.core.map.Map.__mul__` offered to comply with Numpy's interface."""
        return self.__mul__(other)

    def __neg__(self) -> 'MapComp':
        r"""Negates a map. Alias for ``self.__mul__(-1)``."""
        return self.__mul__(-1)

    def __sub__(self, other: Union['Map', np.ndarray]) -> Union['MapSum', 'MapBias']:
        r"""Substracts a map, scalar, or array to another map. Alias for ``self.__add__(other.__neg__())``."""
        other = other.__neg__()
        return self.__add__(other)

    def __pow__(self, power: int) -> 'MapComp':
        r"""Raise a map to a certain ``power``. Alias for ``A*A*...*A`` with ``power`` multiplications."""
        if type(power) is int:
            exp_map = self
            for i in range(1, power):
                exp_map = self.__mul__(exp_map)
            return exp_map
        else:
            raise NotImplementedError

    def __truediv__(self, scalar: Number) -> 'MapComp':
        r"""Divides a map by a ``scalar``. Alias for ``self.__mul__(1 / scalar)``."""
        if isinstance(scalar, Number):
            return self.__mul__(1 / scalar)
        else:
            raise NotImplementedError

class MapShifted(Map):
    def __init__(self, map: Map, shift: Union[Number, np.ndarray]):
        self.map = map
        self.shift = shift
        if shift.size != map.shape[1]:
            raise TypeError('Invalid shift size.')
        Map.__init__(self, shape=map.shape, is_linear=map.is_linear, is_differentiable=map.is_differentiable)

    def __call__(self, arg: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
        return self.map(arg + self.shift)

class MapSum(Map):
    def __init__(self, map1: Map, map2: Map):
        if not is_range_broadcastable(map1.shape, map2.shape):
            raise ValueError('Cannot sum two maps with inconsistent range or domain sizes.')
        else:
            Map.__init__(self, shape=range_broadcast_shape(map1.shape, map2.shape),
                         is_linear=map1.is_linear & map2.is_linear,
                         is_differentiable=map1.is_differentiable & map2.is_differentiable)
            self.map1, self.map2 = map1, map2

    def __call__(self, arg: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
        return self.map1(arg) + self.map2(arg)

# class MapBias(Map):
#     def __init__(self, map_: Map, bias: np.ndarray):
#         if not is_range_broadcastable(map_.shape, bias.shape):
#             raise ValueError('Inconsistent range sizes between map and bias.')
#         else:
#             super(MapBias, self).__init__(shape=map_.shape, is_linear=False, is_differentiable=map_.is_differentiable)
#             self.map, self.bias = map_, bias
#
#     def __call__(self, arg: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
#         return self.map(arg) + self.bias

class MapComp(Map):
    def __init__(self, map1: Map, map2: Map):
        if map1.shape[1] != map2.shape[0] and map2.shape[0] != 1 and map1.shape[1] != 1:
            raise ValueError('Cannot compose two maps with inconsistent range or domain sizes.')
        else:
            Map.__init__(self, shape=(map1.shape[0], map2.shape[1]),
                         is_linear=map1.is_linear & map2.is_linear,
                         is_differentiable=map1.is_differentiable & map2.is_differentiable)
            self.map1 = map1
            self.map2 = map2

    def __call__(self, arg: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
        return self.map1(self.map2(arg))

class DifferentiableMap(Map):
    r"""
    Base class for multidimensional differentiable maps.

    Any instance/subclass of this class must at least implement the abstract methods ``__call__`` and ``jacobianT``.
    The attributes ``lipschitz_cst``, ``diff_lipschitz_cst`` and the method ``jacobianT`` are automatically updated
    using standard differentiation rules when scaling, composing , summing or shifting differentiable maps.

    Examples
    --------
    .. testsetup::

       import numpy as np
       from pycsou.func.penalty import SquaredL2Norm
       from pycsou.linop.diff import FirstDerivative

       x = np.arange(10)
       f = SquaredL2Norm(dim=x.size)
       L1 = FirstDerivative(size=x.size)
       L1.compute_lipschitz_cst()
       L2 = FirstDerivative(size=x.size, kind='centered')
       L2.compute_lipschitz_cst()

    Consider three differentiable maps: a nonlinear differentiable functional :math:`f:\mathbb{R}^{10}\to \mathbb{R}`,
    and two linear operators :math:`L_1:\mathbb{R}^{10}\to \mathbb{R}^{10}`, :math:`L_2:\mathbb{R}^{10}\to \mathbb{R}^{10}`.

    .. doctest::

       >>> print(f.lipschitz_cst, f.diff_lipschitz_cst)
       inf 2
       >>> print(np.round(L1.lipschitz_cst,1), np.round(L1.diff_lipschitz_cst,1))
       1.0 1.0
       >>> print(np.round(L2.lipschitz_cst,1), np.round(L2.diff_lipschitz_cst,1))
       0.9 0.9
       >>> L3 = 2 * L1 + (L2 ** 2) / 3
       >>> np.allclose(L3.lipschitz_cst, 2 * L1.lipschitz_cst + (L2.lipschitz_cst ** 2)/3)
       True
       >>> map_ = f * L1
       >>> np.allclose(map_.lipschitz_cst, f.lipschitz_cst * L1.lipschitz_cst)
       True
       >>> np.allclose(map_.jacobianT(x), L1.jacobianT(x) * f.jacobianT(L1(x)))
       True

    Notes
    -----
    This class supports the following arithmetic operators ``+``, ``-``, ``*``, ``@``, ``**`` and ``/``, implemented with the
    class methods ``__add__``/``__radd__``, ``__sub__``/``__neg__``, ``__mul__``/``__rmul__``, ``__matmul__``, ``__pow__``, ``__truediv__``.
    Such arithmetic operators can be used to *add*, *substract*, *scale*, *compose*, *exponentiate* or *evaluate* ``DifferentiableMap`` instances.

    """

    def __init__(self, shape: Tuple[int, int], is_linear: bool = False, lipschitz_cst: float = np.infty,
                 diff_lipschitz_cst: float = np.infty):
        r"""
        Parameters
        ----------
        shape: Tuple[int, int]
            Shape of the differentiable map.
        is_linear: bool
            Whether the differentiable map is linear or not.
        lipschitz_cst: float
            Lispchitz constant of the differentiable map if it exists/is known. Default to :math:`+\infty`.
        diff_lipschitz_cst: float
            Lispchitz constant of the derivative of the differentiable map if it exists/is known. Default to :math:`+\infty`.
        """
        super(DifferentiableMap, self).__init__(shape=shape, is_linear=is_linear, is_differentiable=True)
        self.lipschitz_cst = lipschitz_cst
        self.diff_lipschitz_cst = diff_lipschitz_cst

    @abstractmethod
    def jacobianT(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
        r"""
        Transpose of the Jacobian matrix of the differentiable map evaluated at ``arg``.

        Parameters
        ----------
        arg: Union[Number, np.ndarray]
            Point at which the transposed Jacobian matrix is evaluated.

        Returns
        -------
        ::py:class:`~pycsou.core.linop.LinearOperator`
            Linear operator associated to the transposed Jacobian matrix.
        """
        pass

    def gradient(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
        r"""
        Alias for ``self.jacobianT``.
        """
        return self.jacobianT(arg)

    def compute_lipschitz_cst(self):
        r"""
        User-implemented method to compute the Lipschitz constant of the map.
        """
        pass

    def compute_diff_lipschitz_cst(self):
        r"""
        User-implemented method to compute the Lipschitz constant of the derivative of the map.
        """
        pass

    def shifter(self, shift: Union[Number, np.ndarray]) -> 'DiffMapShifted':
        r"""
        Returns a shifted version of the map.

        Parameters
        ----------
        shift: Union[Number, np.ndarray]
            Shift vector.

        Returns
        -------
        :py:class:`~pycsou.core.map.DiffMapShifted`
            Shifted map.

        See Also
        --------
        :py:meth:`~pycsou.core.map.Map.shifter`
        """
        return DiffMapShifted(map=self, shift=shift)

    def __add__(self, other: Union['Map', 'DifferentiableMap']) -> Union['MapSum', 'DiffMapSum']:
        # if isinstance(other, np.ndarray):
        #     return DiffMapBias(self, other)
        if isinstance(other, DifferentiableMap):
            return DiffMapSum(self, other)
        elif isinstance(other, Map):
            return MapSum(self, other)
        else:
            raise NotImplementedError

    def __radd__(self, other: Union['Map', 'DifferentiableMap']) -> Union['MapSum', 'DiffMapSum']:
        # if isinstance(other, np.ndarray):
        #     return DiffMapBias(self, other)
        if isinstance(other, DifferentiableMap):
            return DiffMapSum(self, other)
        elif isinstance(other, Map):
            return MapSum(self, other)
        else:
            raise NotImplementedError

    def __mul__(self, other: Union[Number, 'Map', 'DifferentiableMap', np.ndarray]) \
            -> Union['MapComp', 'DiffMapComp', np.ndarray]:
        if isinstance(other, Number):
            from pycsou.linop.base import HomothetyMap

            other = HomothetyMap(constant=other, size=self.shape[1])

        if isinstance(other, np.ndarray):
            return self(other)
        elif isinstance(other, DifferentiableMap):
            return DiffMapComp(self, other)
        elif isinstance(other, Map):
            return MapComp(self, other)
        else:
            raise NotImplementedError

    def __rmul__(self, other: Union[Number, 'Map', 'DifferentiableMap']) -> Union['MapComp', 'DiffMapComp']:
        if isinstance(other, Number):
            from pycsou.linop.base import HomothetyMap

            other = HomothetyMap(constant=other, size=self.shape[0])

        if isinstance(other, DifferentiableMap):
            return DiffMapComp(other, self)
        elif isinstance(other, Map):
            return MapComp(other, self)
        else:
            raise NotImplementedError

class DiffMapShifted(MapShifted, DifferentiableMap):
    def __init__(self, map: DifferentiableMap, shift: Union[Number, np.ndarray]):
        MapShifted.__init__(self, map=map, shift=shift)
        DifferentiableMap.__init__(self, shape=self.shape, is_linear=self.is_linear,
                                   lipschitz_cst=self.map.lipschitz_cst,
                                   diff_lipschitz_cst=self.map.diff_lipschitz_cst)

    def jacobianT(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
        return self.map.jacobianT(arg + self.shift)

class DiffMapSum(MapSum, DifferentiableMap):
    def __init__(self, map1: DifferentiableMap, map2: DifferentiableMap):
        MapSum.__init__(self, map1=map1, map2=map2)
        DifferentiableMap.__init__(self, shape=self.shape, is_linear=self.is_linear,
                                   lipschitz_cst=self.map1.lipschitz_cst + self.map2.lipschitz_cst,
                                   diff_lipschitz_cst=self.map1.diff_lipschitz_cst + self.map2.diff_lipschitz_cst)

    def jacobianT(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
        return self.map1.jacobianT(arg) + self.map2.jacobianT(arg)

# class DiffMapBias(MapBias, DifferentiableMap):
#     def __init__(self, map_: DifferentiableMap, bias: np.ndarray):
#         MapBias.__init__(self, map_=map_, bias=bias)
#         DifferentiableMap.__init__(self, shape=self.shape, is_linear=False, lipschitz_cst=self.map.lipschitz_cst,
#                                    diff_lipschitz_cst=self.map.diff_lipschitz_cst)
#
#     def jacobianT(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
#         return self.map.jacobianT(arg) * self.map2.jacobianT(arg) * self.map1.jacobianT(self.map2(arg))

class DiffMapComp(MapComp, DifferentiableMap):
    def __init__(self, map1: DifferentiableMap, map2: DifferentiableMap):
        from pycsou.linop.base import HomothetyMap

        MapComp.__init__(self, map1=map1, map2=map2)
        lipschitz_cst = self.map2.lipschitz_cst * self.map1.lipschitz_cst
        if isinstance(map1, HomothetyMap):
            diff_lipschitz_cst = self.map1.diff_lipschitz_cst * self.map2.diff_lipschitz_cst
        else:
            diff_lipschitz_cst = self.map1.diff_lipschitz_cst * self.map2.diff_lipschitz_cst * self.map2.lipschitz_cst
        DifferentiableMap.__init__(self, shape=self.shape, is_linear=self.is_linear,
                                   lipschitz_cst=lipschitz_cst, diff_lipschitz_cst=diff_lipschitz_cst)

    def jacobianT(self, arg: Union[Number, np.ndarray]) -> 'LinearOperator':
        return self.map2.jacobianT(arg) * self.map1.jacobianT(self.map2(arg))

class MapStack(Map):
    r"""
    Stack maps together.

    This class constructs a map by stacking multiple maps together, either **vertically** (``axis=0``) or **horizontally** (``axis=1``):

    - **Vertical stacking**: Consider a collection :math:`\{L_i:\mathbb{R}^{N}\to \mathbb{R}^{M_i}, i=1,\ldots, k\}`
      of maps. Their vertical stacking is defined as the operator

      .. math::

         V:\begin{cases}\mathbb{R}^{N}\to \mathbb{R}^{M_1}\times \cdots \times\mathbb{R}^{M_k}\\
         \mathbf{x}\mapsto (L_1\mathbf{x},\ldots, L_k\mathbf{x}).
         \end{cases}

    - **Horizontal stacking**: Consider a collection :math:`\{L_i:\mathbb{R}^{N_i}\to \mathbb{R}^{M}, i=1,\ldots, k\}`
      of maps. Their horizontal stacking is defined as the operator

      .. math::

         H:\begin{cases}\mathbb{R}^{N_1}\times \cdots \times\mathbb{R}^{N_k}\to \mathbb{R}^{M}\\
         (\mathbf{x}_1,\ldots, \mathbf{x}_k)\mapsto \sum_{i=1}^k L_i \mathbf{x}_i.
         \end{cases}

    Examples
    --------

    .. testsetup::

       import numpy as np
       from pycsou.func.penalty import SquaredL2Norm
       from pycsou.linop.diff import FirstDerivative
       from pycsou.core.map import MapStack

       x = np.arange(10)
       y = np.arange(20)
       f = SquaredL2Norm(dim=x.size)
       g = SquaredL2Norm(dim=y.size)
       A1 = FirstDerivative(size=x.size)
       A2 = FirstDerivative(size=y.size, kind='centered')
       A3 = A1 / 2
       K1 = f * A1
       K2 = g * A2
       K3 = A3

    Consider three maps :math:`K_1:\mathbb{R}^{10}\to \mathbb{R}`, :math:`K_2:\mathbb{R}^{20}\to \mathbb{R}` and
    :math:`K_3:\mathbb{R}^{10}\to \mathbb{R}^{10}`

    .. doctest::

       >>> V = MapStack(K1, K3, axis=0)
       >>> V.shape
       (11, 10)
       >>> np.allclose(V(x), np.concatenate((K1(x).flatten(), K3(x).flatten())))
       True
       >>> H = MapStack(K1,K2, axis=1)
       >>> H.shape
       (1, 30)
       >>> np.allclose(H(np.concatenate((x,y))), K1(x) + K2(y))
       True

    See Also
    --------
    :py:class:`~pycsou.core.map.MapVStack`, :py:class:`~pycsou.core.map.MapHStack`, :py:class:`~pycsou.core.map.DiffMapStack`

    """

    def __init__(self, *maps: Map, axis: int):
        r"""
        Parameters
        ----------
        maps: Map
            List of maps to stack.
        axis:
            Stacking direction: 0 for vertical and 1 for horizontal stacking.
        """
        self.maps = list(maps)
        if (np.abs(axis) > 1):
            ValueError('Axis must be one of {0, 1,-1}.')
        self.axis = int(axis)
        self.is_linear_list = [map_.is_linear for map_ in self.maps]
        self.is_differentiable_list = [map_.is_differentiable for map_ in self.maps]
        self.shapes = np.array([map_.shape for map_ in self.maps])
        self.block_sizes = [map_.shape[axis] for map_ in self.maps]
        self.sections = np.cumsum(self.block_sizes)

        if not self.is_valid_stack():
            raise ValueError('Inconsistent map shapes for  stacking.')
        Map.__init__(self, shape=self.get_shape(),
                     is_linear=bool(np.prod(self.is_linear_list).astype(bool)),
                     is_differentiable=bool(np.prod(self.is_differentiable_list).astype(bool)))

    def __call__(self, x: Union[Number, np.ndarray]) -> Union[Number, np.ndarray]:
        if self.axis == 0:
            out_list = [map_.__call__(x).flatten() for map_ in self.maps]
            return np.concatenate(out_list, axis=0)
        else:
            x_split = np.split(x, self.sections)
            result = 0
            for i, map_ in enumerate(self.maps):
                result += map_.__call__(x_split[i])
            return result

    def is_valid_stack(self) -> bool:
        col_sizes = [map_.shape[1 - self.axis] for map_ in self.maps]
        return np.unique(col_sizes).size == 1

    def get_shape(self) -> Tuple[int, int]:
        sizes = [map_.shape[self.axis] for map_ in self.maps]
        if self.axis == 0:
            return (int(np.sum(sizes).astype(int)), self.maps[0].shape[1 - self.axis])
        else:
            return (self.maps[0].shape[1 - self.axis], int(np.sum(sizes).astype(int)))

class MapVStack(MapStack):
    r"""
    Alias for vertical stacking, equivalent to ``MapStack(*maps, axis=0)``.

    Examples
    --------

    .. testsetup::

       from pycsou.core.map import MapVStack

    .. doctest::

       >>> V1 = MapStack(K1, K3, axis=0)
       >>> V2 = MapVStack(K1, K3)
       >>> np.allclose(V1(x), V2(x))
       True

    """

    def __init__(self, *maps: Map):
        r"""
        Parameters
        ----------
        maps: Map
            List of maps to stack.
        """
        super(MapVStack, self).__init__(*maps, axis=0)

class MapHStack(MapStack):
    r"""
    Alias for horizontal stacking, equivalent to ``MapStack(*maps, axis=1)``.

    Examples
    --------

    .. testsetup::

      from pycsou.core.map import MapHStack

    .. doctest::

      >>> V1 = MapStack(K1, K2, axis=1)
      >>> V2 = MapHStack(K1, K2)
      >>> np.allclose(V1(np.arange(V1.shape[1])), V2(np.arange(V1.shape[1])))
      True

    """

    def __init__(self, *maps: Map):
        r"""
        Parameters
        ----------
        maps: Map
            List of maps to stack.
        """
        super(MapHStack, self).__init__(*maps, axis=1)

class DiffMapStack(MapStack, DifferentiableMap):
    r"""
    Stack differentiable maps together.

    This class constructs a differentiable map by stacking multiple maps together, either **vertically** (``axis=0``) or **horizontally** (``axis=1``).
    The attributes ``lipschitz_cst``, ``diff_lipschitz_cst`` and the method ``jacobianT`` are inferred from those of
    the stacked maps.

    - **Vertical stacking**: Consider a collection :math:`\{L_i:\mathbb{R}^{N}\to \mathbb{R}^{M_i}, i=1,\ldots, k\}`
      of maps, with Lipschitz constants :math:`\{\beta_i\in\mathbb{R}_+\cup\{+\infty\}, i=1,\ldots, k\}`, Jacobian matrices
      :math:`\{\mathbf{J}_i(\mathbf{x})\in\mathbb{R}^{M_i\times N},  i=1,\ldots, k\}` for some :math:`\mathbf{x}\in\mathbb{R}^N`.
      Then the vertically stacked operator

      .. math::

         V:\begin{cases}\mathbb{R}^{N}\to \mathbb{R}^{M_1}\times \cdots \times\mathbb{R}^{M_k}\\
         \mathbf{x}\mapsto (L_1\mathbf{x},\ldots, L_k\mathbf{x}).
         \end{cases}

      has a Lipschitz constant bounded by :math:`\sqrt{\sum_{i=1}^k \beta_i^2}`. Moreover the Jacobian matrix of :math:`V` is obtained by stacking the individual Jacobian matrices vertically:

      .. math::

         \mathbf{J}(\mathbf{x})=\left[\begin{array}{c}\mathbf{J}_1(\mathbf{x})\\\vdots\\ \mathbf{J}_k(\mathbf{x}) \end{array}\right]\in\mathbb{R}^{(\sum_{i=1}^k M_i)\times N}.

    - **Horizontal stacking**: Consider a collection :math:`\{L_i:\mathbb{R}^{N_i}\to \mathbb{R}^{M}, i=1,\ldots, k\}`
      of maps, with Lipschitz constants :math:`\{\beta_i\in\mathbb{R}_+\cup\{+\infty\}, i=1,\ldots, k\}`, Jacobian matrices
      :math:`\{\mathbf{J}_i(\mathbf{x}_i)\in\mathbb{R}^{M\times N_i},  i=1,\ldots, k\}` for some :math:`\mathbf{x}_i\in\mathbb{R}^{N_i}`.
      Then the horizontally stacked operator

      .. math::

         H:\begin{cases}\mathbb{R}^{N_1}\times \cdots \times\mathbb{R}^{N_k}\to \mathbb{R}^{M}\\
         (\mathbf{x}_1,\ldots, \mathbf{x}_k)\mapsto \sum_{i=1}^k L_i \mathbf{x}_i.
         \end{cases}

      has a Lipschitz constant bounded by :math:`\sqrt{\sum_{i=1}^k \beta_i^2}`. Moreover the Jacobian matrix of :math:`H` is obtained by stacking the individual Jacobian matrices horizontally:

      .. math::

         \mathbf{J}(\mathbf{x}_1,\ldots,\mathbf{x}_k)=\left[\begin{array}{c}\mathbf{J}_1(\mathbf{x}_1)&\cdots& \mathbf{J}_k(\mathbf{x}_k) \end{array}\right]\in\mathbb{R}^{M\times (\sum_{i=1}^k N_i)}.

    Examples
    --------

    .. testsetup::

       import numpy as np
       from pycsou.func.penalty import SquaredL2Norm
       from pycsou.linop.diff import FirstDerivative
       from pycsou.core.map import MapStack, DiffMapStack

       x = np.arange(10)
       y = np.arange(20)
       f = SquaredL2Norm(dim=x.size)
       g = SquaredL2Norm(dim=y.size)
       A1 = FirstDerivative(size=x.size)
       A1.compute_lipschitz_cst()
       A2 = FirstDerivative(size=y.size, kind='centered')
       A2.compute_lipschitz_cst()
       A3 = A1 / 2
       K1 = f * A1
       K2 = g * A2
       K3 = A3

    .. doctest::

       >>> V = DiffMapStack(K1, K3, axis=0)
       >>> np.allclose(V.lipschitz_cst, np.sqrt(K1.lipschitz_cst ** 2 + K3.lipschitz_cst ** 2))
       True

    See Also
    --------
    :py:class:`~pycsou.core.map.DiffMapVStack`, :py:class:`~pycsou.core.map.DiffMapHStack`, :py:class:`~pycsou.core.map.MapStack`

    """

    def __init__(self, *diffmaps: DifferentiableMap, axis: int):
        r"""
        Parameters
        ----------
        diffmaps: DifferentiableMap
            List of differentiable maps to be stacked
        axis: int
            Stacking direction: 0 for vertical and 1 for horizontal stacking.
        """
        MapStack.__init__(self, *diffmaps, axis=axis)
        lipschitz_cst = np.sqrt(np.sum([diffmap.lipschitz_cst ** 2 for diffmap in self.maps]))
        diff_lipschitz_cst = np.sqrt(np.sum([diffmap.diff_lipschitz_cst ** 2 for diffmap in self.maps]))
        DifferentiableMap.__init__(self, shape=self.shape, is_linear=self.is_linear, lipschitz_cst=lipschitz_cst,
                                   diff_lipschitz_cst=diff_lipschitz_cst)

    def jacobianT(self, arg: Union[Number, np.ndarray]) -> Union['LinOpHStack', 'LinOpVStack']:
        if self.axis == 0:
            jacobianT_list = [diffmap.jacobianT(arg) for diffmap in self.maps]
            return LinOpHStack(*jacobianT_list)
        else:
            arg_split = np.split(arg, self.sections)
            jacobianT_list = [diffmap.jacobianT(arg_split[i]) for i, diffmap in enumerate(self.maps)]
            return LinOpVStack(*jacobianT_list)

class DiffMapVStack(DiffMapStack):
    r"""
    Alias for vertical stacking of differentiable maps, equivalent to ``DiffMapStack(*maps, axis=0)``.
    """

    def __init__(self, *diffmaps: DifferentiableMap):
        r"""
        Parameters
        ----------
        diffmaps: DifferentiableMap
            List of differentiable maps to be stacked
        """
        super(DiffMapVStack, self).__init__(*diffmaps, axis=0)

class DiffMapHStack(DiffMapStack):
    r"""
    Alias for horizontal stacking of differentiable maps, equivalent to ``DiffMapStack(*maps, axis=1)``.
    """

    def __init__(self, *diffmaps: DifferentiableMap):
        r"""
        Parameters
        ----------
        diffmaps: DifferentiableMap
            List of differentiable maps to be stacked.
        """
        super(DiffMapHStack, self).__init__(*diffmaps, axis=1)

if __name__ == '__main__':
    import numpy as np
    from pycsou.linop.base import DenseLinearOperator, LinOpStack, LinOpVStack, LinOpHStack, HomothetyMap
    from pycsou.linop.conv import Convolve1D
    from scipy import signal

    x1 = np.arange(10)
    x2 = np.arange(20)
    filter = signal.hann(5)
    filter[filter.size // 2:] = 0
    L1 = Convolve1D(size=x1.size, filter=filter)
    L2 = DenseLinearOperator(np.arange(x2.size * L1.shape[0]).reshape(L1.shape[0], x2.size))
    L3 = LinOpStack(L1, L2, axis=1)
    L3(np.concatenate((x1, x2)))

max_stars_repo_name: poldap/pycsou

max_issues_repo_name: poldap/pycsou

max_forks_repo_name: poldap/pycsou

reason: import numpy,from scipy


text:
#! python
# -*- coding: utf-8 -*-
import cv2
import numpy as np
from numpy import pi
from numpy.fft import fft,ifft,fft2,ifft2,fftshift,fftfreq
from scipy import signal
from matplotlib import patches
from mwx.controls import LParam
from mwx.graphman import Layer
from plugins.lcrf import Model
from wxpyJemacs import wait
import editor as edi

def logpolar(src, r0, r1, center=None):
    """Log-Polar transform
    The area radii [r0:r1] of radius N/2 mapsto the same size of src image

    cf. cv2.logPolar(src, (nx,ny), M, cv2.INTER_CUBIC)
    """
    h, w = src.shape
    if center is None:
        xc, yc = w//2, h//2
    else:
        xc, yc = center

    x = np.arange(w, dtype=np.float32) /w
    y = np.arange(h, dtype=np.float32) * 2*pi /h
    xx, yy = np.meshgrid(x, y)

    rh0 = np.log(r0) if r0>0 else 0
    rh1 = np.log(r1)
    r = np.exp(rh0 + (rh1 - rh0) * xx)
    map_x = xc + r * np.cos(yy)
    map_y = yc + r * np.sin(yy)
    dst = cv2.remap(src.astype(np.float32), map_x, map_y, cv2.INTER_CUBIC)
    return dst

def find_ring_center(src, lo, hi, N=256, tol=0.01):
    """find ring pattern in buffer with fixed center

    Polar 変換した後，角度セグメントに分割して相互相関をとる．
    center 固定，楕円を計測するために log-polar を使用する．
    theta = 0 を基準として，相対変位 [pixels] を計算する．

    src : source buffer (typ. log(abs(fft)))
  lo-hi : masking size of radial axis
      N : resizing of angular axis (total step in angle [0:2pi])
    tol : remove peaks that leap greater than N * tol
  retval ->
        dst(log-polar-transformed image) and fitting model
    """
    h, w = src.shape

    dst = logpolar(src, lo, hi)

    ## Resize Y (angular) axis (計算を軽くするためリサイズ)
    rdst = cv2.resize(dst.astype(np.float32), (w, N), interpolation=cv2.INTER_AREA)

    rdst -= rdst.mean()
    rdst = cv2.GaussianBlur(rdst, (1,11), 0)

    temp = rdst[0][::-1] # template of corr; distr at theta=0(=2pi)
    data = []
    for fr in rdst:
        p = signal.fftconvolve(fr, temp, mode='same')
        data.append(p.argmax())

    ## 相関の計算は上から行うので，2pi --> 0 の並びのリストになる
    ##   最終的に返す計算結果は逆転させて，0 --> 2pi の並びにする
    Y = np.array(data[::-1]) - w/2
    X = np.arange(0, 1, 1/len(Y)) * 2*pi

    ## remove leaps(2): tol より小さいとびを許容する (画素サイズに比例)
    tolr = max(5, tol * w/2)
    xx, yy = [0.], [0.]
    for x,y in zip(X[1:], Y[1:]):
        if abs(y - yy[-1]) < tolr:
            xx.append(x)
            yy.append(y)

    fitting_curve = Model(xx, yy)

    ## edi.plot(xx, yy, '+', X, fitting_curve(X))

    fitting_curve.params[0] = 0 # :a=0 として(平均を基準とする)全体のオフセット量を評価する
    fitting_curve.params[1] = 0
    fitting_curve.params[2] = 0

    return dst, fitting_curve

def smooth1d(data, tol=0.01):
    w = len(data)
    lw = int(max(3, tol * w/2))
    if lw % 2 == 0:
        lw += 1
    return signal.savgol_filter(data, lw, polyorder=3)

def blur1d(data, tol=0.01):
    w = len(data)
    lw = int(max(3, tol * w/2))
    ## window = np.hanning(lw)
    window = signal.windows.gaussian(lw, std=lw)

    ## padding dumy data at both edge
    data = np.concatenate((data[:lw][::-1], data, data[-lw:][::-1]))
    ys = np.convolve(window/window.sum(), data, mode='same') # ならし
    ## ys = signal.lfilter(window/window.sum(), 1, data) # 短周期は苦手？NG
    return ys[lw:-lw]

def find_radial_peaks(data, tol=0.01):
    """find local maxima/minim's
    smoothing with Gaussian window (signal.windows.gaussian)
    """
    w = len(data)
    lw = int(max(3, tol * w/2))

    ys = blur1d(data, tol)

    ## maxima = signal.find_peaks_cwt(ys, np.arange(3,4))
    ## minima = signal.find_peaks_cwt(-ys, np.arange(3,4))
    maxima,_attr = signal.find_peaks(ys, width=lw/3)
    minima,_attr = signal.find_peaks(-ys, width=lw/3)

    ## maxima = signal.argrelmax(ys)
    ## minima = signal.argrelmin(ys)

    ## remove near-edge peaks
    def _edge(x):
        return x[(lw/2 < x) & (x < w-lw/2)]
    maxima = _edge(maxima)
    minima = _edge(minima)

    return ys, maxima, minima # np.sort(np.append(maxima, minima))

class Plugin(Layer):
    """CTF finder ver 1.0
    """
    menu = "Test"

    debug = 0

    def Init(self):
        self.rmin = LParam("rmin", (0.01, 0.1, 0.001), 0.05,
                           updater=lambda p: self.calc_ring(),
                           tip="Ratio to the radius")

        self.tol = LParam("tol", (0, 0.1, 0.001), 0.01,
                           updater=lambda p: self.calc_peak(),
                           tip="Ratio to the radius of blurring pixels")

        self.layout((
                self.rmin,
                self.tol,
            ),
            title="FFT Cond.",
            type='vspin', style='button', lw=28, tw=50,
        )

    def init_session(self, session):
        self.reset_params(session.get('params'))

        if self.debug:
            print("$(session) = {!r}".format((session)))
            self.__dict__.update(session)

    def save_session(self, session):
        session['params'] = self.parameters

        if self.debug:
            session.update({
                'axis': self.axis,
                'data': self.data,
                'stig': self.stig,
            })

    @property
    def selected_frame(self):
        return self.graph.frame

    @property
    def selected_roi(self):
        src = self.selected_frame.buffer
        h, w = src.shape
        n = pow(2, int(np.log2(min(h,w)))-1) # resize to 2^n squared ROI
        i, j = h//2, w//2
        return src[i-n:i+n,j-n:j+n]

    @wait
    def calc_ring(self, show=True):
        """Calc log-polar of ring pattern
        """
        frame = self.selected_frame
        src = self.selected_roi
        h, w = src.shape

        buf = fftshift(fft2(src))
        buf = np.log(1 + abs(buf))
        buf -= buf.mean()

        self.message("Calculating CTF ring...")
        r0 = w * self.rmin.value
        r1 = w * 0.5
        dst, self.fitting_curve = find_ring_center(buf, r0, r1, N=256, tol=0.05)

        m = w / np.log(r1/r0)

        self.axis = r0 / w * np.exp(np.arange(w) / m) # [R0:R1] <= [0:1/2]
        self.data = self.fitting_curve.mod1d(dst)

        if show:
            self.message("\b Loading log-polar image...")
            dst = self.fitting_curve.mod2d(dst)
            self.output.load(dst, "*log-polar*", pos=0)
            self.output.load(buf, "*fft-of-{}*".format(frame.name),
                             localunit=1/w/frame.unit)
            del self.Arts
        self.message("\b done.")

        ## 拡張 log-polar 変換は振幅が m 倍だけ引き延ばされている
        eps, phi = self.fitting_curve.params[3:5]
        self.stig = eps / m * np.exp(phi * 1j)
        print("$result(eps, phi) = {!r}".format((eps, phi)))

    @wait
    def calc_peak(self, show=True):
        """Calc min/max peak detection
        """
        N = self.data.size
        R0 = self.rmin.value
        R1 = 0.5
        tol = self.tol.value

        ## r2:data の一定間隔補間データを作ってゼロ点を求める
        newaxis = np.linspace(R0**2, R1**2, N)
        orgdata = np.interp(newaxis, self.axis**2, self.data)
        newdata = smooth1d(orgdata, tol)
        ## if show:
        ##     edi.plot(newaxis, newdata, '-', lw=1) # original smoothing data

        newdata, maxima, minima = find_radial_peaks(newdata, tol)

        ## Check validity of zero-points spacing
        hx, hy = newaxis[maxima], newdata[maxima]
        lx, ly = newaxis[minima], newdata[minima]

        self.lxy = np.vstack((lx, ly))
        self.hxy = np.vstack((hx, hy))
        ## lp = np.vstack((lx, ly))

        ## --------------------------------
        ## filter low peaks
        ## --------------------------------
        threshold = tol/10 * R1**2
        print("$(threshold) = {!r}".format((threshold)))
        lxx = []
        lyy = []
        for i, (x, y) in enumerate(zip(lx, ly)):
            ## Eliminate if near one of high peaks
            if min(abs(x - hx)) < threshold:
                continue
            ## Stop if two low pakes are continuous
            if i < len(lx)-1:
                if not np.any((x < hx) & (hx < lx[i+1])):
                    break
            lxx.append(x)
            lyy.append(y)
        lp = np.vstack((lxx, lyy))

        self.lpoints = lp[:,:20] # max N low peaks
        self.newaxis = newaxis
        self.newdata = newdata

        ## --------------------------------
        ## output results to verify it
        ## --------------------------------

        print("+ {} low peaks found".format(lp.shape[1]))
        if show:
            edi.plot(self.axis**2, self.data, '--', lw=0.5) # raw data
            ## edi.plot(newaxis, orgdata, '--', lw=1) # original data
            edi.plot(newaxis, newdata, '-', lw=1) # interpolated
            edi.plot(lx, ly, 'v') # low peaks
            edi.plot(hx, hy, '^') # high peaks
            edi.plot(*self.lpoints, 'o')    # filtered peaks

            try:
                u = self.output.frame.unit
                eps = np.abs(self.stig)
                ang = np.angle(self.stig) * 180/pi

                ## 不特定多数の円を描画する (最大 N まで)
                del self.Arts
                for x in self.lpoints[0,:10]:
                    r = N * np.sqrt(x)
                    art = patches.Circle((0,0), 0, color='r', ls='--', lw=0.5, fill=0, alpha=0.5)
                    art.width = 2 * r * (1 + eps) * u
                    art.height = 2 * r * (1 - eps) * u
                    art.angle = ang / 2
                    self.attach_artists(self.output.axes, art)
                self.output.draw()
            except Exception:
                self.message("- no data.")

reason: import numpy,from numpy,from scipy


text:
# Implementation of the Collaborative Learning Algorithm

from typing import Optional, List, Tuple, Dict
import random
import math
from copy import deepcopy as dcopy

import numpy as np

from banditpylib.data_pb2 import Feedback, Actions
from banditpylib.arms import PseudoArm
from banditpylib import argmax_or_min_tuple
from banditpylib.learners.mab_fcbai_learner import MABFixedConfidenceBAILearner

from .utils import MABCollaborativeFixedTimeBAIAgent, \
    MABCollaborativeFixedTimeBAIMaster, MABCollaborativeFixedTimeBAILearner

class CentralizedLilUCBHeuristic(MABFixedConfidenceBAILearner):
  """LilUCB heuristic policy :cite:`jamieson2014lil`
  Modified implementation to supplement CollaborativeAgent
  along with additional functionality to work on only a subset of arms

  :param int arm_num: number of arms of the bandit
  :param float confidence: confidence level. It should be within (0, 1). The
    algorithm should output the best arm with probability at least this value.
  :param np.ndarray assigned_arms: arm indices the learner has to work with
  :param str name: alias name
  """
  def __init__(self,
               arm_num: int,
               confidence: float,
               assigned_arms: np.ndarray = None,
               name: str = None):
    assert np.max(assigned_arms) < arm_num and len(assigned_arms) <= arm_num, (
        "assigned arms should be a subset of [arm_num]\nReceived: " +
        str(assigned_arms))
    super().__init__(arm_num=arm_num, confidence=confidence, name=name)
    if assigned_arms is not None:
      self.__assigned_arms = assigned_arms
    else:
      self.__assigned_arms = np.arange(arm_num)

  def _name(self) -> str:
    return 'lilUCB_heur_collaborative'

  def reset(self):
    # create only as many local arms as num_assigned_arms
    # entire algo behaves as if there are just num_assigned_arms in the bandit
    self.__pseudo_arms = [PseudoArm() for arm_id in self.__assigned_arms]
    # Parameters suggested by the paper
    self.__beta = 0.5
    self.__a = 1 + 10 / len(self.__assigned_arms)
    self.__eps = 0
    self.__delta = (1 - self.confidence) / 5
    # Total number of pulls used
    self.__total_pulls = 0
    self.__stage = 'initialization'
    self.__ucb = np.array([0.0] * len(self.__assigned_arms))

  def __confidence_radius(self, pulls: int) -> float:
    """
    Args:
      pulls: number of pulls

    Returns:
      confidence radius
    """
    if (1 + self.__eps) * pulls == 1:
      return math.inf
    return (1 + self.__beta) * (1 + math.sqrt(self.__eps)) * math.sqrt(
        2 * (1 + self.__eps) *
        math.log(math.log((1 + self.__eps) * pulls) / self.__delta) / pulls)

  def __update_ucb(self, arm_id: int):
    """
    Args:
      arm_id: index of the arm whose ucb has to be updated
    """
    self.__ucb[arm_id] = self.__pseudo_arms[arm_id].em_mean +\
      self.__confidence_radius(self.__pseudo_arms[arm_id].total_pulls)

  def actions(self, context=None) -> Actions:
    del context
    if self.__stage == 'initialization':
      actions = Actions()  # default state is normal

      # 1 pull each for every assigned arm
      for arm_id in self.__assigned_arms:
        arm_pull = actions.arm_pulls.add()
        arm_pull.arm.id = arm_id
        arm_pull.times = 1
      return actions

    # self.__stage == 'main'
    actions = Actions()

    for pseudo_arm in self.__pseudo_arms:
      if pseudo_arm.total_pulls >= (
          1 + self.__a * (self.__total_pulls - pseudo_arm.total_pulls)):
        return actions

    arm_pull = actions.arm_pulls.add()

    # map local arm index to the bandits arm index
    arm_pull.arm.id = self.__assigned_arms[int(np.argmax(self.__ucb))]
    arm_pull.times = 1

    return actions

  def update(self, feedback: Feedback):
    for arm_feedback in feedback.arm_feedbacks:
      # reverse map from bandit index to local index
      pseudo_arm_index = np.where(
          self.__assigned_arms == arm_feedback.arm.id)[0][0]
      self.__pseudo_arms[pseudo_arm_index].update(
          np.array(arm_feedback.rewards))
      self.__update_ucb(pseudo_arm_index)
      self.__total_pulls += len(arm_feedback.rewards)

    if self.__stage == 'initialization':
      self.__stage = 'main'

  @property
  def best_arm(self) -> int:
    # map best arm local index to actual bandit index
    return self.__assigned_arms[argmax_or_min_tuple([
        (pseudo_arm.total_pulls, arm_id)
        for (arm_id, pseudo_arm) in enumerate(self.__pseudo_arms)
    ])]

  def get_total_pulls(self) -> int:
    return self.__total_pulls

class LilUCBHeuristicAgent(MABCollaborativeFixedTimeBAIAgent):
  r"""Implementation of agent of the Collaborative Learning Algorithm\
  Uses LilUCBHeuristic as the central algorithm

  :param int arm_num: number of arms of the bandit
  :param int rounds: number of rounds of communication allowed
    (this agent uses one more)
  :param int horizon: maximum number of pulls the agent can make
    (over all rounds combined)
  :param Optional[str] name: alias name
  """
  def __init__(self,
               arm_num: int,
               rounds: int,
               horizon: int,
               name: Optional[str] = None):
    super().__init__(name)
    if arm_num <= 1:
      raise ValueError('Number of arms is expected at least 2. Got %d.' %
                       arm_num)
    if rounds <= 1:
      raise ValueError('Number of rounds is expected at least 2. Got %d.' %
                       rounds)
    self.__arm_num = arm_num
    self.__comm_rounds = rounds - 1
    self.__horizon = horizon
    self.__num_pulls_learning = int(0.5 * horizon / self.__comm_rounds)

  def _name(self) -> str:
    return "lilUCBHeur_collaborative_agent"

  def reset(self):
    self.__round_num = 0
    self.__stage = "unassigned"
    # True if action forwarded from central algo
    self.__central_algo_action_taken = False

  def __complete_round(self):
    self.__round_num += 1
    self.__central_algo_action_taken = False
    if self.__round_num < self.__comm_rounds + 1:
      self.__stage = "unassigned"
    else:
      self.__stage = "termination"

  def set_input_arms(self, arms: List[int]):
    if arms[0] < 0:
      # terminate since there is only one active arm
      self.__learning_arm = arms[1]
      self.__stage = "termination"
      return

    self.__assigned_arms = np.array(arms)
    # confidence of 0.01 suggested in the paper
    self.__central_algo = CentralizedLilUCBHeuristic(self.__arm_num, 0.99,
                                                     self.__assigned_arms)
    self.__central_algo.reset()
    if self.__stage == "unassigned":
      self.__stage = "preparation"

  def actions(self, context=None) -> Actions:
    # a core assumption is all non-empty actions immediately receive feedback
    # and hence stage is changed here and not when feedback is received
    del context

    if self.__stage == "unassigned":
      raise Exception("No arms assigned to agent " + self.name)

    # in preparation:
    #   if only one arm is assigned, proceed to learning
    #   else if central_algo is running, forward its actions
    #   and get best arm when central_algo completes
    #   but interrupt central algo after T/2 pulls
    elif self.__stage == "preparation":
      if len(self.__assigned_arms) == 1:
        self.__stage = "learning"
        self.__learning_arm = self.__assigned_arms[0]
        return self.actions()
      if self.__central_algo.get_total_pulls() >= self.__horizon // 2:
        self.__stage = "learning"
        # use whatever best_arm the central algo outputs
        self.__learning_arm = self.__central_algo.best_arm
        return self.actions()

      central_algo_actions = self.__central_algo.actions()
      if not central_algo_actions.arm_pulls:
        # central algo terminated before T/2 pulls
        self.__stage = "learning"
        self.__learning_arm = self.__central_algo.best_arm
        return self.actions()
      self.__central_algo_action_taken = True
      return central_algo_actions

    # in learning:
    #   if learning_arm is none, do no pulls and move to communication
    #   else pull learning_arm and move to communication
    elif self.__stage == "learning":
      actions = Actions()
      self.__stage = "communication"
      if self.__learning_arm is None:
        actions.state = Actions.WAIT
        return actions
      else:
        arm_pull = actions.arm_pulls.add()
        arm_pull.arm.id = self.__learning_arm  # pylint: disable=protobuf-type-error
        arm_pull.times = self.__num_pulls_learning
        return actions

    elif self.__stage == "communication":
      actions = Actions()
      actions.state = Actions.WAIT
      return actions

    elif self.__stage == "termination":
      actions = Actions()
      actions.state = Actions.STOP
      return actions

    else:
      raise Exception(self.name + ": " + self.__stage +
                      " does not allow actions to be played")

  def update(self, feedback: Feedback):
    self.__learning_mean = None  # default in case learning_arm is None
    num_pulls = 0
    for arm_feedback in feedback.arm_feedbacks:
      num_pulls += len(arm_feedback.rewards)
    if self.__central_algo_action_taken:
      self.__central_algo.update(feedback)
    elif num_pulls > 0:
      # non-zero pulls not by central_algo => learning step was done
      for arm_feedback in feedback.arm_feedbacks:
        if arm_feedback.arm.id == self.__learning_arm:
          self.__learning_mean = np.array(arm_feedback.rewards).mean()
          self.__pulls_used = len(arm_feedback.rewards)
    # else ignore feedback (which is empty)

    self.__central_algo_action_taken = False

  @property
  def best_arm(self) -> int:
    # returns arm that the agent chose (could be None)
    if self.__stage != "termination":
      raise Exception('%s: I don\'t have an answer yet!' % self.name)
    return self.__learning_arm

  def broadcast(self) -> Dict[int, Tuple[float, int]]:
    if self.__stage != "communication":
      raise Exception('%s: I can\'t broadcast in stage %s!'\
        % (self.name, self.__stage))
    return_dict: Dict[int, Tuple[float, int]] = {}
    if self.__learning_arm:
      return_dict[self.__learning_arm] = (self.__learning_mean,
                                          self.__pulls_used)  # type: ignore
    self.__complete_round()
    return return_dict

class LilUCBHeuristicMaster(MABCollaborativeFixedTimeBAIMaster):
  r"""Implementation of master in Collaborative Learning Algorithm

  :param int arm_num: number of arms of the bandit
  :param int rounds: number of rounds of communication allowed
    (this agent uses one more)
  :param int horizon: maximum number of pulls the agent can make
    (over all rounds combined)
  :param int num_agents: number of agents
  :param Optional[str] name: alias name
  """
  def __init__(self,
               arm_num: int,
               rounds: int,
               horizon: int,
               num_agents: int,
               name: Optional[str] = None):
    super().__init__(name)
    if arm_num <= 1:
      raise ValueError('Number of arms is expected at least 2. Got %d.' %
                       arm_num)
    if rounds <= 1:
      raise ValueError('Number of rounds is expected at least 2. Got %d.' %
                       rounds)
    self.__arm_num = arm_num
    self.__comm_rounds = rounds - 1
    self.__T = horizon
    self.__num_agents = num_agents

  def _name(self) -> str:
    return "lilUCBHeur_collaborative_master"

  def reset(self):
    self.__active_arms = list(range(self.__arm_num))

  def __assign_arms(self, agent_ids: List[int]) -> Dict[int, List[int]]:
    agent_arm_assignment: Dict[int, List[int]] = {}

    def random_round(x: float) -> int:
      # if x = 19.3
      # rounded to 19 with probability 0.7
      # rounded to 20 with probability 0.3
      frac_x = x - int(x)
      if np.random.uniform() > frac_x:
        return int(x)
      return int(x) + 1

    if len(self.__active_arms) == 1:
      # use -1 as the first arm if there is only 1 active arm
      agent_arm_assignment = {}
      for agent_id in agent_ids:
        agent_arm_assignment[agent_id] = [-1, self.__active_arms[0]]
      return agent_arm_assignment

    if len(self.__active_arms) < len(agent_ids):
      arms_assign_list = []
      num_agents_per_arm = len(agent_ids) / len(self.__active_arms)

      # match agent to all but the last arm
      for arm in self.__active_arms[:-1]:
        arms_assign_list += [[arm]] * \
          random_round(num_agents_per_arm)

      # match remaining agents to last arm
      arms_assign_list += [[self.__active_arms[-1]]] * \
        (len(agent_ids) - len(arms_assign_list))
      random.shuffle(arms_assign_list)
      for i, agent_id in enumerate(agent_ids):
        agent_arm_assignment[agent_id] = arms_assign_list[i]

    else:
      active_arms_copy = dcopy(self.__active_arms)
      random.shuffle(active_arms_copy)
      for agent_id in agent_ids:
        agent_arm_assignment[agent_id] = []

      # assign atleast 1 arm per agent
      for i, arm in enumerate(active_arms_copy[:len(agent_ids)]):
        agent_arm_assignment[agent_ids[i]].append(arm)

      # assign remaining arms
      num_arms_per_agent = (len(active_arms_copy) - len(agent_ids))\
      / len(agent_ids)
      i = len(agent_ids)
      for agent_id in agent_ids[:-1]:
        if i >= len(active_arms_copy):
          break
        num_arms = random_round(num_arms_per_agent)
        agent_arm_assignment[agent_id] += active_arms_copy[i:i + num_arms]
        i += num_arms
      if i < len(active_arms_copy):
        agent_arm_assignment[agent_ids[-1]] += active_arms_copy[i:]

    return agent_arm_assignment

  def initial_arm_assignment(self) -> Dict[int, List[int]]:
    return self.__assign_arms(list(range(self.__num_agents)))

  def elimination(
      self, messages: Dict[int, Dict[int,
                                     Tuple[float,
                                           int]]]) -> Dict[int, List[int]]:

    aggregate_messages: Dict[int, Tuple[float, int]] = {}
    for agent_id in messages.keys():
      message_from_agent = messages[agent_id]
      for arm_id in message_from_agent:
        if arm_id not in aggregate_messages:
          aggregate_messages[arm_id] = (0.0, 0)
        arm_info = message_from_agent[arm_id]
        new_pulls = aggregate_messages[arm_id][1] + arm_info[1]
        new_em_mean_reward = (aggregate_messages[arm_id][1] * \
         aggregate_messages[arm_id][0] + arm_info[1] * arm_info[0]) / new_pulls
        aggregate_messages[arm_id] = (new_em_mean_reward, new_pulls)

    accumulated_arm_ids = np.array(list(aggregate_messages.keys()))
    accumulated_em_mean_rewards = np.array(
        list(map(lambda x: aggregate_messages[x][0],
                 aggregate_messages.keys())))

    # elimination
    confidence_radius = np.sqrt(
        self.__comm_rounds *
        np.log(200 * self.__num_agents * self.__comm_rounds) /
        (self.__T * max(1, self.__num_agents / len(self.__active_arms))))
    highest_em_reward = np.max(accumulated_em_mean_rewards)
    self.__active_arms = list(
        accumulated_arm_ids[accumulated_em_mean_rewards >= highest_em_reward -
                            2 * confidence_radius])

    return self.__assign_arms(list(messages.keys()))

class LilUCBHeuristicCollaborative(MABCollaborativeFixedTimeBAILearner):
  """Colaborative learners using lilucb heuristic as centralized policy

  :param int num_agents: number of agents
  :param int arm_num: number of arms of the bandit
  :param int rounds: number of rounds of communication allowed
  :param int horizon: maximum number of pulls the agent can make
    (over all rounds combined)
  :param Optional[str] name: alias name
  """
  def __init__(self,
               num_agents: int,
               arm_num: int,
               rounds: int,
               horizon: int,
               name: Optional[str] = None):
    super().__init__(agent=LilUCBHeuristicAgent(arm_num=arm_num,
                                                rounds=rounds,
                                                horizon=horizon),
                     master=LilUCBHeuristicMaster(arm_num=arm_num,
                                                  rounds=rounds,
                                                  horizon=horizon,
                                                  num_agents=num_agents),
                     num_agents=num_agents,
                     name=name)

max_stars_repo_name: sheelfshah/banditpylib

max_issues_repo_name: sheelfshah/banditpylib

max_forks_repo_name: sheelfshah/banditpylib


text:
from copy import deepcopy
import requests
import os
import bs4
from openpyxl import load_workbook
import pandas as pd
from ..helpers.db_funcs import get_ep_id_by_number, get_season_id_by_number_type
from ..helpers.extract_helpers import search_for_new_seasons
import glob
import re
import numpy as np

DOCS_URL_TEMPLATE = 'https://docs.google.com/spreadsheets/d/{id}/export?format=xlsx&id={id}'
SURVIVOR_SOURCE = 'https://www.truedorktimes.com/survivor/boxscores/data.htm'

def create_data_dict(subset=None):
    ret_dict = {}
    sp = bs4.BeautifulSoup(requests.get(SURVIVOR_SOURCE).content)
    cast_elements = sp.find_all('ul', attrs={'class': 'cast'})

    for e in cast_elements:
        attrs = e.find('a').attrs

        try:
            if 'spreadsheet' in attrs['href']:
                v = attrs['href'][:-1].split('/')[-1]
                k = str(e.text.lower())
                for p in '- ':
                    k = k.replace(p, '_')
                for p in ':.-,':
                    k = k.replace(p, '')
                k = k.replace('\n', '')[1:]

                if subset:
                    if k.split('_')[0] not in subset:
                        continue

                ret_dict[k] = v

            else:
                pass
        except KeyError:
            pass

    return ret_dict

def save_survivor_excel(sheets_id, readable_name, dest_folder='../data/raw'):
    url = DOCS_URL_TEMPLATE.format(**dict(id=sheets_id))
    f_name = '{readable_name}.xlsx'.format(readable_name=readable_name)

    req = requests.get(url)

    with open(os.path.join(dest_folder, f_name), 'wb') as f:
        f.write(req.content)
    req.close()

def pull_and_save_excels(data_dict=None, subset=None, dest_folder='../data/raw'):
    if not data_dict:
        data_dict = create_data_dict(subset=subset)
    for k, v in data_dict.items():
        save_survivor_excel(v, k, dest_folder=dest_folder)

# Above is for the actual excels...

def empty_cond(ws, cell, *args, **kwargs):
    return not cell.value

def vertical_cond_tc(ws, cell, *args, **kwargs):

    return empty_cond(ws, cell) or (cell.value == 'wanda')

def any_cond(ws, cell):
    return False

def rc_horizontal_cond(ws, cell, col_names, *args, **kwargs):
    above = ws.cell(row=cell.row - 1, column=cell.column).value
    if isinstance(above, str):
        ic_bool = ('IC' in above) or (
            'Immunity challenge' in above) or ('RC' in above)
        ic_bool = ic_bool and (len(col_names) != 0)
    else:
        ic_bool = False
    return (not cell.value) or (ic_bool)

def ep_horizontal_cond(ws, cell, col_names, nblanks=5, *args, **kwargs):
    add_numbers = [x for x in range(1, nblanks + 1)]
    right_two = [not ws.cell(
        row=cell.row, column=cell.column + i).value for i in add_numbers]
    return all(right_two) and not cell.value

def normal_extract_values(ws, row, column_start, width, *args, **kwargs):
    return pd.Series([ws.cell(row=row, column=column_start + i + 1).value for i in range(width)])

def vote_extract_values(ws, row, column_start, width, col_names, *args, **kwargs):
    values = normal_extract_values(ws, row, column_start, width)
    values = pd.Series([c for i, c in enumerate(
        col_names) if pd.notnull(values[i])])
    return values if len(values) > 0 else pd.Series([None])

def identity_pp(df, col_names, *args, **kwargs):
    df.columns = col_names

    df = df.loc[:, ~df.columns.duplicated()]
    return df

def ep_pp(df, col_names, *args, **kwargs):
    df = identity_pp(df, col_names, *args, **kwargs)
    df = df[df.columns[~df.columns.isna()]]
    return df

def vote_pp(df, col_names, *args, **kwargs):
    if df.shape[1] > 1:
        df = pd.DataFrame(pd.concat([df[col] for col in df]))

    df.columns = ['voted_for']
    df['vote_counted'] = ~df['voted_for'].isna()
    df = df.loc[:, ~df.columns.duplicated()]
    return df

def extract_subtable(ws, start_row, start_column, index_column=None, horizontal_condition=None,
                     vertical_condition=None, extract_values=None, postprocess=None):

    if not horizontal_condition:
        horizontal_condition = empty_cond

    if not vertical_condition:
        vertical_condition = empty_cond

    if not extract_values:
        extract_values = normal_extract_values

    if not postprocess:
        postprocess = identity_pp
    row = start_row
    col = start_column

    col_names = []
    rows = []
    idx = []

    while True:
        cell = ws.cell(row=row, column=col)
        v = cell.value
        if horizontal_condition(ws, cell, col_names):
            break
        else:
            col_names.append(v)
            col += 1

    n_voted_against = len(col_names)
    col -= (n_voted_against + 1)
    row += 1

    while True:
        idx_cell = ws.cell(
            row=row, column=index_column if index_column else col)
        if vertical_condition(ws, idx_cell):
            break
        else:
            values = extract_values(ws, row, col, n_voted_against, col_names)
            rows.append(values)
            idx.append(idx_cell.value)

            row += 1
    df = pd.DataFrame(rows)
    df.index = idx
    df.index.name = 'contestant'

    df = postprocess(df, col_names)

    df.reset_index(inplace=True)

    return df

def extract_rc_challenge(ws, c_row, c_column):
    return extract_subtable(ws, c_row + 1, c_column, horizontal_condition=rc_horizontal_cond, index_column=1)

def extract_ic_challenge(ws, c_row, c_column):
    return extract_subtable(ws, c_row + 1, c_column, horizontal_condition=rc_horizontal_cond, index_column=1)

def extract_tc(ws, c_row, c_column):
    return extract_subtable(ws, c_row + 1, c_column, vertical_condition=vertical_cond_tc, extract_values=vote_extract_values, postprocess=vote_pp, index_column=1)

def extract_ep(ws, c_row, c_column):
    return extract_subtable(ws, c_row, c_column, index_column=1, horizontal_condition=ep_horizontal_cond, postprocess=ep_pp)

def append_tc_data(df, ws, cell):
    v = cell.value
    try:
        total_players = int(re.search('F(\d+)', v).group(1))
    except AttributeError:
        if 'No' in v:
            return pd.DataFrame()
        elif 'Morgan' in v:
            total_players = 10
        elif 'Drake' in v:
            total_players = 9
        elif 'Mokuta' in v:
            total_players = 18
        elif 'Vakama' in v:
            total_players = 17
        else:
            raise
    episode = ws.title[1:]
    new_df = extract_tc(ws, cell.row, cell.column)
    new_df['total_players_remaining'] = total_players
    new_df['episode'] = int(re.match('(\d+).*', episode).group(1))
    df = pd.concat([df, new_df], ignore_index=True)
    return df

def append_challenge_data(df, ws, cell, challenge_type):
    search = re.search('F(\d+)', cell.value)
    if not search:
        # We don't have information about the "final" amount, so we don't fill this in
        final = None
    else:
        final = int(search.group(1))

    if challenge_type == 'RC':
        extract_f = extract_rc_challenge
    elif challenge_type == 'IC':
        extract_f = extract_ic_challenge
    else:
        raise ValueError

    episode = ws.title[1:]
    new_df = extract_f(ws, cell.row, cell.column)
    new_df['total_players_remaining'] = final
    new_df['episode'] = int(re.match('(\d+).*', episode).group(1))
    try:
        df = pd.concat([df, new_df], ignore_index=True)
    except:
        import pdb
        pdb.set_trace()
    return df

def append_episode_data(df, ws, cell):
    episode = ws.title[1:]
    new_df = extract_ep(ws, cell.row, cell.column)
    new_df['episode'] = int(re.match('(\d+).*', episode).group(1))
    df = pd.concat([df, new_df], ignore_index=True)
    return df

def append_rc_data(df, ws, cell):
    return append_challenge_data(df, ws, cell, 'RC')

def append_ic_data(df, ws, cell):
    return append_challenge_data(df, ws, cell, 'IC')

def ic_transform(df, full_name_dict_to_id):
    df['win'] = df['win?']
    if 1 in df:
        df['win'] = df['win?'].fillna(df[1])
    if 'win? ' in df:
        df['win'] = df['win'].fillna(df['win? '])

    if 'sitout' in df:
        df['sitout'] = df['SO'].fillna('sitout')
    else:
        df['sitout'] = df['SO']

    merge_key = df['contestant'].str.replace(
        ' ', '_') + '_' + df['season_id'].astype(str)
    df['contestant_id'] = merge_key.map(full_name_dict_to_id)

    rel_columns = df.columns[df.columns.isin(
        ['win?', 'contestant', 'SO', 'win? '])]
    df.rename(columns={'#people': 'team', 'win%': 'win_pct',
                       'total win%': 'episode_win_pct'}, inplace=True)
    df.drop(columns=rel_columns, inplace=True)

    df = df[df['win'].notnull()].reset_index(drop=True)

    return df

def extract_episodal_data(ws):
    results = {'tribal_council': pd.DataFrame(),
               'reward_challenge': pd.DataFrame(),
               'immunity_challenge': pd.DataFrame(),
               'overall_episode': pd.DataFrame()}

    for row in ws['A1:AZ100']:
        for cell in row:
            if isinstance(cell.value, str):

                if 'Tribal Council voting' in cell.value:
                    # we have to do this first to get the total players/ final N
                    results['tribal_council'] = append_tc_data(
                        results['tribal_council'], ws, cell)

                elif ('RC' in cell.value) and (cell.row > 1) and (len(cell.value) < 25):
                    results['reward_challenge'] = append_rc_data(
                        results['reward_challenge'], ws, cell)

                elif (('IC' in cell.value) or ('Immunity challenge' in cell.value)) and (cell.row > 1) and (len(cell.value) < 25):
                    results['immunity_challenge'] = append_ic_data(
                        results['immunity_challenge'], ws, cell)

                elif (cell.row == 1) and (cell.column == 2):
                    results['overall_episode'] = append_episode_data(
                        results['overall_episode'], ws, cell)

    for challenge_key in ['reward_challenge', 'immunity_challenge']:
        if results[challenge_key].shape[0] == 0:
            continue
        if results[challenge_key]['total_players_remaining'].isna().any():
            try:
                results[challenge_key]['total_players_remaining'] = results['tribal_council']['total_players_remaining'].min()
            except KeyError:
                break

            n_occur = results[challenge_key].groupby(
                'contestant')['episode'].transform(lambda x: np.argsort(x.index))

            if challenge_key == 'immunity_challenge':
                results[challenge_key]['total_players_remaining'] -= n_occur
            else:
                results[challenge_key]['challenge_number'] = n_occur + 1
    return results

def find_data_for_season(wb):
    results = {'tribal_council': pd.DataFrame(),
               'reward_challenge': pd.DataFrame(),
               'immunity_challenge': pd.DataFrame(),
               'overall_episode': pd.DataFrame()}

    sheets = [s for s in wb.worksheets if re.match('e\d+', s.title)]

    for sheet in sheets:
        df_dict = extract_episodal_data(sheet)

        for k, v in df_dict.items():
            current_cols = results[k].columns
            if any(x not in v.columns for x in current_cols):
                v = v.reindex(columns=current_cols)
            try:
                results[k] = pd.concat([results[k], v])
            except:
                import pdb
                pdb.set_trace()
    for k, df in results.items():
        if df.shape[0] == 0:
            continue

        if k != 'overall_episode':
            df['tc_number'] = df['total_players_remaining'].max() + 1 - \
                df['total_players_remaining']

            df.sort_values(by='tc_number', inplace=True)
        df.reset_index(drop=True, inplace=True)
    return results

season_type_map = {
    's': 'Survivor',
    'au': 'Australian Survivor',
    'nz': 'Survivor New Zealand',
    'ssa': 'Survivor South Africa'
}

def extract_episode_stats(eng, data_path=None, asof=None):

    if data_path is None:
        data_path = os.path.join(os.path.dirname(__file__),
                                 '../../../data/raw/truedorks')
    new_seasons = search_for_new_seasons(eng, asof=asof)

    new_seasons_df = pd.read_sql(con=eng,
                                 sql='SELECT * FROM survivor.season').set_index('name').loc[new_seasons]

    search_string = new_seasons_df['type'].map(
        {v: k for k, v in season_type_map.items()}) + new_seasons_df['season_number'].astype(int).astype(str)
    print(search_string)

    subset_dl = [y for y in search_string]
    pull_and_save_excels(None, subset=subset_dl, dest_folder=data_path)

    raw_files = glob.glob('{data_path}/*'.format(data_path=data_path))

    raw_files = [f for f in raw_files if any(
        re.search(y, f) for y in search_string)]
    results = {'tribal_council': pd.DataFrame(),
               'reward_challenge': pd.DataFrame(),
               'immunity_challenge': pd.DataFrame(),
               'overall_episode': pd.DataFrame()}
    for f in raw_files:
        if '~' in f or 'total_voting' in f:
            continue
        print('Extracting data from {path}...'.format(
            path=os.path.realpath(f)))
        wb = load_workbook(f, data_only=True)
        season_dfs = find_data_for_season(wb)
        season_raw = os.path.basename(f).split('_')[0]
        season_parts = re.match('(.*?)(\d+)', season_raw).groups(0)
        season_type = season_type_map[season_parts[0]]
        season_num = int(season_parts[1])

        for k, df in season_dfs.items():

            df['season_id'] = get_season_id_by_number_type(
                eng, season_num, season_type)
            if not df.empty:
                df['episode_id'] = df[['episode', 'season_id']].apply(
                    lambda x: get_ep_id_by_number(eng, x[0], x[1]), axis=1)
            if df['episode_id'].isnull().sum() > 0:
                import pdb
                pdb.set_trace()
            try:
                results[k] = pd.concat([results[k], df])
            except:
                import pdb
                pdb.set_trace()
        print('Extraction complete!')
    return results


text:
import pandas as pd
import numpy as np
from sys import argv, exit

if len(argv) != 2:
    print("python {0} ase_bayes_flagged_counts.csv".format(argv[0]))
    exit(1)

def fix_flags():
    """Fix the flags so that sum > 0 == 1 else 0"""
    df = pd.read_csv(argv[1], index_col=0)
    df.pop("flag_analyze")
    df["flag_analyze"]=np.where(df.sum(axis=1)>0,1,0)
    output = argv[1].split(".csv")[0] + "_re-flagged.csv"
    df.to_csv(output)

if __name__ == "__main__":
    fix_flags()


text:
import numpy as np
import matplotlib.pyplot as plt
from  matplotlib import animation
from time import time

tiempoini = time()      # Inicialización del contador.

"""Implementación del algoritmo de Runge-Kutta de segundo orden para
    ecuaciones diferenciales de segundo orden."""

A = 56                                     # Número másico.
r0 = 1.285                                 # Constante en m.
a = 0.65                                   # Consante en m.
V0 = 47.78                                 # Potencial subcero en MeV.
R = r0*A**(1/3)                            # Constante radial corteza.
Emax = -30
Emin = -40
En = (Emax + Emin)/2                       # Energía MeV.
cte = 0.0483
dE0 = Emax-Emin
Nmax = 100
prec = 1e-10
l = 0
n = 0
dr = 0.01

def Runge2(h,f,g,w,u,r, E): # Método de Runge-Kutta Orden 4 para ec. dif 2º ord.

    k1 = h*f(w,u,r,E)
    k1_g = h*g(w,u,r, E)
    k2 = h*f(w +1/2*k1_g, u +1/2*k1, r +1/2*h, E)
    k2_g = h*g(w +1/2*k1_g, u +1/2*k1, r +1/2*h, E)
    u2 = u + k2
    w2 = w + k2_g
    return w2, u2

def f1(w,u,r, E):
    dudr = w
    return dudr

def f2(w,u,r, E):
    dwdt=-u*0.0483*(E + V0/(1+np.exp((r-R)/a)) -l*(l+1)/(0.0483*2*r**2))
    return dwdt

def norm(array, h):
    tot = 0
    for i in range(len(array)):
        tot += abs(array[i])*h
    return (array)/tot

x = np.linspace(1e-5, R + 15, 1000)      # Array de posiciones.
dx = x[1]-x[0]

c = 0                                      # Contador

while dE0 > prec:
    c += 1
    if c == Nmax:
        break
    u0 = 0                                  # Condición inicial para r*R(r)
    w0 = 1e-5                               # Condición inicial para du/dr
    urunge2 = []
    wrunge2 = []
    urunge2.append(u0)
    wrunge2.append(w0)
    for i in range(len(x)-1):
        sol = Runge2(dx,f1, f2,wrunge2[i], urunge2[i], x[i], En)
        urunge2.append(sol[1])
        wrunge2.append(sol[0])
        if i == len(x)-2:
            if urunge2[-1] > 0:
                Emin = En
                En = Emin + (Emax-Emin)/2
                break
            elif urunge2[-1] < 0:
                Emax = En
                En = Emax - (Emax-Emin)/2
                break
    dE0 = abs(Emax-Emin)
    print(dE0)
tiempo = time()-tiempoini # Tiempo final.
print("Se ha tardado {} segundos.".format(tiempo))
print(En, c)

fig1 = plt.figure()
plt.title("U(r) frente a la coordenada radial")
plt.xlabel("r (fm)")
plt.ylabel("U(r)")
ur2n = norm(urunge2, dx)
plt.plot(x, ur2n, color ="k", label="n = {}". format(n))
plt.legend(loc="best", frameon = False)
plt.xlim(0,max(x))
#plt.ylim(-max(ur2n) - max(ur2n)/10, max(ur2n) + max(ur2n)/10)

fig2 = plt.figure()
plt.title("R(r) frente a la coordenada radial")
plt.xlabel("r (fm)")
plt.ylabel("R(r)")
rr = urunge2/x
rrn = norm(rr[1:], dx)
plt.plot(x[1:], rrn, color="k", label="n = {}". format(n))
plt.legend(loc="best", frameon = False)
plt.xlim(0,max(x))
#plt.ylim(-max(rrn)-max(rrn)/10, max(rrn) + max(rrn)/10)

#np.savetxt("U_Runge2bipart_n2.txt", ur2n)
#np.savetxt("R_Runge2bipart_n2.txt", rrn)

max_stars_repo_path: Por separado/runge2bipart_woods.py

max_stars_repo_name: Cajondesastre-cloud/Woods-Saxon

max_issues_repo_path: Por separado/runge2bipart_woods.py

max_issues_repo_name: Cajondesastre-cloud/Woods-Saxon

max_forks_repo_path: Por separado/runge2bipart_woods.py

max_forks_repo_name: Cajondesastre-cloud/Woods-Saxon


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

# In[1]:

import dask
from dask_kubernetes import KubeCluster
import numpy as np
import pyarrow

# In[2]:

# Specify a remote deployment using a load blanacer if we are running the NB outside of the cluster
#dask.config.set({"kubernetes.scheduler-service-type": "LoadBalancer"})

# In[3]:

cluster = KubeCluster.from_yaml('worker-spec.yaml', namespace='dask') # deploy_mode='remote')

# In[4]:

cluster.adapt(minimum=1, maximum=10)

# In[5]:

# Example usage
from dask.distributed import Client
import dask.array as da

# Connect Dask to the cluster
client = Client(cluster)
client # the repr gives us useful links

# In[6]:

client.scheduler_comm.comm.handshake_info()

# In[7]:

# Create a large array and calculate the mean
array = da.ones((1000, 1000, 1000))
print(array.mean().compute())  # Should print 1.0|

# So now we know the cluster is doing ok :)

# Configure dask to talk to our local MinIO

# In[8]:

# The anon false wasted so much time
minio_storage_options = {
#    "anon": "false",
    "key": "YOURACCESSKEY",
    "secret": "YOURSECRETKEY",
    "client_kwargs": {
        "endpoint_url": "http://minio-1602984784.minio.svc.cluster.local:9000",
        "region_name": 'us-east-1'
    },
    "config_kwargs": {"s3": {"signature_version": 's3v4'}},
}

#tag::minio_storage_options[]
minio_storage_options = {
    "key": "YOURACCESSKEY",
    "secret": "YOURSECRETKEY",
    "client_kwargs": {
        "endpoint_url": "http://minio-1602984784.minio.svc.cluster.local:9000",
        "region_name": 'us-east-1'
    },
    "config_kwargs": {"s3": {"signature_version": 's3v4'}},
}
#end::minio_storage_options[]

# Download the GH archive data

# In[9]:

import datetime
import dask.dataframe as dd

# In[10]:

current_date=datetime.datetime(2020,10,1, 1)

# In[11]:

# In[12]:

#tag::make_file_list[]
gh_archive_files=[]
while current_date < datetime.datetime.now() -  datetime.timedelta(days=1):
    current_date = current_date + datetime.timedelta(hours=1)
    datestring = f'{current_date.year}-{current_date.month:02}-{current_date.day:02}-{current_date.hour}'
    gh_url = f'http://data.githubarchive.org/{datestring}.json.gz'
    gh_archive_files.append(gh_url)
#end::make_file_list[]

# In[13]:

gh_archive_files[0]

# In[14]:

#tag::load_data[]
df = dd.read_json(gh_archive_files, compression='gzip')
df.columns
#end::load_data[]

# In[15]:

len(df)

# In[16]:

# What kind of file systems are supported?
#tag::known_fs[]
from fsspec.registry import known_implementations
known_implementations
#end::known_fs[]
#tag::known_fs_result[]

# In[ ]:

#end::known_fs_result[]

# What does our data look like?

# In[17]:

df.columns

# In[18]:

h = df.head()
h

# In[19]:

import pandas as pd
j = pd.io.json.json_normalize(h.repo)
j

# In[20]:

j.name

# Since we want to partition on the repo name, we need to extract that to it's own column

# In[21]:

data_bag = df.to_bag()

# In[22]:

# The records returned by the bag are tuples not named tuples, so use the df columns to look up the tuple index
cols = df.columns

# In[23]:

def parse_record(record):
    r = {
        "repo": pd.io.json.json_normalize(record[cols.get_loc("repo")]),
        "repo_name": record[cols.get_loc("repo")]["name"],
        "type": record[cols.get_loc("type")],
        "id": record[cols.get_loc("id")],
        "created_at": record[cols.get_loc("created_at")],
        "payload": pd.io.json.json_normalize(record[cols.get_loc("payload")])}
    return r

#tag::cleanup[]
def clean_record(record):
    r = {
        "repo": record[cols.get_loc("repo")],
        "repo_name": record[cols.get_loc("repo")]["name"],
        "type": record[cols.get_loc("type")],
        "id": record[cols.get_loc("id")],
        "created_at": record[cols.get_loc("created_at")],
        "payload": record[cols.get_loc("payload")]}
    return r

cleaned_up_bag = data_bag.map(clean_record)
res = cleaned_up_bag.to_dataframe()
#end::cleanup[]
parsed_bag = data_bag.map(parse_record)

# In[24]:

cleaned_up_bag.take(1)

# In[25]:

res = cleaned_up_bag.to_dataframe()
parsed_res = parsed_bag.to_dataframe()
h = res.head()
h_parsed = parsed_res.head()

# In[26]:

h

# In[27]:

type(h.iloc[0]["repo"])

# In[28]:

type(h_parsed.iloc[0]["repo"])

# In[29]:

#to_csv brings it back locally, lets try parquet. csv doesn't handle nesting so well so use original json inside
df.to_csv("s3://dask-test/boop-test-csv", storage_options=minio_storage_options)

# In[30]:

# This will probably still bring everything back to the client? I'm guessing though.
res.to_parquet("s3://dask-test/boop-test-pq", compression="gzip", storage_options=minio_storage_options, engine="pyarrow")

# In[31]:

parsed_res.to_parquet("s3://dask-test/boop-test-pq-p-nested", compression="gzip", storage_options=minio_storage_options, engine="fastparquet")

# In[ ]:

#tag::write[]
res.to_parquet("s3://dask-test/boop-test-partioned",
              partition_on=["type", "repo_name"], # Based on " there will be no global groupby." I think this is the value we want.
              compression="gzip",
              storage_options=minio_storage_options, engine="pyarrow")
#end::write[]

# In[ ]:

max_stars_repo_name: mlkimmins/scalingpythonml

max_issues_repo_name: mlkimmins/scalingpythonml

max_forks_repo_name: mlkimmins/scalingpythonml


text:
import numpy as np

class TestVMPS(object):