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

import os
import re as reg

from sympy import adjoint, Symbol, expand, conjugate, pi, simplify, transpose

from Definitions import splitPow, Trace

from Logging import loggingCritical

from Substitutions import getSubstitutions, doSubstitutions

class UFOExport():
    def __init__(self, model):
        self._Name = model._Name.replace('-', '').replace('+', '')
        self.model = model
        self.mapping = model.saveSettings['UFOMapping']

        self.string = ""

        # BetaFunc definition
        self.betaFactor = model.betaFactor
        self.kappa = lambda n: 1/(4*pi)**model.betaExponent(n)

        self.translation = {'GaugeCouplings': ('Gauge Couplings', 'gauge'),
                            'Yukawas': ('Yukawa Couplings', 'yuk'),
                            'QuarticTerms': ('Quartic Couplings', 'quartic'),
                            'TrilinearTerms' : ('Trilinear Couplings', 'trilinear'),
                            'ScalarMasses': ('Scalar Mass Couplings', 'scalarMass'),
                            'FermionMasses': ('Fermion Mass Couplings', 'fermionMass'),
                            'Vevs': ('Vacuum-expectation Values', 'vev')}

        self.couplingStructure = {str(model.allCouplings[k][1]): v for k,v in model.couplingStructure.items()}

        self.inconsistentRGset = (model.NonZeroCouplingRGEs != {} or model.NonZeroDiagRGEs != {})
        if self.inconsistentRGset:
            raise TypeError("     -> Error : The RGE set is inconsistent. Please refer to the latex output.")

        try:
            self.UFOsubstitutions(model, inconsistentRGEerror=True)
        except TypeError:
            raise TypeError("     -> Error : The RGE set in the Yukawa sector is inconsistent.")

        self.yukStrucGenerator = False
        self.RGEsByValue = {}
        self.prepareRGEs(model)

        self.preamble(model)
        self.writeRGEs()

    def write(self, folder):
        fileName = os.path.join(folder, 'running.py')
        try:
            self.file = open(fileName, 'w')
        except:
            exit('ERROR while creating the UFO output file')

        self.file.write(self.string)
        self.file.close()

    def preamble(self, model):
        self.string += """### UFO 'running.py' automatically generated by PyR@TE 3 ; using the FeynRules-PyR@TE Interface ###\n
import parameters as P
from object_library import all_running_elements, Running
from function_library import complexconjugate as cc
"""
        if self.yukStrucGenerator:
            self.string += """
############################################################
#     The next few lines help keep the notation compact    #
#  when products or traces of Yukawa matrices are involved #
############################################################

"""
            self.string += "from itertools import product\n\n"

            for k in self.couplingStructure:
                self.string += k + f" = lambda i,j : P.__getattribute__('{k}%dx%d' % (i+1, j+1))\n"

            self.string += """
def generateYukTerm(couplings, indices, ranges, conj=()):
    def conjugate(couplingPos):
        if couplingPos in conj:
            return lambda x: cc(x)
        return lambda x: x

    indsProduct = product(*[range(r) for r in ranges])
    return [couplings[0] + [conjugate(cPos)(c(pInd[i[0]], pInd[i[1]])) for cPos, (c, i) in enumerate(zip(couplings[1], indices))] + couplings[2] for pInd in indsProduct]
"""

    def UFOsubstitutions(self, model, inconsistentRGEerror=True):
        """ Transform the UFO mapping dict into substitutions, and apply them to the RGEs """

        subsDic = {}
        yukMats = {}

        for k,v in self.mapping.items():
            if not '[' in k and not ']' in k:
                subsDic[k] = v
            else:
                # Handle Yukawa-matrix assumptions
                base = k[:k.find('[')]
                inds = eval(k.replace(base, ''))

                if base not in yukMats:
                    if base not in self.couplingStructure:
                        loggingCritical(f"Error : coupling matrix '{base}' is unknown.")
                        continue
                    yukMats[base] = (self.couplingStructure[base], {})

                if v != '0':
                    yukMats[base][1][tuple(inds)] = v

        for k, (shape, dic) in yukMats.items():
            matList = [[dic[(i,j)] if (i,j) in dic else 0 for j in range(shape[1])] for i in range(shape[0])]
            subsDic[k] = str(matList).replace("\\'", "'")

        subsDic = getSubstitutions(model, subsDic)
        doSubstitutions(model, subsDic, inconsistentRGEerror=inconsistentRGEerror)

    def prepareRGEs(self, model):
        """ Gather all the terms of the RGES by coupling type and by value of the numerical coeff """
        self.indexCount = 0

        def wCoeff(coeff):
            return reg.sub(r'(?<!\*\*)(\d+)',r'\1.', str(coeff), 1).replace('pi', 'cmath.pi')

        def wSymb(symb):
            s = str(symb)
            if 'conjugate' in s:
                return 'cc(P.' + s[10:]
            return 'P.' + s

        def listToString(l):
            return '[' + (wSymb(l[0]) if len(l) > 0 else '') + (', ' + ','.join([wSymb(el) for el in l[1:]]) if len(l) > 1 else '') + ']'

        for cType, loopDic in model.couplingRGEs.items():
            if 'Anomalous' in cType:
                continue

            if cType not in self.RGEsByValue:
                self.RGEsByValue[cType] = {}

            for nLoop, RGEdic in loopDic.items():
                for c, RGE in RGEdic.items():
                    RGE = expand(RGE/self.betaFactor * self.kappa(nLoop+1))
                    termsAdd = RGE.as_coeff_add()[1]

                    for term in termsAdd:
                        symbs = []
                        mats = []
                        traces = []
                        coeff = 1

                        for el in splitPow(term.args):
                            if not el.is_commutative:
                                mats.append(el)
                            elif isinstance(el, Trace):
                                traces.append(el)
                            elif not el.is_number:
                                symbs.append(el)
                            else:
                                coeff *= el

                        coeff = wCoeff(simplify(coeff))
                        if coeff not in self.RGEsByValue[cType]:
                            self.RGEsByValue[cType][coeff] = []

                        if mats == traces == []:
                            self.RGEsByValue[cType][coeff].append(listToString([c] + symbs))
                        else:
                            (pre, symbs, post), inds, ranges, cjs = self.buildIndexStructure(c, coeff, mats, traces, symbs)
                            self.RGEsByValue[cType][coeff].append('*generateYukTerm(['
                                                                    +listToString(pre)
                                                                    +', '+str(symbs)
                                                                    +', '+listToString(post)+']'
                                                                  +', '+str(inds)
                                                                  +', '+str(ranges)
                                                                  +(', '+str(cjs) if cjs != () else '')
                                                                  + ')')

    def generateIndex(self):
        self.indexCount += 1
        return self.indexCount - 1

    def buildIndexStructure(self, c, coeff, mats, traces, symbs):
        if self.yukStrucGenerator is False:
            self.yukStrucGenerator = True

        self.indexCount = 0
        allIndices = []
        allSymbs = []
        pre = []
        if c in self.couplingStructure:
            indices = (self.generateIndex(), self.generateIndex())
            allIndices = [tuple(indices)]
            allSymbs = [Symbol(c)]
        else:
            pre = [str(c)]

        conjs = []
        iPrev = ''

        for p, mat in enumerate(mats):

            if len(mats) == 1:
                inds = indices
            elif p == 0:
                iPrev = self.generateIndex()
                inds = [indices[0], iPrev]
            elif p+1 == len(mats):
                inds = [iPrev, indices[1]]
            else:
                tmp, iPrev = iPrev, self.generateIndex()
                inds = [tmp, iPrev]

            if isinstance(mat, transpose) or isinstance(mat, adjoint):
                inds = inds[::-1]
            if isinstance(mat, conjugate) or isinstance(mat, adjoint):
                conjs.append(len(allSymbs))

            symb = mat
            if isinstance(mat, transpose) or isinstance(mat, adjoint) or isinstance(mat, conjugate):
                symb = symb.args[0]

            allIndices.append(tuple(inds))
            allSymbs.append(symb)

        for tr in traces:
            iFirst = ''
            trMats = tr.args[0].args
            for p, mat in enumerate(trMats):
                if p == 0:
                    iFirst = self.generateIndex()
                    iPrev = self.generateIndex()
                    inds = [iFirst, iPrev]
                elif p+1 == len(trMats):
                    inds = [iPrev, iFirst]
                else:
                    tmp, iPrev = iPrev, self.generateIndex()
                    inds = [tmp, iPrev]

                if isinstance(mat, transpose) or isinstance(mat, adjoint):
                    inds = inds[::-1]
                if isinstance(mat, conjugate) or isinstance(mat, adjoint):
                    conjs.append(len(allSymbs))

                symb = mat
                if isinstance(mat, transpose) or isinstance(mat, adjoint) or isinstance(mat, conjugate):
                    symb = symb.args[0]

                allIndices.append(tuple(inds))
                allSymbs.append(symb)

        # Now determine the ranges
        ranges = {}
        for p, t in enumerate(allIndices):
            if t == ():
                continue

            c = str(allSymbs[p])
            struc = self.couplingStructure[c]

            for iPos, iRange in zip(t, struc):
                if iPos not in ranges:
                    ranges[iPos] = iRange
                else:
                    if iRange != ranges[iPos]:
                        loggingCritical("Error : found two contracted flavor indices with different ranges ...")
                        exit()

        ranges = tuple([ranges[i] for i in range(len(ranges))])

        return [pre, allSymbs, [str(el) for el in symbs]], allIndices, ranges, tuple(conjs)

    def writeRGEs(self):

        for cType, dic in self.RGEsByValue.items():
            self.string += '\n\n'
            strType = self.translation[cType][0]
            self.string += '#' * (len(strType)+4) + '\n'
            self.string += '# ' + self.translation[cType][0] + ' #\n'
            self.string += '#' * (len(strType)+4) + '\n'

            tName = ''
            pad1 = 0
            pad2 = 0
            for i, (val, terms) in enumerate(dic.items()):
                tName = 'R' + self.translation[cType][1] + str(i+1)
                pad1 = len(tName) + 11
                pad2 = pad1 + 16
                self.string += '\n' + tName + " = Running(name        = '" + tName + "',\n"
                self.string += ' '*pad1 + 'run_objects = [\n'

                for j, term in enumerate(terms):
                    self.string += pad2*' ' + term + (',' if j+1 != len(terms) else '') + '\n'

                self.string += ' '*(pad2-2) + '],\n'
                self.string += ' '*pad1 + 'value = ' + val + ')\n'

max_stars_repo_name: NuxDD/pyrate

max_issues_repo_name: NuxDD/pyrate

max_forks_repo_name: NuxDD/pyrate

reason: from sympy


text:
#
# Collective Knowledge ()
#
#
#
#
# Developer:
#

cfg={}  # Will be updated by CK (meta description of this module)
work={} # Will be updated by CK (temporal data)
ck=None # Will be updated by CK (initialized CK kernel)

import os
import sys
import time

import pandas as pd
import numpy as np

#default_repo_uoa = ''
#default_repo_uoa = 'local'
default_repo_uoa = 'ck-quantum-hackathon-20190315'

##############################################################################
# Initialize module

def init(i):
    """

    Input:  {}

    Output: {
              return       - return code =  0, if successful
                                         >  0, if error
              (error)      - error text if return > 0
            }

    """
    return {'return':0}

##############################################################################
# get raw data for repo-widget

def get_raw_data(i):
    """
    Input:  {
            }

    Output: {
              return       - return code =  0, if successful
                                         >  0, if error
              (error)      - error text if return > 0
            }

    """

    selected_repo_uoa = i.get('repo_uoa', default_repo_uoa)

    def get_experimental_results(repo_uoa=selected_repo_uoa, tags='qck,hackathon-20190315', module_uoa='experiment'):
        r = ck.access({'action':'search', 'repo_uoa':repo_uoa, 'module_uoa':module_uoa, 'tags':tags})
        if r['return']>0:
            print('Error: %s' % r['error'])
            exit(1)
        experiments = r['lst']

        index = [
            'team', 'problem_index'
        ]

        dfs = []
        for experiment in experiments:
            data_uoa = experiment['data_uoa']
            r = ck.access({'action':'list_points', 'repo_uoa':repo_uoa, 'module_uoa':module_uoa, 'data_uoa':data_uoa})
            if r['return']>0:
                print('Error: %s' % r['error'])
                exit(1)

            # Get all the parameters from meta.json -> "meta"
            mmeta                   = r['dict']['meta']
            team                    = mmeta.get('team', 'UNKNOWN_TEAM')

            experiment_entry_path   = r['path']
            entry_modification_epoch_secs   = int( os.path.getmtime(experiment_entry_path) )
            entry_modification_utc_human    = time.asctime(time.gmtime( entry_modification_epoch_secs ))

            point_ids = r['points']

            for point_id in point_ids:
                load_point_adict = {    'action':           'load_point',
                                        'module_uoa':       module_uoa,
                                        'data_uoa':         data_uoa,
                                        'point':            point_id,
                }
                r=ck.access( load_point_adict )
                if r['return']>0: return r

                if i.get('out')=='con':
                    ck.out( "Loading  {}:experiment:{}  point_id={} (recorded {})".format(repo_uoa, data_uoa, point_id, entry_modification_utc_human) )

                point_data_raw = r['dict']['0001']
                choices = point_data_raw['choices']
                characteristics_list = point_data_raw['characteristics_list']
                num_repetitions = len(characteristics_list)

                data = [
                    {
                        # statistical repetition
                        'repetition_id': repetition_id,
                        # runtime characteristics
                        'problem_name': characteristics['run'].get('problem_name','problem_x'),
                        'problem_index': characteristics['run'].get('problem_index',-1),
                        'training_accuracy': np.float64(characteristics['run'].get('training_accuracy',1e6)),
                        'training_time': np.float64(characteristics['run'].get('training_time',0.0)),
                        'training_vectors_limit': np.int64(characteristics['run'].get('training_vectors_limit') or -1),
                        'solution_function_name': characteristics['run'].get('solution_function_name',''),
                        'source_code': characteristics['run'].get('source_code',''),
                        'circuit_str': characteristics['run'].get('circuit_str',''),
                        'test_accuracy': np.float64(characteristics['run'].get('test_accuracy',0.0)),

                        'team': team,
                        'timestamp_epoch_secs': entry_modification_epoch_secs,
                        'timestamp_utc_human': entry_modification_utc_human,

                        'success?': characteristics['run'].get('run_success','N/A'),
                    }
                    for (repetition_id, characteristics) in zip(range(num_repetitions), characteristics_list)
                    if len(characteristics['run']) > 0
                ]

                # Construct a DataFrame.
                df = pd.DataFrame(data)
                df = df.set_index(index, drop=False)
                # Append to the list of similarly constructed DataFrames.
                dfs.append(df)
        if dfs:
            # Concatenate all thus constructed DataFrames (i.e. stack on top of each other).
            result = pd.concat(dfs)
            result.sort_index(ascending=True, inplace=True)
        else:
            # Construct a dummy DataFrame the success status of which can be safely checked.
            result = pd.DataFrame(columns=['success?'])

        return result

    # prepare table
    df = get_experimental_results()

    if not df.empty:
        df.reset_index(inplace=True, drop=True)     # remove the index as it is in the way of complex grouping

        ## Sorting in place allows us to preserve this order as the initial order in the output
        df.sort_values(['problem_index', 'test_accuracy', 'timestamp_epoch_secs'], ascending=[True, False, True], inplace=True)
        df['rank'] = df.groupby('problem_index').cumcount()+1

        df['seconds_since_start']   = df['timestamp_epoch_secs']-df['timestamp_epoch_secs'].min()
        df['hours_since_start']     = df['seconds_since_start'] / 3600.0

#    from IPython.display import display
#    pd.options.display.max_columns = len(df.columns)
#    pd.options.display.max_rows = len(df.index)
#    display(df)

    table = []

    def to_value(i):
        if type(i) is np.ndarray:
            return i.tolist()

        if isinstance(i, np.int64):
            return int(i)

        if isinstance(i, np.float64):
            return float(i)

        return i

    props = [
        'problem_name',
        'problem_index',
        'training_vectors_limit',
        'training_time',
        'training_accuracy',
        'solution_function_name',
        'source_code',
        'circuit_str',
        'test_accuracy',
        'rank',

        'team',
        'timestamp_epoch_secs',
        'timestamp_utc_human',
        'seconds_since_start',
        'hours_since_start',

        'success?',
    ]

    for record in df.to_dict(orient='records'):
        row = {}
        for prop in props:
            row[prop] = to_value(record.get(prop, ''))

#        row['##data_uid'] = "{}:{}".format(record['_point'], record['_repetition_id'])

        row['source_code'] = {
            'title': record.get('solution_function_name','Show source'),
            'cmd': record['source_code'],
        }

        if record['circuit_str']:
            row['circuit_str'] = {
                'title': record.get('solution_circuit_name', 'Show circuit'),
                'cmd': record['circuit_str'],
            }
        else:
            row['circuit_str']='N/A'

        table.append(row)

    return {'return':0, 'table':table}

##############################################################################
# get raw config for repo widget

def get_raw_config(i):
    """
    Input:  {
            }

    Output: {
              return       - return code =  0, if successful
                                         >  0, if error
              (error)      - error text if return > 0
            }

    """

    data_config = cfg['data_config']
    data_config['return'] = 0

    return data_config

ext: py

max_stars_repo_name: ctuning/ck-quantum

max_issues_repo_name: ctuning/ck-quantum

max_forks_repo_name: ctuning/ck-quantum

reason: import numpy


text:
"""
This module does the heavy lifting and is responsible for converting the input image into a low-poly stylized image.
"""
import cv2
from PIL import Image
import numpy as np
import time
from scipy import spatial

# Data flow
# pre-process image (Image) -> Image
# get polygons from image (Image) -> List[Polygon]
# shade polygons (List[Point], Image) -> Image
# post-process image (image) -> Image

class LowPolyfier:
    def __init__(self, **kwargs):
        self.options = {k: v for k, v in kwargs.items() if v is not None}

    @staticmethod
    def rescale_image(image: np.ndarray, longest_edge: int):
        height, width = image.shape[:2]
        if height >= width:
            ratio = longest_edge / height
            new_height = longest_edge
            new_width = int(width * ratio)
        else:
            ratio = longest_edge / width
            new_width = longest_edge
            new_height = int(height * ratio)
        return cv2.resize(image, (new_width, new_height))

    def preprocess(self, image: np.ndarray) -> np.ndarray:
        if image.shape[-1] == 4:
            float_image = image.astype(np.float64) / 255.0
            float_image[..., :3] *= float_image[..., -1, np.newaxis]
            float_image = float_image[..., :3]
            image = (float_image * 255).astype(np.uint8)
        image = image[..., ::-1]
        longest_edge = self.options["longest_edge"]
        if longest_edge != "auto":
            image = self.rescale_image(image, longest_edge)
        # image = cv2.bilateralFilter(image, 9, 200, 200)
        return image

    @staticmethod
    def visualize_image(image: np.ndarray, name=None):
        name = name or str(time.time())
        cv2.imshow(name, image)
        cv2.waitKey(0)

    def visualize_points(self, image: np.ndarray, points: np.ndarray, name=None):
        image = image.copy() // 2
        points = (*np.squeeze(np.split(points, 2, axis=1))[::-1],)
        image[points] = (0, 255, 0)
        self.visualize_image(image, name=name)

    def visualize_canny(self, image: np.ndarray, canny: np.ndarray):
        image = image.copy()
        image[np.where(canny)] = (0, 255, 0)
        self.visualize_image(image, name="canny")

    @staticmethod
    def get_random_points(image: np.ndarray, num_points=100) -> np.ndarray:
        if num_points <= 0:
            return np.zeros((0, 2), dtype=np.uint8)
        return np.array(
            [
                np.random.uniform((0, 0), image.shape[:2]).astype(np.int32)[::-1]
                for _ in range(num_points)
            ]
        )

    def get_canny_points(
        self, image: np.ndarray, low_thresh: int, high_thresh: int, num_points=500
    ) -> np.ndarray:
        if num_points <= 0:
            return np.zeros((0, 2), dtype=np.uint8)
        canny = cv2.Canny(image, low_thresh, high_thresh)
        if self.options["visualize_canny"]:
            self.visualize_canny(image, canny)
        canny_points = np.argwhere(canny)[..., ::-1]
        step_size = len(canny_points) // num_points
        canny_points = canny_points[::step_size]
        return canny_points

    def get_laplace_points(self, image: np.ndarray, num_points=500) -> np.ndarray:
        if num_points <= 0:
            return np.zeros((0, 2), dtype=np.uint8)
        image = cv2.GaussianBlur(image, (15, 15), 0)
        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        image = np.uint8(np.absolute(cv2.Laplacian(image, cv2.CV_64F, 19)))
        image = cv2.GaussianBlur(image, (15, 15), 0)
        image = (image * (255 / image.max())).astype(np.uint8)
        image = image.astype(np.float32) / image.sum()
        if self.options["visualize_laplace"]:
            self.visualize_image(image, "laplace")
        weights = np.ravel(image)
        coordinates = np.arange(0, weights.size, dtype=np.uint32)
        choices = np.random.choice(
            coordinates, size=num_points, replace=False, p=weights
        )
        raw_points = np.unravel_index(choices, image.shape)
        points = np.stack(raw_points, axis=-1)[..., ::-1]
        return points

    def randomize_points(
        self, image: np.ndarray, points: np.ndarray, ratio: float
    ) -> None:
        num_random_points = int(len(points) * ratio)
        if num_random_points:
            replace_indices = np.random.uniform(
                0, len(points), num_random_points
            ).astype(np.int32)
            points[replace_indices] = self.get_random_points(
                image, num_points=num_random_points
            )

    def jiggle_keypoints(
        self, image: np.ndarray, points: np.ndarray, jiggle_ratio: float = 0.01
    ) -> None:
        ratios = np.random.uniform(-jiggle_ratio, jiggle_ratio, points.shape)
        jiggles = (image.shape[:2] * ratios).astype(np.int32)
        points += jiggles
        self.constrain_points(image.shape[:2], points)

    def get_keypoints(self, image: np.ndarray, include_corners=True) -> np.ndarray:
        canny_low_threshold = self.options["canny_low_threshold"]
        canny_high_threshold = self.options["canny_high_threshold"]
        random_replace_ratio = self.options["random_replace_ratio"]
        num_random_points = self.options["num_random_points"]
        num_canny_points = self.options["num_canny_points"]
        num_laplace_points = self.options["num_laplace_points"]
        jiggle_ratio = self.options["jiggle_ratio"]
        # TODO: add points to image edges
        canny_points = self.get_canny_points(
            image,
            canny_low_threshold,
            canny_high_threshold,
            num_points=num_canny_points,
        )
        laplace_points = self.get_laplace_points(image, num_points=num_laplace_points)
        random_points = self.get_random_points(image, num_random_points)
        points = np.concatenate((canny_points, laplace_points, random_points))
        self.randomize_points(image, points, random_replace_ratio)
        self.jiggle_keypoints(image, points, jiggle_ratio=jiggle_ratio)
        if include_corners:
            corners = np.array(
                [
                    (0, 0),  # top left
                    (image.shape[1] - 1, 0),  # top right
                    (0, image.shape[0] - 1),  # bottom left
                    (image.shape[1] - 1, image.shape[0] - 1),  # bottom right
                ]
            )
            points = np.concatenate((points, corners))
        if self.options["visualize_points"]:
            self.visualize_points(image, points)
        return points

    @staticmethod
    def validate_polygons(max_dims: tuple, polygons):
        height, width = max_dims
        if isinstance(polygons, np.ndarray):
            assert len(polygons.shape) == 3
            assert polygons.shape[1] >= 3
            assert polygons.shape[2] == 2
        elif isinstance(polygons, list):
            raise NotImplementedError(
                "Expressing polygons as a list is currently unsupported. "
                "Convert it to a numpy array instead."
            )
        try:
            assert polygons[..., 0].min() >= -1
            assert polygons[..., 1].min() >= -1
            assert polygons[..., 0].max() < width
            assert polygons[..., 1].max() < height
        except AssertionError as e:
            print(
                "ERROR: Some polygon coordinates are out of image bounds. This is a bug. Please report this."
            )
            print(
                "(Xmin, Ymin, Xmax, Ymax) = ({}, {}, {}, {})".format(
                    polygons[..., 0].min(),
                    polygons[..., 1].min(),
                    polygons[..., 0].max(),
                    polygons[..., 1].max(),
                )
            )
            print("Max dims: {}".format(max_dims))
            raise e

    @staticmethod
    def constrain_points(max_dims: tuple, points: np.ndarray):
        height, width = max_dims
        if isinstance(points, np.ndarray):
            points[..., 0][points[..., 0] < 0] = 0
            points[..., 0][points[..., 0] >= width] = width - 1
            points[..., 1][points[..., 1] < 0] = 0
            points[..., 1][points[..., 1] >= height] = height - 1
        elif isinstance(points, list):
            raise NotImplementedError(
                "Expressing points as a list is currently unsupported."
            )
        return points

    @staticmethod
    def finitize_voronoi(vor, radius=None):
        """
        Reconstruct infinite voronoi regions in a 2D diagram to finite regions.

        Args:
            vor:
            radius:

        Returns:

        """

        if vor.points.shape[1] != 2:
            raise ValueError("Requires 2D input")

        new_regions = []
        new_vertices = vor.vertices.tolist()

        center = vor.points.mean(axis=0)
        if radius is None:
            radius = vor.points.ptp().max()

        # Construct a map containing all ridges for a given point
        all_ridges = {}
        for (p1, p2), (v1, v2) in zip(vor.ridge_points, vor.ridge_vertices):
            all_ridges.setdefault(p1, []).append((p2, v1, v2))
            all_ridges.setdefault(p2, []).append((p1, v1, v2))

        # Reconstruct infinite regions
        for p1, region in enumerate(vor.point_region):
            vertices = vor.regions[region]

            if all(v >= 0 for v in vertices):
                # finite region
                new_regions.append(vertices)
                continue

            # reconstruct a non-finite region
            ridges = all_ridges[p1]
            new_region = [v for v in vertices if v >= 0]

            for p2, v1, v2 in ridges:
                if v2 < 0:
                    v1, v2 = v2, v1
                if v1 >= 0:
                    # finite ridge: already in the region
                    continue

                # Compute the missing endpoint of an infinite ridge
                t = vor.points[p2] - vor.points[p1]  # tangent
                t /= np.linalg.norm(t)
                n = np.array([-t[1], t[0]])  # normal

                midpoint = vor.points[[p1, p2]].mean(axis=0)
                direction = np.sign(np.dot(midpoint - center, n)) * n
                far_point = vor.vertices[v2] + direction * radius

                new_region.append(len(new_vertices))
                new_vertices.append(far_point.tolist())

            # sort region counterclockwise
            vs = np.asarray([new_vertices[v] for v in new_region])
            c = vs.mean(axis=0)
            angles = np.arctan2(vs[:, 1] - c[1], vs[:, 0] - c[0])
            new_region = np.array(new_region)[np.argsort(angles)]

            # finish
            new_regions.append(new_region.tolist())

        return new_regions, np.array(new_vertices, dtype=np.int32)

    def get_delaunay_triangles(self, image: np.ndarray):
        points = self.get_keypoints(image)
        delaunay = spatial.Delaunay(points)
        return np.array(
            [[points[i] for i in simplex] for simplex in delaunay.simplices]
        )

    def get_voronoi_polygons(self, image: np.ndarray):
        points = self.get_keypoints(image, include_corners=False)
        voronoi = spatial.Voronoi(points)
        regions, vertices = self.finitize_voronoi(voronoi)
        self.constrain_points(image.shape[:2], vertices)
        max_polygon_points = len(max(regions, key=lambda x: len(x)))
        polygons = np.full((len(regions), max_polygon_points, 2), -1)
        for i, region in enumerate(regions):
            polygon_points = vertices[region]
            polygons[i][: polygon_points.shape[0]] = polygon_points
        return polygons

    def get_polygons(self, image: np.ndarray) -> np.ndarray:
        polygon_method = self.options["polygon_method"]
        if polygon_method == "delaunay":
            polygons = self.get_delaunay_triangles(image)
        elif polygon_method == "voronoi":
            polygons = self.get_voronoi_polygons(image)
        else:
            raise ValueError(f"Unknown Polygonization method '{polygon_method}'")
        self.validate_polygons(image.shape[:2], polygons)
        return polygons

    def get_output_dimensions(self, image):
        longest_edge = self.options["output_size"]
        ratio = image.shape[1] / image.shape[0]
        return int(longest_edge / ratio), longest_edge, 3

    @staticmethod
    def strip_negative_points(polygon):
        return polygon[polygon.min(axis=1) >= 0]

    @staticmethod
    def get_dominant_color(pixels, clusters, attempts):
        """
        Given a (N, Channels) array of pixel values, compute the dominant color via K-means
        """
        clusters = min(clusters, len(pixels))
        flags = cv2.KMEANS_RANDOM_CENTERS
        criteria = (cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_MAX_ITER, 1, 10)
        _, labels, centroids = cv2.kmeans(
            pixels.astype(np.float32), clusters, None, criteria, attempts, flags
        )
        _, counts = np.unique(labels, return_counts=True)
        dominant = centroids[np.argmax(counts)]
        return dominant

    def shade(self, polygons: np.ndarray, image: np.ndarray) -> np.ndarray:
        canvas_dimensions = self.get_output_dimensions(image)
        scale_factor = max(canvas_dimensions) / max(image.shape)
        scaled_polygons = polygons * scale_factor
        output_image = np.zeros(canvas_dimensions, dtype=np.uint8)
        for polygon, scaled_polygon in zip(polygons, scaled_polygons):
            polygon = self.strip_negative_points(polygon)
            scaled_polygon = self.strip_negative_points(scaled_polygon)
            if len(polygon) < 3:
                continue
            mask = np.zeros(image.shape[:2], dtype=np.uint8)
            cv2.fillConvexPoly(mask, polygon, (255,))
            color = self.get_dominant_color(image[mask > 0], 3, 3).tolist()
            # color = cv2.mean(image, mask)[:3]
            cv2.fillConvexPoly(
                output_image,
                scaled_polygon.astype(np.int32),
                color,
                lineType=cv2.LINE_AA,
            )
        return output_image

    @staticmethod
    def saturation(image: np.ndarray, alpha: float, beta: float):
        hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
        hsv[..., 1] = cv2.convertScaleAbs(hsv[..., 1], alpha=alpha, beta=beta)
        image = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)
        return image

    @staticmethod
    def brightness_contrast(image: np.ndarray, alpha, beta):
        image = cv2.convertScaleAbs(image, alpha=alpha, beta=beta)
        return image

    def postprocess(self, image: np.ndarray) -> np.ndarray:
        saturation = self.options["post_saturation"]
        contrast = self.options["post_contrast"]
        brightness = self.options["post_brightness"]
        image = self.saturation(image, 1 + saturation, 0)
        image = self.brightness_contrast(image, 1 + contrast, 255 * brightness)
        return image

    def lowpolyfy(self, image):
        image = np.array(image).copy()
        pre_processed_image = self.preprocess(image)
        polygons = self.get_polygons(pre_processed_image)
        shaded_image = self.shade(polygons, pre_processed_image)
        post_processed_image = self.postprocess(shaded_image)
        return Image.fromarray(post_processed_image[..., ::-1])

max_stars_repo_name: acashjos/lowpolypy

max_issues_repo_name: acashjos/lowpolypy

max_forks_repo_name: acashjos/lowpolypy

reason: import numpy,from scipy


text:
import pylab as plt
from matplotlib.transforms import Affine2D
import numpy as np
import glob

def main():
    data_files = sorted(glob.glob("*D_n*.npz"))
    ndims, num_live_points = [],[]
    run_times, logZs, logZerrs, true_logZs = [],[],[],[]
    for file in data_files:
        print("parsing",file)
        ndims.append(int(file.split("D")[0]))
        num_live_points.append(int(file.split("_n")[-1].split('.')[0]))
        run_time, logZ, logZerr = np.load(file)['run_data']
        run_times.append(run_time)
        logZs.append(logZ)
        logZerrs.append(logZerr)
        true_logZs.append(np.load(file)['true_logZ'])
    ndims = np.array(ndims)
    a = np.argsort(ndims)

    run_times = np.array(run_times)
    true_logZs = np.array(true_logZs)
    logZs = np.array(logZs)
    logZerrs = np.array(logZerrs)

    names = ['Dynesty', 'PolyChord', 'MultiNEST', 'JAXNS']
    colors = ['blue','green','red','black']
    markers = ['o','^', 'v', '*']
    linestyles = ['solid', 'dotted', 'dashed', 'dashdot']

    fig, axs = plt.subplots(2,1, sharex=True, figsize=(5,5))
    for i, name in enumerate(names):
        axs[0].plot(ndims[a], run_times[a,i], c=colors[i], marker=markers[i], ls=linestyles[i], label=name)
        if name == 'JAXNS':
            alt_settings = [5, 6]
            select = [list(ndims).index(j) for j in alt_settings]
            axs[0].scatter(alt_settings, run_times[select,i], s=150,
                           facecolors='none', ec='black')

    axs[0].set_yscale('log')
    axs[0].set_ylabel('Run time [s]')
    axs[0].grid()
    axs[0].legend()

    axs[1].hlines(0., np.min(ndims), np.max(ndims), linestyles='dotted', colors='grey')
    transforms = [Affine2D().translate(-0.1, 0.0) + axs[1].transData,
                  Affine2D().translate(-0.05, 0.0) + axs[1].transData,
                  Affine2D().translate(0.0, 0.0) + axs[1].transData,
                  Affine2D().translate(0.05, 0.0) + axs[1].transData]
    for i, name in enumerate(names):
        axs[1].errorbar(ndims[a], logZs[a,i] - true_logZs[a], logZerrs[a,i], c=colors[i], marker=markers[i],
                     transform=transforms[i], ls=linestyles[i], label=name)
        if name == 'JAXNS':
            alt_settings = [5,6]
            select = [list(ndims).index(j) for j in alt_settings]
            axs[1].scatter(alt_settings, logZs[select, i] - true_logZs[select], transform=transforms[i], s=150, facecolors='none', ec='black')
    axs[1].set_ylabel(r'$\Delta \log Z$')
    axs[1].set_xlabel('Prior dimension')
    axs[1].legend()
    plt.tight_layout()
    fig.subplots_adjust(hspace=0)
    plt.savefig('performance_second_model.pdf')
    plt.show()

if __name__ == '__main__':
    main()


text:
from datetime import timedelta

import numpy
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.mixins import LoginRequiredMixin
from django.shortcuts import render, redirect, reverse, get_object_or_404
from django.utils import timezone
from django.utils.safestring import mark_safe
from django.views.generic import TemplateView, View, RedirectView

from braindump.models import CardPlacement
from categories.models import Category, ShareContract

class BraindumpViewMixin:
    """Mixin for Braindump views
    """
    def handle_query_string(self, request):
        """Handle query string
        """
        min_area, max_area = self.validate_min_max_area(request)
        query_string = self.generate_min_max_area_query_string(min_area, max_area)
        if query_string:
            query_string = '?{}'.format(query_string)

        return query_string

    def validate_min_max_area(self, request):
        """Validate min_area and max_area query string attributes
        """
        # Validate the min area:
        min_area = int(request.GET.get('min_area', 1))
        if not 1 <= min_area <= 6:
            messages.error(request, 'The first area is area 1.')
            return redirect(request.META.get('HTTP_REFERER', reverse('braindump-index')))

        # Validate the max area:
        max_area = int(request.GET.get('max_area', 6))
        if not 1 <= max_area <= 6:
            messages.error(request, 'The last area is area 6.')
            return redirect(request.META.get('HTTP_REFERER', reverse('braindump-index')))

        if max_area < min_area:
            messages.error(request, 'The max area cannot be lower than the min area.')
            return redirect(request.META.get('HTTP_REFERER', reverse('braindump-index')))

        return min_area, max_area

    def generate_min_max_area_query_string(self, min_area=1, max_area=6):
        """Build a query string containing min_area and max_area
        """
        query_string = dict()

        if min_area != 1:
            query_string['min_area'] = min_area

        if max_area != 6:
            query_string['max_area'] = max_area

        return '&'.join('{}={}'.format(key, value) for key, value in query_string.items())

    def get_probability_weighted_area(self, min_area=1, max_area=6):
        """Generate a random area by probability
        """
        # Generate range of selected areas:
        area_range = numpy.arange(min_area, max_area + 1)

        # Generate probability:
        area_probability = list()
        for _ in range(len(area_range)):
            area_probability.append(1 / 2 ** _)

        # Normalization of the probability:
        area_probability = numpy.array(area_probability)
        area_probability /= area_probability.sum()

        return numpy.random.choice(area_range, 1, p=area_probability)[0]

class BraindumpIndex(LoginRequiredMixin, TemplateView):
    """List all possible Braindump sessions
    """
    template_name = 'braindump/braindump_index.html'

    def get_context_data(self):
        share_contract_requests = ShareContract.user_objects.all(self.request.user).filter(accepted=False)

        owned_category_list = Category.owned_objects.all(self.request.user)
        shared_category_list = Category.shared_objects.all(self.request.user)
        category_list = owned_category_list | shared_category_list

        context = {
            'share_contract_requests': share_contract_requests,
            'category_list': category_list,
        }

        return context

class BraindumpSession(LoginRequiredMixin, View, BraindumpViewMixin):
    """Run a Braindump session for all cards in a certain category
    """
    http_method_names = ['get']

    def get(self, request, category_pk):
        owned_category_list = Category.owned_objects.all(self.request.user)
        shared_category_list = Category.shared_objects.all(self.request.user)
        category_list = owned_category_list | shared_category_list
        get_object_or_404(category_list, pk=category_pk)
        query_string = self.handle_query_string(request)

        # Maybe a better choice: https://eli.thegreenplace.net/2010/01/22/weighted-random-generation-in-python/

        min_area, max_area = self.validate_min_max_area(request)

        # Correct min/max area for getting only options with actually existing cards
        # (this does not exclude areas *between* min_area and max_area!):
        card_placement_filter = CardPlacement.user_objects.all(self.request.user).filter(
            card__category_id=category_pk,
            area__gte=min_area,
            area__lte=max_area,
            postpone_until__lte=timezone.now(),
        ).order_by('area')

        if card_placement_filter.exists():
            adjusted_min_area = card_placement_filter.first().area
            adjusted_max_area = card_placement_filter.last().area

            # If no card matches the selected area, retry:
            card_placement = False
            retries = 0
            while not card_placement and retries < 10:
                # Select an area:
                randomly_selected_area = self.get_probability_weighted_area(adjusted_min_area, adjusted_max_area)

                # Query a random card of the selected area (order_by('?') returns *one* random Card object):
                card_placement = CardPlacement.user_objects.all(self.request.user).filter(
                    card__category_id=category_pk,
                    area=randomly_selected_area,
                    postpone_until__lte=timezone.now(),
                ).order_by('?').first()

                retries += 1

            if card_placement:
                # Render a Braindump session containing the selected card:
                context = {
                    'card': card_placement.card,
                    'card_placement': card_placement,
                    'braindump_ok_query_string': query_string,
                    'braindump_nok_query_string': query_string,
                    'braindump_try_again_query_string': query_string,
                }

                return render(request, 'braindump/braindump_session.html', context)
            else:
                card_error = True
        else:
            card_error = True

        if card_error:
            messages.warning(
                request,
                'I cannot find any cards for this category in the desired areas. '
                'Is it possible that you postponed all possible cards? '
                'You can remove the postpone marker with the "Expedite" button.'
            )
            return redirect(reverse('category-detail', args=(category_pk,)))

class BraindumpOK(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Handle clicks on the "OK" button
    """
    permanent = False

    def get_redirect_url(self, card_pk, category_pk):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)
        card_placement.move_forward()
        card_placement.set_last_interaction()

        query_string = self.handle_query_string(self.request)

        return '{}{}'.format(reverse('braindump-session', args=(category_pk,)), query_string)

class BraindumpNOK(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Handle clicks on the "Not OK" button
    """
    permanent = False

    def get_redirect_url(self, card_pk, category_pk):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)

        if card_placement.card.category.mode == 1:
            card_placement.reset()
        elif card_placement.card.category.mode == 2:
            card_placement.move_backward()

        card_placement.set_last_interaction()

        query_string = self.handle_query_string(self.request)

        return '{}{}'.format(reverse('braindump-session', args=(category_pk,)), query_string)

class BraindumpPostpone(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Handle clicks on the "Postpone" button
    """
    permanent = False

    def get_redirect_url(self, card_pk, category_pk, seconds):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)

        if int(seconds) > settings.BRAINDUMP_MAX_POSTPONE_SECONDS:
            messages.error(
                self.request,
                'Cannot postpone a card for more than {} seconds.'.format(settings.BRAINDUMP_MAX_POSTPONE_SECONDS)
            )
        else:
            card_placement.postpone_until = timezone.now() + timedelta(seconds=int(seconds))
            messages.info(self.request, 'Ok, I will not show the card for 15 minutes from now.')

        card_placement.set_last_interaction()

        query_string = self.handle_query_string(self.request)

        return '{}{}'.format(reverse('braindump-session', args=(category_pk,)), query_string)

class CardExpedite(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Expedite a card (undo postpone)
    """
    permanent = False

    def get_redirect_url(self, card_pk):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)
        card_placement.expedite()
        messages.success(
            self.request,
            mark_safe('The postpone marker of {} has been removed.'.format(
                reverse('card-detail', args=(card_placement.card.pk,)),
                card_placement.card,
            ))
        )
        return self.request.META.get('HTTP_REFERER', reverse('card-detail', args=(card_placement.card.pk,)))

class CardReset(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Handle clicks on the "Reset" button of a card
    """
    permanent = False

    def get_redirect_url(self, card_pk):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)
        prev_area = card_placement.area

        if prev_area != 1:
            card_placement.reset()
            undo_url = reverse('card-set-area', args=(card_placement.card.pk, prev_area))
            messages.success(
                self.request,
                mark_safe('{} moved from area {} to area 1. Undo'.format(card_placement.card,
                                                                                prev_area, undo_url))
            )
        else:
            messages.success(self.request, 'Card is already in area 1.')

        return self.request.META.get('HTTP_REFERER', reverse('card-detail', args=(card_placement.pk,)))

class CardSetArea(LoginRequiredMixin, RedirectView, BraindumpViewMixin):
    """Handle manual movements of a card
    """
    permanent = False

    def get_redirect_url(self, card_pk, area):
        card_placement = get_object_or_404(CardPlacement.user_objects.all(self.request.user), card_id=card_pk)
        prev_area = card_placement.area
        card_placement.area = area
        card_placement.save()
        messages.success(self.request, '{} moved from area {} to area {}.'.format(card_placement.card, prev_area,
                                                                                card_placement.area))
        return self.request.META.get('HTTP_REFERER', reverse('card-detail', args=(card_placement.card.pk,)))

max_stars_repo_name: mhndlsz/memodrop

max_issues_repo_name: mhndlsz/memodrop

max_forks_repo_name: mhndlsz/memodrop


text:
#!/usr/bin/env python3
"""Contains code to slice up an image into smaller tiles."""

__author__ = "Kadhir Umasankar"

import argparse
import pathlib
from typing import List, Tuple

import numpy as np
from PIL import Image
import tqdm

from hawk_eye.data_generation import generate_config as config

def slice_image(
    images: List, tile_size: Tuple[int, int], overlap: int, save_dir: pathlib.Path
) -> None:
    """ Take in an image and slice it into smaller images.
    Args:
        images: A list of paths to images to be sliced.
        tile_size: The (width, height) of the tiles.
        overlap: The overlap between adjacent tiles.
        save_dir: The path to the directory within which to save slices.
    Returns:
        None.
    """
    for filename in tqdm.tqdm(images, desc="Slicing images", total=len(images)):

        image = Image.open(filename)
        width, height = image.size

        # Cropping logic repurposed from hawk_eye.inference.find_targets.tile_image()
        for x in range(0, width - overlap, tile_size[0] - overlap):

            # Shift back to extract tiles on the image
            if x + tile_size[0] >= width and x != 0:
                x = width - tile_size[0]

            for y in range(0, height - overlap, tile_size[1] - overlap):

                if y + tile_size[1] >= height and y != 0:
                    y = height - tile_size[1]

                tile = image.crop((x, y, x + tile_size[0], y + tile_size[1]))

                tile.save(save_dir / f"{filename.stem}-{x}-{y}{filename.suffix}")

    (save_dir / "labels.txt").write_text("\n".join(config.SHAPE_TYPES))

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--image_dir",
        type=pathlib.Path,
        required=False,
        help="Path to directory of images to be sliced.",
    )
    parser.add_argument(
        "--image_path",
        type=pathlib.Path,
        required=False,
        help="Path to an image to be sliced.",
    )
    parser.add_argument(
        "--image_extensions",
        type=str,
        required=False,
        default=".JPG,.jpg",
        help="A comma-separated list of extensions of images to slice.",
    )
    parser.add_argument(
        "--save_dir",
        type=pathlib.Path,
        required=True,
        help="Path to directory in which to store sliced images.",
    )
    parser.add_argument(
        "--overlap",
        type=int,
        default=50,
        help="Number of pixels of overlap between adjacent slices.",
    )
    args = parser.parse_args()

    # Raises an error if none of image arguments are provided
    if args.image_path is None and args.image_dir is None:
        raise ValueError("Please supply either an image or directory of images.")

    # Extracts extensions from the input argument
    exts = [
        f".{ext}" if "." not in ext else ext for ext in args.image_extensions.split(",")
    ]

    if args.image_path is not None:
        if args.image_path.suffix in exts:
            images = [args.image_path.expanduser()]

    elif args.image_dir is not None:

        # If image_dir points to a directory, paths to all the images with the correct
        # extension are put in a list
        image_dir = args.image_dir.expanduser()

        if image_dir.is_dir():
            images = []

            for ext in exts:
                images.extend(list(image_dir.glob(f"*{ext}")))
        else:
            raise ValueError("Please supply a valid path to a directory.")

    # Makes a directory to save slices in
    save_dir = args.save_dir.expanduser()
    save_dir.mkdir(exist_ok=True, parents=True)

    slice_image(
        images=images,
        tile_size=config.CROP_SIZE,
        overlap=args.overlap,
        save_dir=save_dir,
    )

max_stars_repo_name: uavaustin/image-rec

max_issues_repo_name: uavaustin/image-rec

max_forks_repo_name: uavaustin/hawk-eye


text:
"""Module for batch experiments simulations (O-Dimensinal): reactions and
pH equlibrium simulations
"""
import numpy as np
import porousmedialab.phcalc as phcalc
import porousmedialab.plotter as plotter
from porousmedialab.dotdict import DotDict
from porousmedialab.lab import Lab

class Batch(Lab):
    """The batch experiments simulations

    Attributes:
        acid_base_system (obj): object of acid-base system
        N (int): spatial resolution (=1)
        plot (TYPE):
        plot_delta (TYPE): Description
        plot_deltas (TYPE): Description
        plot_fractions (TYPE): Description
        plot_profiles (TYPE): Description
        plot_rate (TYPE): Description
        plot_rates (TYPE): Description
    """

    def __init__(self, tend, dt):
        """Summary

        Args:
            tend (TYPE): Description
            dt (TYPE): Description
        """
        super().__init__(tend, dt)
        self.N = 1

    def add_species(self, name, init_conc):
        """Summary

        Args:
            name (string): name of the element
            init_conc (float): initial concentration
        """
        self.species[name] = DotDict({})
        self.species[name]['init_conc'] = init_conc
        self.species[name]['concentration'] = np.zeros((self.N,
                                                           self.time.size))
        self.species[name]['alpha'] = np.zeros((self.N, self.time.size))
        self.species[name]['rates'] = np.zeros((self.N, self.time.size))
        self.species[name]['concentration'][:, 0] = self.species[name][
            'init_conc']
        self.profiles[name] = self.species[name]['concentration'][:, 0]
        self.species[name]['int_transport'] = False
        self.dcdt[name] = '0'

    def integrate_one_timestep(self, i):
        """Summary

        Args:
            i (TYPE): Description
        """
        if i == 1:
            self.pre_run_methods()
        self.reactions_integrate_scipy(i)
        if self.henry_law_equations:
            self.henry_equilibrium_integrate(i)
        if self.acid_base_components:
            self.acid_base_equilibrium_solve(i)

    def add_time_variable(self):
        """Temporal hack of adding time variable with dctdt=1
        Needed for simulation of time dependent variables in
        """
        self.add_species(name='TIME', init_conc=0)
        self.dcdt['TIME'] = '1'

    def create_acid_base_system(self):
        """creates an object of acid-base system stores it as instance variable
        creates variable 'pH' with initial pH=7.
        """
        self.add_species(name='pH', init_conc=7)
        self.acid_base_system = phcalc.System(
            *[c['pH_object'] for c in self.acid_base_components])

    def acid_base_update_concentrations(self, i):
        """Summary

        Args:
            i (TYPE): Description
        """
        for component in self.acid_base_components:
            init_conc = 0
            alphas = component['pH_object'].alpha(
                self.species['pH']['concentration'][:, i])
            for idx in range(len(component['species'])):
                init_conc += self.species[component['species'][idx]][
                    'concentration'][:, i]
            for idx in range(len(component['species'])):
                self.species[component['species'][idx]][
                    'concentration'][:, i] = init_conc * alphas[idx]
                self.profiles[component['species'][idx]] = self.species[
                    component['species'][idx]]['concentration'][:, i]
                self.species[component['species'][idx]]['alpha'][:, i] = alphas[
                    idx]

    plot = plotter.plot_depth_index
    plot_profiles = plotter.all_plot_depth_index
    plot_fractions = plotter.plot_fractions
    plot_rates = plotter.plot_batch_rates
    plot_rate = plotter.plot_batch_rate
    plot_deltas = plotter.plot_batch_deltas
    plot_delta = plotter.plot_batch_delta

max_issues_repo_name: biogeochemistry/PorousMediaLab

max_forks_repo_name: biogeochemistry/PorousMediaLab


text:
'''Example code for producing ROC plots for evaluating model performance.
Also includes reshape function as before. Takes in a pre-made model and compares
model output to target output to find false positive rate and true positive rates
for plotting.'''

import tensorflow as tf
import pickle
import numpy as np
import matplotlib.pyplot as plt
import cv2
import scipy
from scipy import mean
import statistics as s
from sklearn.metrics import roc_curve, auc

length = 25 #dimensions of images model was trained with
width = 25

def reshape_function_validate(length, width):
    pickle_in = open("Your file path here","rb")
    X_test = pickle.load(pickle_in)
    X_test = np.array(X_test)

    A = []
    for i in range(0, (len(X_test))):
        A.append(cv2.resize(X_test[i][:,:,0], (length, width)))  #resize images as necessary
    A = np.array(A).reshape(-1, length, width, 1)
    A = np.array(A)
    A = np.array(A/np.amax(A))
    return A

pickle_in = open("Your file path here","rb") #load in target values
y_test = pickle.load(pickle_in)

def model_predictionTTB(path_to_model, length, width):
    model = tf.keras.models.load_model(path_to_model)
    predictions = model.predict(reshape_function_validate(length, width))
    return predictions #make model predictions using previously saved model

fpr, tpr, _ =  roc_curve(y_test, model_predictionTTB("Your file path to model here", length, width)) #calculates false positive rate and true positive rate
plt.plot(fpr, tpr)
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.show()

print('The AUC value is: ', auc(fpr,tpr)) #finds area under curve values for fpr and tpr

max_stars_repo_name: ATGSilva/ML-TTB-VBF-Detection

max_issues_repo_name: ATGSilva/ML-TTB-VBF-Detection

max_forks_repo_name: ATGSilva/ML-TTB-VBF-Detection

reason: import numpy,import scipy,from scipy


text:
# -*- coding: utf-8 -*-
"""
Created on Fri Jan 27 17:58:35 2017

@author: Suhas Somnath
"""

from __future__ import division, print_function, absolute_import, unicode_literals

from os import path, remove
import sys
import h5py
import numpy as np  # For array operations

from .write_utils import Dimension
from .translator import Translator, generate_dummy_main_parms
from .hdf_utils import write_main_dataset, write_simple_attrs

if sys.version_info.major == 3:
    unicode = str

class NumpyTranslator(Translator):
    """
    Writes a numpy array to .h5
    """

    def translate(self, h5_path, data_name, raw_data, quantity, units, pos_dims, spec_dims,
                  translator_name='NumpyTranslator', parm_dict=None, extra_dsets=None, **kwargs):
        """
        Writes the provided datasets and parameters to an h5 file

        Parameters
        ----------
        h5_path : String / Unicode
            Absolute path of the h5 file to be written
        data_name : String / Unicode
            Name of the scientific data type. Example - 'SEM'
        raw_data : np.ndarray
            2D matrix formatted as [position, spectral]
        quantity : String / Unicode
            Name of the physical quantity stored in the dataset. Example - 'Current'
        units : String / Unicode
            Name of units for the quantity stored in the dataset. Example - 'A' for amperes
        pos_dims : Dimension or array-like of Dimension objects
            Sequence of Dimension objects that provides all necessary instructions for constructing the
            indices and values datasets
            Object specifying the instructions necessary for building the Position indices and values datasets
        spec_dims : Dimension or array-like of Dimension objects
            Sequence of Dimension objects that provides all necessary instructions for constructing the
            indices and values datasets
            Object specifying the instructions necessary for building the Spectroscopic indices and values datasets
        translator_name : String / unicode, Optional
            Name of the translator. Example - 'HitachiSEMTranslator'
        parm_dict : dictionary (Optional)
            Dictionary of parameters that will be written under the group 'Measurement_000'
        extra_dsets : dictionary (Optional)
            Dictionary whose values will be written into individual HDF5 datasets and whose corresponding keys provide
            the names of the datasets. You are recommended to limit these to simple and small datasets.
        kwargs: will be passed onto hdf_utils.write_main_dset() which will in turn will be passed onto the creation of
            the dataset. Please pass chunking, compression, dtype, and other arguments this way

        Returns
        -------
        h5_path : String / unicode
            Absolute path of the written h5 file

        """
        for arg in [h5_path, data_name, translator_name, quantity, units]:
            assert isinstance(arg, (str, unicode))
            assert len(arg) > 0
        assert isinstance(raw_data, np.ndarray)
        assert raw_data.ndim == 2

        for ind, dimensions in enumerate([pos_dims, spec_dims]):
            if isinstance(dimensions, Dimension):
                dimensions = [dimensions]
            if not isinstance(dimensions, (list, np.ndarray, tuple)):
                raise TypeError('dimensions should be array-like ')
            if not np.all([isinstance(x, Dimension) for x in dimensions]):
                raise TypeError('dimensions should be a sequence of Dimension objects')
            # Check to make sure that the product of the position and spectroscopic dimension sizes match with
            # that of raw_data
            assert raw_data.shape[ind] == np.product([len(x.values) for x in dimensions])

        if extra_dsets is not None:
            if not isinstance(extra_dsets, dict):
                raise TypeError('extra_dsets should be specified as dictionaries')
            for key, val in extra_dsets.items():
                if not isinstance(key, (str, unicode)):
                    raise TypeError('keys for extra_dsets should be strings')
                if len(key.strip()) == 0:
                    raise ValueError('keys for extra_dsets should not be empty')
                if np.any([key in x for x in ['Spectroscopic_Indices', 'Spectroscopic_Values', 'Position_Indices',
                                              'Position_Values', 'Raw_Data']]):
                    raise ValueError('keys for extra_dsets cannot match reserved names for existing datasets')
                # Now check for data:
                if not isinstance(val, (list, tuple, np.ndarray)):
                    raise TypeError('values for extra_dsets should be a tuple, list, or numpy array')
        else:
            extra_dsets = dict()

        if path.exists(h5_path):
            remove(h5_path)

        if parm_dict is None:
            parm_dict = {}

        global_parms = generate_dummy_main_parms()
        global_parms['data_type'] = data_name
        global_parms['translator'] = translator_name

        # Begin writing to file:
        with h5py.File(h5_path) as h5_f:

            # Root attributes first:
            write_simple_attrs(h5_f, global_parms)

            # measurement group next
            meas_grp = h5_f.create_group('Measurement_000')
            write_simple_attrs(meas_grp, parm_dict)

            # channel group next
            chan_grp = meas_grp.create_group('Channel_000')

            _ = write_main_dataset(chan_grp, raw_data, 'Raw_Data', quantity, units, pos_dims, spec_dims, **kwargs)

            for key, val in extra_dsets.items():
                chan_grp.create_dataset(key.strip(), data=val)

        return h5_path

max_stars_repo_name: ealopez/pycroscopy

max_issues_repo_name: ealopez/pycroscopy

max_forks_repo_name: ealopez/pycroscopy


text:
import numpy as np
import os
import torch
import pickle
import random
from math import log2
from torch.utils.data import Dataset
from copy import deepcopy

config = {
    # movielens
    'n_rate': 6,
    'n_year': 81,
    'n_genre': 25,
    'n_director': 2186,
    'n_gender': 2,
    'n_age': 7,
    'n_occupation': 21,
    # bookcrossing
    'n_year_bk': 80,
    'n_author': 25593,
    'n_publisher': 5254,
    'n_age_bk': 106,
    'n_location': 65,
    # TV shows
    'n_user': 70,
    'n_service': 7,
    'n_engagement': 3,
    'n_tv_genre': 11,
    # sample_size
    'sample_size': 20
}

default_info = {
    'movielens': {'n_y': 5, 'u_in_dim': 3, 'i_in_dim': 4},
    'bookcrossing': {'n_y': 10, 'u_in_dim': 2, 'i_in_dim': 3},
    'tv_shows': {'n_y': 3, 'u_in_dim': 1, 'i_in_dim': 2}
}

def to_torch(in_list):
    return torch.from_numpy(np.array(in_list))

# ===================== Load data ==========================
def train_test_user_list(dataset='movielens', rand=True, random_state=32, train_test_split_ratio=0.8, store=False):
    path = 'data_processed/' + dataset + '/raw/'
    path_store = 'data_processed/' + dataset + '/'
    len_user = int(len(os.listdir(path)) / 4)  # movielens 6040, bookcrossing 4323
    training_size = int(len_user * train_test_split_ratio)  # movielens: , bookcrossing:
    user_id_list = list(range(1, len_user))
    if rand:
        random.shuffle(user_id_list)
    else:
        random.seed(random_state)
        random.shuffle(user_id_list)

    train_user_set, test_user_set = user_id_list[:training_size], user_id_list[training_size:]
    if store:
        pickle.dump(train_user_set, open(path_store + 'train_user_set.p', 'wb'))
        pickle.dump(test_user_set, open(path_store + 'test_user_set.p', 'wb'))
    return train_user_set, test_user_set

def load_user_info(user_id, dataset='movielens', support_size=16, query_size=4, device=torch.device('cpu')):
    path = 'data_processed/' + dataset + '/raw/'
    u_x1 = pickle.load(open('{}sample_{}_x1.p'.format(path, str(user_id)), 'rb'))
    u_x2 = pickle.load(open('{}sample_{}_x2.p'.format(path, str(user_id)), 'rb'))
    u_y = pickle.load(open('{}sample_{}_y.p'.format(path, str(user_id)), 'rb'))
    u_y0 = pickle.load(open('{}sample_{}_y0.p'.format(path, str(user_id)), 'rb'))

    u_x1 = np.tile(u_x1, (config['sample_size'], 1))
    u_y = u_y-1

    sup_x1, que_x1 = to_torch(u_x1[:support_size]).to(device), \
                     to_torch(u_x1[support_size:support_size+query_size]).to(device)
    sup_x2, que_x2 = to_torch(u_x2[:support_size]).to(device), \
                     to_torch(u_x2[support_size:support_size+query_size]).to(device)
    sup_y, que_y = to_torch(u_y[:support_size]).to(device), \
                   to_torch(u_y[support_size:support_size+query_size]).to(device)
    sup_y0, que_y0 = to_torch(u_y0[:support_size]).to(device), \
                     to_torch(u_y0[support_size:support_size+query_size]).to(device)

    return sup_x1, sup_x2, sup_y, sup_y0, que_x1, que_x2, que_y, que_y0

# ==========================================
class UserDataLoader(Dataset):
    def __init__(self, x1, x2, y, y0, transform=None):
        self.x1 = x1
        self.x2 = x2
        self.y = y
        self.y0 = y0
        self.transform = transform

    def __len__(self):
        return len(self.y)

    def __getitem__(self, idx):
        if torch.is_tensor(idx):
            idx = idx.tolist()

        user_info = self.x1[idx]
        item_info = self.x2[idx]
        ratings = self.y[idx]
        cold_labels = self.y0[idx]

        return user_info, item_info, ratings, cold_labels

# =============================================
def get_params(param_list):
    params = []
    count = 0
    for param in param_list:
        if count % 2 == 0:
            value = deepcopy(param.data)
            params.append(value)
            del value
        count += 1
    return params

def get_zeros_like_params(param_list):
    zeros_like_params = []
    count = 0
    for param in param_list:
        if count % 2 == 0:
            value = deepcopy(torch.zeros_like(param.data))
            zeros_like_params.append(value)
        count += 1
    return zeros_like_params

def init_params(param_list, init_values):
    count = 0
    init_count = 0
    for param in param_list:
        if count % 2 == 0:
            param.data.copy_(init_values[init_count])
            init_count += 1
        count += 1

def init_u_mem_params(param_list, init_values, bias_term, tao):
    count = 0
    init_count = 0
    for param in param_list:
        if count % 2 == 0:
            param.data.copy_(init_values[init_count]-tao*bias_term[init_count])
            init_count += 1
        count += 1

def init_ui_mem_params(param_list, init_values):
    count = 0
    for param in param_list:
        if count % 2 == 0:
            param.data.copy_(init_values)
        count += 1

def get_grad(param_list):
    count = 0
    param_grads = []
    for param in param_list:
        if count % 2 == 0:
            value = deepcopy(param.grad)
            param_grads.append(value)
            del value
        count += 1
    return param_grads

def grads_sum(raw_grads_list, new_grads_list):
    return [raw_grads_list[i]+new_grads_list[i] for i in range(len(raw_grads_list))]

def update_parameters(params, grads, lr):
    return [params[i] - lr*grads[i] for i in range(len(params))]

# ===============================================
def activation_func(name):
    name = name.lower()
    if name == "sigmoid":
        return torch.nn.Sigmoid()
    elif name == "tanh":
        return torch.nn.Tanh()
    elif name == "relu":
        return torch.nn.ReLU()
    elif name == 'softmax':
        return torch.nn.Softmax()
    elif name == 'leaky_relu':
        return torch.nn.LeakyReLU(0.1)
    else:
        return torch.nn.Sequential()

# ===============================================
def mae(ground_truth, test_result):
    if len(ground_truth) > 0:
        pred_y = torch.argmax(test_result, dim=1)
        sub = ground_truth-pred_y
        abs_sub = torch.abs(sub)
        out = torch.mean(abs_sub.float(), dim=0)
    else:
        out = 1
    return out

def ndcg(ground_truth, test_result, top_k=3):
    pred_y = torch.argmax(test_result, dim=1)
    sort_real_y, sort_real_y_index = ground_truth.clone().detach().sort(descending=True)
    sort_pred_y, sort_pred_y_index = pred_y.clone().detach().sort(descending=True)
    pred_sort_y = ground_truth[sort_pred_y_index][:top_k]
    top_pred_y, _ = pred_sort_y.sort(descending=True)

    ideal_dcg = 0
    n = 1
    for value in sort_real_y[:top_k]:
        i_dcg = (2**float(value+1) - 1)/log2(n+1)
        ideal_dcg += i_dcg
        n += 1

    pred_dcg = 0
    n = 1
    for value in top_pred_y:
        p_dcg = (2**float(value+1) - 1)/log2(n+1)
        pred_dcg += p_dcg
        n += 1

    n_dcg = pred_dcg/ideal_dcg
    return n_dcg


text:
"""A mini synthetic dataset for graph classification benchmark."""
import math
import networkx as nx
import numpy as np

from ..graph import DGLGraph

__all__ = ['MiniGCDataset']

class MiniGCDataset(object):
    """The dataset class.

    The datset contains 8 different types of graphs.

    * class 0 : cycle graph
    * class 1 : star graph
    * class 2 : wheel graph
    * class 3 : lollipop graph
    * class 4 : hypercube graph
    * class 5 : grid graph
    * class 6 : clique graph
    * class 7 : circular ladder graph

    .. note::
        This dataset class is compatible with pytorch's :class:`Dataset` class.

    Parameters
    ----------
    num_graphs: int
        Number of graphs in this dataset.
    min_num_v: int
        Minimum number of nodes for graphs
    max_num_v: int
        Maximum number of nodes for graphs
    """
    def __init__(self, num_graphs, min_num_v, max_num_v):
        super(MiniGCDataset, self).__init__()
        self.num_graphs = num_graphs
        self.min_num_v = min_num_v
        self.max_num_v = max_num_v
        self.graphs = []
        self.labels = []
        self._generate()

    def __len__(self):
        """Return the number of graphs in the dataset."""
        return len(self.graphs)

    def __getitem__(self, idx):
        """Get the i^th sample.

        Paramters
        ---------
        idx : int
            The sample index.

        Returns
        -------
        (dgl.DGLGraph, int)
            The graph and its label.
        """
        return self.graphs[idx], self.labels[idx]

    @property
    def num_classes(self):
        """Number of classes."""
        return 8

    def _generate(self):
        self._gen_cycle(self.num_graphs // 8)
        self._gen_star(self.num_graphs // 8)
        self._gen_wheel(self.num_graphs // 8)
        self._gen_lollipop(self.num_graphs // 8)
        self._gen_hypercube(self.num_graphs // 8)
        self._gen_grid(self.num_graphs // 8)
        self._gen_clique(self.num_graphs // 8)
        self._gen_circular_ladder(self.num_graphs - len(self.graphs))
        # preprocess
        for i in range(self.num_graphs):
            self.graphs[i] = DGLGraph(self.graphs[i])
            # add self edges
            nodes = self.graphs[i].nodes()
            self.graphs[i].add_edges(nodes, nodes)

    def _gen_cycle(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            g = nx.cycle_graph(num_v)
            self.graphs.append(g)
            self.labels.append(0)

    def _gen_star(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            # nx.star_graph(N) gives a star graph with N+1 nodes
            g = nx.star_graph(num_v - 1)
            self.graphs.append(g)
            self.labels.append(1)

    def _gen_wheel(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            g = nx.wheel_graph(num_v)
            self.graphs.append(g)
            self.labels.append(2)

    def _gen_lollipop(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            path_len = np.random.randint(2, num_v // 2)
            g = nx.lollipop_graph(m=num_v - path_len, n=path_len)
            self.graphs.append(g)
            self.labels.append(3)

    def _gen_hypercube(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            g = nx.hypercube_graph(int(math.log(num_v, 2)))
            g = nx.convert_node_labels_to_integers(g)
            self.graphs.append(g)
            self.labels.append(4)

    def _gen_grid(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            assert num_v >= 4, 'We require a grid graph to contain at least two ' \
                                   'rows and two columns, thus 4 nodes, got {:d} ' \
                                   'nodes'.format(num_v)
            n_rows = np.random.randint(2, num_v // 2)
            n_cols = num_v // n_rows
            g = nx.grid_graph([n_rows, n_cols])
            g = nx.convert_node_labels_to_integers(g)
            self.graphs.append(g)
            self.labels.append(5)

    def _gen_clique(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            g = nx.complete_graph(num_v)
            self.graphs.append(g)
            self.labels.append(6)

    def _gen_circular_ladder(self, n):
        for _ in range(n):
            num_v = np.random.randint(self.min_num_v, self.max_num_v)
            g = nx.circular_ladder_graph(num_v)
            self.graphs.append(g)
            self.labels.append(7)

max_stars_repo_name: jackroos/dgl

max_issues_repo_name: jackroos/dgl

max_forks_repo_name: jackroos/dgl

reason: import numpy,import networkx


text:
"""
File: examples/basis/training_set_plot.py
Author: Keith Tauscher
Date: 22 Feb 2019

Description: Example script showing how to use the plot_training_set_with_modes
             function to plot a three-panel figure summarizing a training set.
"""
from __future__ import division
import numpy as np
from distpy import DistributionSet, GaussianDistribution
from pylinex import plot_training_set_with_modes, GaussianModel

seed = 0
np.random.seed(seed)

num_modes = 5
x_values = np.linspace(40, 120, 81)
integration_time = 100
error = 83 * np.power(x_values / 50, -2.5) / np.sqrt(integration_time)
xlabel = '$\\nu$ [MHz]'
extra_ylabel_string = ' [mK]'
title = 'Gaussian training set'
fontsize = 24
curve_slice = slice(0, None, 2)
alpha = 1

model = GaussianModel(x_values)
distribution_set = DistributionSet()
distribution_set.add_distribution(GaussianDistribution(-100, 50 ** 2),\
    'amplitude')
distribution_set.add_distribution(GaussianDistribution(80, 20 ** 2), 'center')
distribution_set.add_distribution(GaussianDistribution(10, 2 ** 2), 'scale')
ndraw = 10
return_parameters = False
training_set = model.curve_sample(distribution_set, ndraw,\
    return_parameters=return_parameters)

plot_training_set_with_modes(training_set, num_modes, x_values=x_values,\
    curve_slice=curve_slice, alpha=alpha, fontsize=fontsize, xlabel=xlabel,\
    extra_ylabel_string=extra_ylabel_string, title=title, show=True)

max_stars_repo_name: CU-NESS/pylinex

max_issues_repo_name: CU-NESS/pylinex

max_forks_repo_name: CU-NESS/pylinex


text:
'''
Objective: This is the script for generating test job for cpu usage tracking

'''

import numpy as np
from sklearn.ensemble import GradientBoostingRegressor

X = np.random.random(size=1000)
X = 100 * X * np.sin(X)
X = X.reshape(1000,1)

y = np.random.random(size=1000)
y = 1000 * y
y = y.reshape(1000,)

for i in range(100):
    print('='*30)
    print('This is fitting iteration {}/100'.format(i+1))
    model = GradientBoostingRegressor(n_estimators=10000, tol=1e-14, n_iter_no_change=10000)
    model.fit(X,y)
    print(model.score(X,y))
    print('='*30)


text:
import pathlib
import logging
from collections import OrderedDict

import numpy as np
import h5py

from .. import tools

logger = logging.getLogger(__name__)

BACKENDS = OrderedDict()

def check_if_raw_data(path):
    '''Checks if the given path is a supported raw data file and returns the backend if it exists, else returns None.
    '''
    for backend, (_, validate) in BACKENDS.items():
        if validate(path):
            return backend
    return None

def backend_loader(name):
    '''Decorator to register function as a raw-data backend loader

    #TODO: add descripiton of backend load function here (what it should return ect)
    '''

    def backend_wrapper(func):
        logger.debug(f'Registering loader for {name} backend')
        if name in BACKENDS:
            BACKENDS[name][0] = func
        else:
            BACKENDS[name] = [func, None]
        return func

    return backend_wrapper

def backend_validator(name):
    '''Decorator to register function as a raw-data backend validator

    #TODO: add descripiton of backend load function here (what it should return ect)
    '''

    def backend_wrapper(func):
        logger.debug(f'Registering validator for {name} backend')
        if name in BACKENDS:
            BACKENDS[name][1] = func
        else:
            BACKENDS[name] = [None, func]
        return func

    return backend_wrapper

class RawDataInterface:
    '''Common interface between metecho functionality and different radar-data backends.
    '''

    DATA_AXIS = [
        'channel',
        'sample',
        'pulse',
        'polarization',
    ]

    META_KEYS = [
        'start_time',
    ]

    @tools.profiling.timeing(f'{__name__}.RawDataInterface')
    def __init__(self, path, backend=None, load_on_init=True, **kwargs):
        self._clear()
        if path is not None:
            self.path = pathlib.Path(path)
        self.backend = backend
        if load_on_init:
            self.load(**kwargs)

    def _clear(self):
        self.data = None
        self.axis = {key: None for key in self.DATA_AXIS}
        self.meta = {key: None for key in self.META_KEYS}

    @tools.profiling.timeing(f'{__name__}.RawDataInterface')
    def load(self, **kwargs):
        self._clear()

        if self.backend is None:
            backend_found = False

            for backend, (load_func, validate) in BACKENDS.items():
                if validate(self.path):
                    data, axis, meta = load_func(self.path, **kwargs)
                    backend_found = True
                    self.backend = backend
                else:
                    continue

                break

            if not backend_found:
                raise ValueError('No backend found for given input path')
        else:
            if self.backend not in BACKENDS:
                raise ValueError(f'Given backend {self.backend} does not exist')

            load_func, validate = BACKENDS[self.backend]
            if validate(self.path):
                data, axis, meta = load_func(self.path, **kwargs)
            else:
                raise ValueError(f'Given backend {self.backend} does not validate input path')

        if not isinstance(data, np.ndarray):
            raise ValueError(f'Backend must return data as a numpy ndarray not "{type(data)}"')

        self.data = data
        self.meta.update(meta)
        self.axis.update(axis)


text:
import datetime
from collections import OrderedDict

import mock
import numpy
import pytest
from pytz import timezone, utc

from aa import data

@pytest.mark.parametrize(
    "tz", (utc, timezone("Europe/Amsterdam"), timezone("US/Eastern"))
)
def test_ArchiveEvent_datetime_returns_correct_datetime(event_1d, tz):
    dt = datetime.datetime(1970, 1, 1, 0, 1, 40, 100000)
    utc_dt = utc.localize(dt)
    local_dt = utc_dt.astimezone(tz)
    assert event_1d.datetime(tz) == local_dt

def test_ArchiveEvent_utc_datetime_returns_correct_datetime(event_1d):
    dt = datetime.datetime(1970, 1, 1, 0, 1, 40, 100000)
    utc_dt = utc.localize(dt)
    assert event_1d.utc_datetime == utc_dt

def test_ArchiveEvent_datetime_returns_numpy_array_of_datetimes(data_1d):
    assert isinstance(data_1d.datetimes(utc), numpy.ndarray)
    dt = datetime.datetime(1970, 1, 1, 0, 1, 40, 100000)
    utc_dt = utc.localize(dt)
    assert data_1d.datetimes(utc)[0] == utc_dt

def test_ArchiveEvent_utc_datetimes_returns_numpy_array_of_datetimes(data_1d):
    assert isinstance(data_1d.utc_datetimes, numpy.ndarray)
    dt = datetime.datetime(1970, 1, 1, 0, 1, 40, 100000)
    utc_dt = utc.localize(dt)
    assert data_1d.utc_datetimes[0] == utc_dt

def test_ArchiveEvent_str(dummy_pv, event_1d):
    assert dummy_pv in str(event_1d)

def test_ArchiveData_has_enum_strings(data_1d):
    assert not data_1d.has_enum_options
    data_1d._enum_options = OrderedDict([(1, "One")])
    assert data_1d.has_enum_options

def test_ArchiveData_concatenate_with_different_pv_names_raises_AssertionError():
    array = numpy.zeros((1,))
    data1 = data.ArchiveData("dummy1", array, array, array)
    data2 = data.ArchiveData("dummy2", array, array, array)
    with pytest.raises(AssertionError):
        data1.concatenate(data2)

def test_ArchiveData_concatenate_raises_ValueError_for_different_sized_array(
    data_1d, data_2d
):
    with pytest.raises(ValueError):
        # default is zero_pad=False and the arrays are different dimensions
        data_2d.concatenate(data_1d)

def test_ArchiveData_concatenate_correctly_zero_pads(data_1d, data_2d):
    result = data_2d.concatenate(data_1d, zero_pad=True)
    expected = numpy.array(((1.1, 2, 3), (1, 0, 0)))
    numpy.testing.assert_equal(result.values, expected)

@mock.patch("logging.warning")
def test_ArchiveData_concatenate_works_for_two_items(mock_warning):
    zeros = numpy.zeros((1,))
    ones = numpy.ones((1,))
    enum_options1 = {0: "dummy", 1: "fake"}
    enum_options2 = {0: "rubbish", 2: "nonsense"}
    data1 = data.ArchiveData("dummy", zeros, zeros, zeros, enum_options1)
    data2 = data.ArchiveData("dummy", ones, ones, ones, enum_options1)
    expected = numpy.array((0, 1))

    # Concat with enum options the same
    data3 = data1.concatenate(data2)
    # Note that values is always a 2d array.
    numpy.testing.assert_equal(data3.values, expected.reshape(2, 1))
    numpy.testing.assert_equal(data3.timestamps, expected)
    numpy.testing.assert_equal(data3.severities, expected)

    assert data3.enum_options == enum_options1
    mock_warning.assert_not_called()

    # Concat with enum options different
    # Should use enum options from first and issue warning.
    data2._enum_options = enum_options2
    data3 = data1.concatenate(data2)
    assert data3.enum_options == enum_options1
    mock_warning.assert_called_once()

def test_ArchiveData_concatenate_works_for_2d_arrays():
    zeros = numpy.zeros((1,))
    zeros_2d = numpy.zeros((1, 2))
    ones = numpy.ones((1,))
    ones_2d = numpy.ones((1, 2))
    data1 = data.ArchiveData("dummy", zeros_2d, zeros, zeros)
    data2 = data.ArchiveData("dummy", ones_2d, ones, ones)
    expected = numpy.array((0, 1))
    expected_2d = numpy.array(((0, 0), (1, 1)))
    data3 = data1.concatenate(data2)
    numpy.testing.assert_equal(data3.values, expected_2d)
    numpy.testing.assert_equal(data3.timestamps, expected)
    numpy.testing.assert_equal(data3.severities, expected)

def test_ArchiveData_constructor_raises_AssertionError_if_array_lengths_different(
    dummy_pv,
):
    empty_10 = numpy.zeros((10,))
    empty_11 = numpy.zeros((11,))
    with pytest.raises(AssertionError):
        data.ArchiveData(dummy_pv, empty_10, empty_10, empty_11)

@mock.patch("logging.warning")
def test_warning_logged_by_ArchiveData_constructor_if_timestamps_descending(
    patched_warning, dummy_pv
):
    empty_array = numpy.zeros((10,))
    desc = numpy.arange(2, 1, -0.1)
    data.ArchiveData(dummy_pv, empty_array, desc, empty_array)
    assert patched_warning.called

# Test both ascending array and constant array.
@pytest.mark.parametrize("timestamps", (numpy.arange(1, 2, 0.1), numpy.ones(10,)))
def test_ArchiveData_constructor_raises_no_exception_if_timestamps_valid(
    dummy_pv, timestamps
):
    empty_array = numpy.zeros((10,))
    data.ArchiveData(dummy_pv, empty_array, timestamps, empty_array)

def test_empty_ArchiveData_iterates_zero_times(empty_data):
    for _ in empty_data:
        assert False  # we shouldn't get here

def test_empty_ArchiveData_raises_IndexError_if_indexed(empty_data):
    with pytest.raises(IndexError):
        empty_data[0]

def test_empty_ArchiveData_str(empty_data):
    assert "Empty" in str(empty_data)

def test_ArchiveData_str(dummy_pv, data_2d):
    assert dummy_pv in str(data_2d)

def test_indexing_ArchiveData(data_1d, event_1d):
    assert data_1d[0] == event_1d

def test_empty_ArchiveData_evaluates_as_False(empty_data):
    assert not empty_data

def test_non_empty_ArchiveData_evaluates_as_True(data_1d):
    assert data_1d

def test_ArchiveData_equality():
    zeros = numpy.zeros((1,))
    ones = numpy.ones((1,))
    enum_options1 = {0: "dummy", 1: "fake"}
    enum_options2 = {0: "rubbish", 2: "nonsense"}
    data1 = data.ArchiveData("dummy", zeros, zeros, zeros, enum_options1)
    data2 = data.ArchiveData("dummy", zeros, zeros, zeros, enum_options1)
    assert data1 == data2
    data3 = data.ArchiveData("dummy", ones, zeros, zeros, enum_options1)
    data4 = data.ArchiveData("dummy", ones, ones, zeros, enum_options1)
    data5 = data.ArchiveData("dummy", ones, ones, ones, enum_options1)
    data6 = data.ArchiveData("dummy", ones, ones, ones, enum_options2)
    assert data1 != data3 != data4 != data5 != data6

def test_ArchiveData_iterates_multiple_times():
    array = numpy.zeros((1,))
    enum_options = OrderedDict([(0, "Good"), (1, "Bad")])
    zero_event = data.ArchiveEvent("dummy", array, array, array, enum_options)
    for i in range(3):
        array = numpy.zeros((3,))
        d = data.ArchiveData("dummy", array, array, array, enum_options)
        for event in d:
            assert event == zero_event

def test_data_from_events_returns_empty_data_if_no_events_provided(
    dummy_pv, empty_data
):
    assert data.data_from_events(dummy_pv, []) == empty_data

def test_data_from_events_handles_one_event(dummy_pv, event_1d, data_1d):
    assert data.data_from_events(dummy_pv, (event_1d,)) == data_1d

def test_data_from_events_handles_two_events(
    dummy_pv, event_1d, event_1d_alt, data_2_events
):
    result = data.data_from_events(dummy_pv, (event_1d, event_1d_alt))
    assert result == data_2_events

def test_data_from_events_handles_two_2d_events(
    dummy_pv, event_2d, event_2d_alt, data_2d_2_events
):
    assert data.data_from_events(dummy_pv, (event_2d, event_2d_alt)) == data_2d_2_events

def test_parse_enum_options_expected_output():
    test_input = {
        "name": "CS-CS-MSTAT-01:MODE",
        "DRVH": "0.0",
        "ENUM_1": "Injection",
        "HIGH": "0.0",
        "HIHI": "0.0",
        "ENUM_0": "Shutdown",
        "DRVL": "0.0",
        "ENUM_3": "Mach. Dev.",
        "PREC": "0.0",
        "LOLO": "0.0",
        "ENUM_2": "No Beam",
        "LOPR": "0.0",
        "ENUM_5": "Special",
        "ENUM_4": "User",
        "HOPR": "0.0",
        "ENUM_7": "Unknown",
        "ENUM_6": "BL Startup",
        "LOW": "0.0",
        "NELM": "1",
        "ENUM_8": "Eight",
        "ENUM_9": "Nine",
        "ENUM_10": "Ten",
        "ENUM_11": "Eleven",
    }
    expect = OrderedDict(
        [
            (0, "Shutdown"),
            (1, "Injection"),
            (2, "No Beam"),
            (3, "Mach. Dev."),
            (4, "User"),
            (5, "Special"),
            (6, "BL Startup"),
            (7, "Unknown"),
            (8, "Eight"),
            (9, "Nine"),
            (10, "Ten"),
            (11, "Eleven"),
        ]
    )
    result = data.parse_enum_options(test_input)
    assert result == expect

max_stars_repo_name: willrogers/aapy

max_issues_repo_name: willrogers/aapy

max_forks_repo_name: willrogers/aapy