text:
"""Test tools"""

from __future__ import division, print_function
import itertools
import logging

import numpy as np
from osgeo import gdal, osr

import buzzard as buzz

LOGGER = logging.getLogger('buzzard')

if int(gdal.VersionInfo()) >= 3040000:
    EPSG_2154_SRS = """PROJCS["RGF93 v1 / Lambert-93",GEOGCS["RGF93 v1",DATUM["Reseau_Geodesique_Francais_1993_v1",SPHEROID["GRS 1980",6378137,298.257222101,AUTHORITY["EPSG","7019"]],AUTHORITY["EPSG","6171"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4171"]],PROJECTION["Lambert_Conformal_Conic_2SP"],PARAMETER["latitude_of_origin",46.5],PARAMETER["central_meridian",3],PARAMETER["standard_parallel_1",49],PARAMETER["standard_parallel_2",44],PARAMETER["false_easting",700000],PARAMETER["false_northing",6600000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["Easting",EAST],AXIS["Northing",NORTH],AUTHORITY["EPSG","2154"]]""" # pylint: disable=line-too-long
    EPSG_2154_CENTERED_SRS = """PROJCS["RGF93 / Lambert-93",GEOGCS["RGF93 v1",DATUM["Reseau_Geodesique_Francais_1993_v1",SPHEROID["GRS 1980",6378137,298.257222101004,AUTHORITY["EPSG","7019"]],AUTHORITY["EPSG","6171"]],PRIMEM["Greenwich",0],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4171"]],PROJECTION["Lambert_Conformal_Conic_2SP"],PARAMETER["latitude_of_origin",46.5],PARAMETER["central_meridian",3],PARAMETER["standard_parallel_1",49],PARAMETER["standard_parallel_2",44],PARAMETER["false_easting",210646.41],PARAMETER["false_northing",12447.7999999998],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["Easting",EAST],AXIS["Northing",NORTH]]""" # pylint: disable=line-too-long
else:
    EPSG_2154_SRS = """PROJCS["RGF93 / Lambert-93",GEOGCS["RGF93",DATUM["Reseau_Geodesique_Francais_1993",SPHEROID["GRS 1980",6378137,298.257222101,AUTHORITY["EPSG","7019"]],TOWGS84[0,0,0,0,0,0,0],AUTHORITY["EPSG","6171"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4171"]],PROJECTION["Lambert_Conformal_Conic_2SP"],PARAMETER["standard_parallel_1",49],PARAMETER["standard_parallel_2",44],PARAMETER["latitude_of_origin",46.5],PARAMETER["central_meridian",3],PARAMETER["false_easting",700000],PARAMETER["false_northing",6600000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH],AUTHORITY["EPSG","2154"]]""" # pylint: disable=line-too-long
    EPSG_2154_CENTERED_SRS = """PROJCS["RGF93 / Lambert-93",GEOGCS["RGF93",DATUM["Reseau_Geodesique_Francais_1993",SPHEROID["GRS 1980",6378137,298.257222101,AUTHORITY["EPSG","7019"]],TOWGS84[0,0,0,0,0,0,0],AUTHORITY["EPSG","6171"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4171"]],PROJECTION["Lambert_Conformal_Conic_2SP"],PARAMETER["standard_parallel_1",49],PARAMETER["standard_parallel_2",44],PARAMETER["latitude_of_origin",46.5],PARAMETER["central_meridian",3],PARAMETER["false_easting",210646.41],PARAMETER["false_northing",12447.79999999981],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH]]""" # pylint: disable=line-too-long

_SRS = [
    ('EPSG:2154', 2154, 489353.59, 6587552.20, -378305.81, 6093283.21, 1212610.74, 7186901.68, EPSG_2154_SRS),
    ('EPSG:2154 centered', 0, 0, 0,-867659.4, -494268.99, 723257.15, 599349.48, EPSG_2154_CENTERED_SRS),
    ('EPSG:27561', 27561, 552587.60, 215784.19, 64398.20, 74408.53, 1012167.98, 398560.88, """PROJCS["NTF (Paris) / Lambert Nord France",GEOGCS["NTF (Paris)",DATUM["Nouvelle_Triangulation_Francaise_Paris",SPHEROID["Clarke 1880 (IGN)",6378249.2,293.4660212936265,AUTHORITY["EPSG","7011"]],TOWGS84[-168,-60,320,0,0,0,0],AUTHORITY["EPSG","6807"]],PRIMEM["Paris",2.33722917,AUTHORITY["EPSG","8903"]],UNIT["grad",0.01570796326794897,AUTHORITY["EPSG","9105"]],AUTHORITY["EPSG","4807"]],PROJECTION["Lambert_Conformal_Conic_1SP"],PARAMETER["latitude_of_origin",55],PARAMETER["central_meridian",0],PARAMETER["scale_factor",0.999877341],PARAMETER["false_easting",600000],PARAMETER["false_northing",200000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH],AUTHORITY["EPSG","27561"]]"""), # pylint: disable=line-too-long
    ('EPSG:5627', 5627, 836179.68, 5889250.81, -927864.00, 3972451.32, 1395572.16, 8094223.77, """PROJCS["ED50 / TM 6 NE",GEOGCS["ED50",DATUM["European_Datum_1950",SPHEROID["International 1924",6378388,297,AUTHORITY["EPSG","7022"]],TOWGS84[-87,-98,-121,0,0,0,0],AUTHORITY["EPSG","6230"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4230"]],PROJECTION["Transverse_Mercator"],PARAMETER["latitude_of_origin",0],PARAMETER["central_meridian",6],PARAMETER["scale_factor",0.9996],PARAMETER["false_easting",500000],PARAMETER["false_northing",0],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["Easting",EAST],AXIS["Northing",NORTH],AUTHORITY["EPSG","5627"]]"""), # pylint: disable=line-too-long
    ('EPSG:3950', 3950, 1488760.97, 8798334.99, 613684.96, 8306333.45, 2211114.59, 9398784.57, """PROJCS["RGF93 / CC50",GEOGCS["RGF93",DATUM["Reseau_Geodesique_Francais_1993",SPHEROID["GRS 1980",6378137,298.257222101,AUTHORITY["EPSG","7019"]],TOWGS84[0,0,0,0,0,0,0],AUTHORITY["EPSG","6171"]],PRIMEM["Greenwich",0,AUTHORITY["EPSG","8901"]],UNIT["degree",0.0174532925199433,AUTHORITY["EPSG","9122"]],AUTHORITY["EPSG","4171"]],PROJECTION["Lambert_Conformal_Conic_2SP"],PARAMETER["standard_parallel_1",49.25],PARAMETER["standard_parallel_2",50.75],PARAMETER["latitude_of_origin",50],PARAMETER["central_meridian",3],PARAMETER["false_easting",1700000],PARAMETER["false_northing",9200000],UNIT["metre",1,AUTHORITY["EPSG","9001"]],AXIS["X",EAST],AXIS["Y",NORTH],AUTHORITY["EPSG","3950"]]"""), # pylint: disable=line-too-long
]

SRS = [
    dict(
        name=name,
        epsg=epsg,
        cx=cx, cy=cy, minx=minx, miny=miny, maxx=maxx, maxy=maxy, wkt=wkt,
        w=maxx - minx, h=maxy - miny
    )
    for name, epsg, cx, cy, minx, miny, maxx, maxy, wkt in _SRS
]

def get_srs_by_name(name):
    return next(
        srs
        for srs in SRS
        if srs['name'] == name
    )

def eq(*items, **kwargs):
    """Numeric items are all almost equal"""
    tol = kwargs.pop('tol', 10e-5)
    assert not kwargs
    for a, b in itertools.combinations(items, 2):
        diff = np.abs(np.asarray(a) - np.asarray(b))
        if not (diff <= tol).all():
            return False
    return True

def eqall(items, **kwargs):
    """Numeric items are all almost equal"""
    return eq(*items, **kwargs)

def fpeq(*items, **kwargs):
    """Footprint items are all almost equal"""
    tol = kwargs.pop('tol', 10e-5)
    assert not kwargs
    for a, b in itertools.combinations(items, 2):
        diff = np.abs(a.gt - b.gt)
        if not (diff <= tol).all():
            return False
        diff = np.abs(a.rsize - b.rsize)
        if not (diff <= tol).all():
            return False
    return True

def sreq(*items, verbose_on=None):
    """SR items are all equal"""
    v = tuple(map(int, gdal.__version__.split('.')))
    for a, b in itertools.combinations(items, 2):
        a = osr.SpatialReference(osr.GetUserInputAsWKT(a))
        if v < (3,):
            a.StripCTParms()
        a = a.ExportToProj4()
        a = osr.SpatialReference(osr.GetUserInputAsWKT(a))
        if v < (3,):
            a.StripCTParms()

        b = osr.SpatialReference(osr.GetUserInputAsWKT(b))
        if v < (3,):
            b.StripCTParms()
        b = b.ExportToProj4()
        b = osr.SpatialReference(osr.GetUserInputAsWKT(b))
        if v < (3,):
            b.StripCTParms()

        res = bool(a.IsSame(b))
        if res is verbose_on:
            print('')
            print(a.ExportToPrettyWkt())
            print('---------- vs ----------')
            print(b.ExportToPrettyWkt())
            print('')

        if not res:
            return False
    return True

def poly_relation(a, b):
    """Describe 2 polygons relation"""
    rels = {
        'a contains b': a.contains(b),
        'a crosses b': a.crosses(b),
        'a disjoint b': a.disjoint(b),
        'a intersects b': a.intersects(b),
        'a touches b': a.touches(b),
        'a within b': a.within(b),
        'a covers b': a.covers(b),
        'a overlaps b': a.overlaps(b),
        'b contains a': b.contains(a),
        'b crosses a': b.crosses(a),
        'b disjoint a': b.disjoint(a),
        'b intersects a': b.intersects(a),
        'b touches a': b.touches(a),
        'b within a': b.within(a),
        'b covers a': b.covers(a),
        'b overlaps a': b.overlaps(a),
    }
    return ', '.join(
        k for k, v in rels.items() if v
    )

ROOT_TL = np.asarray([296455., 71495.])

def make_tif(path, tloffset=(0, 0), reso=(0.25, -0.25), rsize=(20, 10),
             proj=SRS[0]['wkt'], channel_count=1, dtype=gdal.GDT_Float32):
    """Create a tiff files and return info about it"""
    tl = ROOT_TL + tloffset
    reso = np.asarray(reso)
    fp = buzz.Footprint(tl=tl, rsize=rsize, size=np.abs(reso * rsize))
    x, y = fp.meshgrid_spatial
    x = np.abs(x) - abs(ROOT_TL[0])
    y = abs(ROOT_TL[1]) - np.abs(y)
    x *= 15
    y *= 15
    a = x / 2 + y / 2
    a = np.around(a).astype('float32')
    driver = gdal.GetDriverByName('GTiff')
    dataset = driver.Create(path, rsize[0], rsize[1], channel_count, dtype)
    dataset.SetGeoTransform(fp.gt)
    dataset.SetProjection(proj)
    for i in range(channel_count):
        dataset.GetRasterBand(i + 1).WriteArray(a)
        dataset.GetRasterBand(i + 1).SetNoDataValue(-32000.)
    dataset.FlushCache()
    return path, fp, a

def make_tif2(path, reso=(1., -1.), rsize=(10, 10), tl=(0., 10.),
              proj=SRS[0]['wkt'], channel_count=1, dtype=gdal.GDT_Float32,
              nodata=-32000, nodata_border_size=(0, 0, 0, 0)):
    """Create a tiff files"""
    reso = np.asarray(reso)
    fp = buzz.Footprint(tl=tl, rsize=rsize, size=np.abs(reso * rsize))
    x, y = fp.meshgrid_raster
    a = x + y
    if nodata_border_size != 0:
        l, r, t, b = nodata_border_size
        if t != 0:
            a[None:t, None:None] = nodata
        if b != 0:
            a[-b:None, None:None] = nodata
        if l != 0:
            a[None:None, None:l] = nodata
        if r != 0:
            a[None:None, -r:None] = nodata

    LOGGER.info('TIFF ARRAY:%s\n', a)
    gdal.UseExceptions()
    driver = gdal.GetDriverByName('GTiff')
    dataset = driver.Create(path, int(rsize[0]), int(rsize[1]), channel_count, dtype)
    dataset.SetGeoTransform(fp.gt)
    dataset.SetProjection(proj)
    for i in range(channel_count):
        dataset.GetRasterBand(i + 1).WriteArray(a)
        dataset.GetRasterBand(i + 1).SetNoDataValue(nodata)
    dataset.FlushCache()

def dump_tiles(tiles):
    """Print tiles"""
    for y, x in itertools.product(range(tiles.shape[0]), range(tiles.shape[1])):
        print(tiles[y, x])

def assert_tiles_eq(mata, matb, tol=10e-5):
    """Assert that two matrices contain the same tiles"""
    mata = np.asarray(mata)
    matb = np.asarray(matb)

    def _dump():
        print('--------')
        dump_tiles(mata)
        print('--------')
        dump_tiles(matb)

    if not mata.shape == matb.shape:
        _dump()
        assert mata.shape == matb.shape
    for a, b in zip(mata.flatten(), matb.flatten()):
        if not fpeq(a, b, tol=tol):
            _dump()
            assert fpeq(a, b, tol=tol)


text:
#from __future__ import absolute_import, division, print_function, unicode_literals

# import numpy as np
# from numpy.polynomial import legendre

from webbpsf_ext.maths import jl_poly, jl_poly_fit, fit_bootstrap

ext: py

max_stars_repo_name: JarronL/pyNRC

max_issues_repo_name: JarronL/pyNRC

max_forks_repo_name: JarronL/pyNRC

reason: import numpy,from numpy


text:
from __future__ import division, print_function
import numpy as np

def crystal_vectors(xtal):
    '''
    Calculates the direct and reciprocal unit vectors for a given
    crystal data obtained from xraylib.Crystal_GetCrystal(). Assumes the direct
    vector a1 to be perpendicular to x-axis and a2 to lie in xy-plane.

    Parameters
    ----------

    xtal : dict
        Representation of a crystal returned by xraylib.Crystal_GetCrystal()

    Returns
    -------

        a_matrix : 3x3 Numpy array
            Matrix containing the direct unit vectors as columns in
            angstroms.

        b_matrix : 3x3 Numpy array
            Matrix containing the reciprocal unit vectors as columns in
            inverse angstroms.

    '''

    a = xtal['a']; b = xtal['b']; c = xtal['c']
    alpha = np.radians(xtal['alpha'])
    beta  = np.radians(xtal['beta'])
    gamma = np.radians(xtal['gamma'])

    #calculate the direct vectors
    a1 = a*np.array([[1, 0, 0]])
    a2 = b*np.array([[np.cos(gamma), np.sin(gamma), 0]])

    aux1 = np.cos(beta) * np.sin(gamma)
    aux2 = np.cos(alpha) - np.cos(beta) * np.cos(gamma)
    aux3 = np.sqrt(  np.sin(gamma)**2 - np.cos(alpha)**2 - np.cos(beta)**2
                   - 2*np.cos(alpha)*np.cos(beta)*np.cos(gamma))

    a3 = c/np.sin(gamma)*np.array([[aux1,aux2,aux3]])

    #calculate the reciprocal vectors
    volume = np.dot(np.cross(a1, a2), a3.T)
    b1 = 2*np.pi*np.cross(a2, a3)/volume
    b2 = 2*np.pi*np.cross(a3, a1)/volume
    b3 = 2*np.pi*np.cross(a1 ,a2)/volume

    a_matrix = np.concatenate((a1, a2, a3)).T
    b_matrix = np.concatenate((b1, b2, b3)).T

    return a_matrix, b_matrix

max_stars_repo_name: aripekka/pyTTE

max_issues_repo_name: aripekka/pyTTE

max_forks_repo_name: aripekka/pyTTE

reason: import numpy


text:
# Copyright (c) 2019
# Manuel Cherep <manuel.cherep@epfl.ch>

"""
This module is intended for processing the hand db dataset
from CMU to save the data in the format needed to train.
"""

import matplotlib.pyplot as plt
import numpy as np
import os
import cv2
import pandas as pd
import shutil
import random
import json

DATA_PATH = '../data/handdb'  # Path for the final dataset
CLASS = 1  # There's only one class (i.e. hand)
TRAIN_PER = .7
VAL_PER = .1
TEST_PER = .2

def main():
    create_labels()
    clean()
    split_train_test()

def create_labels():
    """ This function saves the labels for all images in a .csv """

    labels = []
    data_dirs = [os.path.join(DATA_PATH, 'manual_train'),
                 os.path.join(DATA_PATH, 'manual_test')]

    # Iterate over all dirs in the original dataset
    for data_dir in data_dirs:
        # Select only the json files
        json_files = [f for f in os.listdir(data_dir) if f.endswith('.json')]

        for json_f in json_files:
            with open(os.path.join(data_dir, json_f)) as json_file:
                data = json.load(json_file)

            # Use only valid points (i.e. 3rd argument is 1)
            hand_pts = np.array(data['hand_pts'])
            hand_pts = hand_pts[hand_pts[:, 2] == 1]

            [xmin, ymin, _] = np.min(hand_pts, axis=0)
            [xmax, ymax, _] = np.max(hand_pts, axis=0)

            width = xmax-xmin
            height = ymax-ymin

            # Resize boxes because points are inside the hand
            # and the box needs to be surrounding
            xmin = xmin-width*0.1
            ymin = ymin-height*0.1
            xmax = xmax+width*0.1
            ymax = ymax+height*0.1

            # Remove the l/r and .json from the filename
            filename = json_f[:-7] + ".jpg"
            # Save as bounding box
            labels.append({"frame": filename,
                           "xmin": int(np.floor(xmin)),
                           "xmax": int(np.ceil(xmax)),
                           "ymin": int(np.floor(ymin)),
                           "ymax": int(np.ceil(ymax)),
                           "class_id": CLASS})

    # Pandas dataframe to save in .csv
    labels_df = pd.DataFrame(labels)
    labels_df.to_csv(os.path.join(DATA_PATH, 'labels.csv'), index=False)

def clean():
    """ This function relocates images and removes useless files """

    data_dirs = [os.path.join(DATA_PATH, 'manual_train'),
                 os.path.join(DATA_PATH, 'manual_test')]

    # Move all images to DATA_PATH
    for data_dir in data_dirs:
        for f in os.listdir(data_dir):
            if ".jpg" in f:
                # Move only .jpg
                os.rename(os.path.join(data_dir, f),
                          os.path.join(DATA_PATH, f[:-6] + ".jpg"))

    # Remove unnecessary files (i.e. all except .jpg or .csv)
    useless_files = list(
        filter(lambda f: (".jpg" not in f) and (".csv" not in f), os.listdir(DATA_PATH)))
    # List is only here to avoid lazy evaluation
    list(map(lambda f: remove_file_or_directory(
        os.path.join(DATA_PATH, f)), useless_files))

def split_train_test():
    # Create directories for the data
    os.makedirs(os.path.join(DATA_PATH, 'train'))
    os.makedirs(os.path.join(DATA_PATH, 'validation'))
    os.makedirs(os.path.join(DATA_PATH, 'test'))

    # Read .csv with labels
    labels_df = pd.read_csv(os.path.join(DATA_PATH, 'labels.csv'))

    images = [f for f in os.listdir(DATA_PATH) if '.jpg' in f]
    # Shuffle dataframe rows before splitting
    random.seed(1)
    random.shuffle(images)
    # Divide in train and validation. The rest is for testing
    train_idxs = int(TRAIN_PER*len(images))
    val_idxs = int((TRAIN_PER+VAL_PER)*len(images))
    train_imgs = images[:train_idxs]
    val_imgs = images[train_idxs:val_idxs]
    test_imgs = images[val_idxs:]

    train_df = labels_df[labels_df['frame'].isin(train_imgs)]
    val_df = labels_df[labels_df['frame'].isin(val_imgs)]
    test_df = labels_df[labels_df['frame'].isin(test_imgs)]

    # Save labels in different directories
    train_df.to_csv(os.path.join(DATA_PATH, 'train/labels.csv'), index=False)
    val_df.to_csv(os.path.join(
        DATA_PATH, 'validation/labels.csv'), index=False)
    test_df.to_csv(os.path.join(DATA_PATH, 'test/labels.csv'), index=False)
    # Remove old labels.csv
    os.remove(os.path.join(DATA_PATH, 'labels.csv'))

    # Move all images to corresponding directory
    for f in os.listdir(DATA_PATH):
        if '.jpg' in f:
            if f in train_imgs:
                os.rename(os.path.join(DATA_PATH, f),
                          os.path.join(DATA_PATH, 'train', f))
            elif f in val_imgs:
                os.rename(os.path.join(DATA_PATH, f),
                          os.path.join(DATA_PATH, 'validation', f))
            elif f in test_imgs:
                os.rename(os.path.join(DATA_PATH, f),
                          os.path.join(DATA_PATH, 'test', f))

def plot_bbox(img_filename):
    """ This function plots an image with the corresponding boxes """

    # Read the image
    img = cv2.imread(os.path.join(DATA_PATH, 'test', img_filename))
    img = img[:, :, ::-1]  # RGB -> BGR

    # Read .csv with labels
    labels_df = pd.read_csv(os.path.join(DATA_PATH, 'test', 'labels.csv'))
    boxes_df = labels_df[labels_df.frame == img_filename]

    plt.figure(figsize=(20, 12))
    current_axis = plt.gca()

    for idx, row in boxes_df.iterrows():
        xmin = row.xmin
        xmax = row.xmax
        ymin = row.ymin
        ymax = row.ymax

        # Add rectangle to the plot
        current_axis.add_patch(plt.Rectangle(
            (xmin, ymin), xmax-xmin, ymax-ymin, color='g', fill=False, linewidth=2))

    plt.imshow(img)
    return plt.show()

def remove_file_or_directory(f):
    if os.path.isfile(f):
        os.remove(f)
    if os.path.exists(f):
        shutil.rmtree(f)

if __name__ == "__main__":
    main()

max_stars_repo_name: mcherep/handde

max_issues_repo_name: mcherep/handde

max_forks_repo_name: mcherep/handdet


text:
import sys
from typing import List, Dict, Any, Tuple, Set
import logging
import numpy as np
import pandas as pd

from d3m import container, utils
from d3m.metadata.pipeline import Pipeline, PrimitiveStep
from d3m.metadata.base import ArgumentType
from d3m.metadata import hyperparams

from distil.primitives.ragged_dataset_loader import RaggedDatasetLoaderPrimitive
from distil.primitives.timeseries_reshaper import TimeSeriesReshaperPrimitive
from distil.primitives.timeseries_neighbours import TimeSeriesNeighboursPrimitive

from common_primitives.dataset_to_dataframe import DatasetToDataFramePrimitive
from common_primitives.remove_columns import RemoveColumnsPrimitive
from common_primitives.construct_predictions import ConstructPredictionsPrimitive
from common_primitives.column_parser import ColumnParserPrimitive
from common_primitives.extract_columns_semantic_types import ExtractColumnsBySemanticTypesPrimitive

PipelineContext = utils.Enum(value='PipelineContext', names=['TESTING'], start=1)

def create_pipeline(metric: str) -> Pipeline:
    input_val = 'steps.{}.produce'

    # create the basic pipeline
    ts_pipeline = Pipeline(context=PipelineContext.TESTING)
    ts_pipeline.add_input(name='inputs')

    # 0 - Extract ragged dataset to a dataframe
    step = PrimitiveStep(primitive_description=RaggedDatasetLoaderPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='inputs.0')
    step.add_output('produce')
    step.add_output('produce_collection')
    ts_pipeline.add_step(step)

    # 1 - Parse columns.
    step = PrimitiveStep(primitive_description=ColumnParserPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.0.produce')
    step.add_output('produce')
    semantic_types = ('http://schema.org/Boolean', 'http://schema.org/Integer', 'http://schema.org/Float',
                      'https://metadata.datadrivendiscovery.org/types/FloatVector')
    step.add_hyperparameter('parse_semantic_types', ArgumentType.VALUE, semantic_types)
    ts_pipeline.add_step(step)

    # 2 - Extract attributes
    step = PrimitiveStep(primitive_description=ExtractColumnsBySemanticTypesPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.1.produce')
    step.add_output('produce')
    step.add_hyperparameter('semantic_types', ArgumentType.VALUE, ('https://metadata.datadrivendiscovery.org/types/Attribute',))
    ts_pipeline.add_step(step)

    # 3- Extract targets
    step = PrimitiveStep(primitive_description=ExtractColumnsBySemanticTypesPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.1.produce')
    step.add_output('produce')
    target_types = ('https://metadata.datadrivendiscovery.org/types/Target', 'https://metadata.datadrivendiscovery.org/types/TrueTarget')
    step.add_hyperparameter('semantic_types', ArgumentType.VALUE, target_types)
    ts_pipeline.add_step(step)

    # 4 - Reformats timeseries data if sparse, truncates if required
    step = PrimitiveStep(primitive_description=TimeSeriesReshaperPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.0.produce_collection')
    step.add_output('produce')
    ts_pipeline.add_step(step)

    # 5 - Classification / regression of series data
    step = PrimitiveStep(primitive_description=TimeSeriesNeighboursPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.4.produce')
    step.add_argument(name='outputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.3.produce')
    step.add_output('produce')
    step.add_hyperparameter('metric', ArgumentType.VALUE, metric)
    ts_pipeline.add_step(step)

    # step 6 - convert predictions to expected format
    step = PrimitiveStep(primitive_description=ConstructPredictionsPrimitive.metadata.query())
    step.add_argument(name='inputs', argument_type=ArgumentType.CONTAINER, data_reference='steps.5.produce')
    step.add_argument(name='reference', argument_type=ArgumentType.CONTAINER, data_reference='steps.1.produce')
    step.add_output('produce')
    ts_pipeline.add_step(step)

    # Adding output step to the pipeline
    ts_pipeline.add_output(name='output', data_reference='steps.6.produce')

    return ts_pipeline


text:
from __future__ import absolute_import
import numpy as np
import pandas as pd
import random
import scipy
import copy
from pysurvival import utils
from pysurvival.models import BaseModel

# %matplotlib inline

# List of Survival Distributions
DISTRIBUTIONS = ['Exponential',
                 'Weibull',
                 'Gompertz',
                 'Log-Logistic',
                 'Log-Normal', ]

# List of risk types
RISK_TYPES = ['Linear', 'Square', 'Gaussian']

class SimulationModel(BaseModel):
    """
    A general framework for simulating right-censored survival data
    for proportional hazards models by incorporating
        * a baseline hazard function from a known survival distribution,
        * a set of covariates.

    The framework is based on "Generating Survival Times to Simulate
    Cox Proportional Hazards Models"
    https://www.ncbi.nlm.nih.gov/pubmed/22763916

    The formula for the different survival times and functions, and hazard
    functions can be found at :
    http://data.princeton.edu/pop509/ParametricSurvival.pdf

    Parameters:
    -----------

    * survival_distribution: string (default = 'exponential')
        Defines a known survival distribution. The available options are:
            - Exponential
            - Weibull
            - Gompertz
            - Log-Logistic
            - Log-Normal

    * risk_type: string (default='linear')
        Defines the type of risk function. The available options are:
            - Linear
            - Square
            - Gaussian

    * alpha: double (default = 1.)
         the scale parameter

    * beta: double (default = 1.)
         the shape parameter

    * bins: int (default=100)
        the number of bins of the time axis

    * censored_parameter: double (default = 1.)
         coefficient used to calculate the censored distribution. This
         distribution is a normal such that N(loc=censored_parameter, scale=5)

    * risk_parameter: double (default = 1.)
        Scaling coefficient of the risk score such that:
            - linear: r(x) = exp(<x, W>)
            - square: r(x) = exp(risk_parameter*(<x, W>)^2)
            - gaussian: r(x) = exp( exp(-(<x, W>)^2/risk_parameter) )
        <.,.> is the dot product
    """

    def __init__(self, survival_distribution='exponential',
                 risk_type='linear', censored_parameter=1., alpha=1, beta=1.,
                 bins=100, risk_parameter=1.):

        # Saving the attributes
        self.censored_parameter = censored_parameter
        self.alpha = alpha
        self.beta = beta
        self.risk_parameter = risk_parameter
        self.bins = bins
        self.features = []

        # Checking risk_type
        if any([risk_type.lower() == r.lower() for r in RISK_TYPES]):
            self.risk_type = risk_type
        else:
            error = "{} isn't a valid risk type. "
            error += "Only {} are currently available."
            error = error.format(risk_type, ", ".join(RISK_TYPES))
            raise NotImplementedError(error)

        # Checking distribution
        if any([survival_distribution.lower() == d.lower() \
                for d in DISTRIBUTIONS]):
            self.survival_distribution = survival_distribution
        else:
            error = "{} isn't a valid survival distribution. "
            error += "Only {} are currently available."
            error = error.format(survival_distribution, ", ".join(DISTRIBUTIONS))
            raise NotImplementedError(error)

        # Initializing the elements from BaseModel
        super(SimulationModel, self).__init__(auto_scaler=True)

    @staticmethod
    def random_data(N):
        """
        Generating a array of size N from a random distribution -- the available
        distributions are:
            * binomial,
            * chisquare,
            * exponential,
            * gamma,
            * normal,
            * uniform
            * laplace
        """

        index = np.random.binomial(n=4, p=0.5)
        distributions = {
            'binomial_a': np.random.binomial(n=20, p=0.6, size=N),
            'binomial_b': np.random.binomial(n=200, p=0.6, size=N),
            'chisquare': np.random.chisquare(df=10, size=N),
            'exponential_a': np.random.exponential(scale=0.1, size=N),
            'exponential_b': np.random.exponential(scale=0.01, size=N),
            'gamma': np.random.gamma(shape=2., scale=2., size=N),
            'normal_a': np.random.normal(loc=-1.0, scale=5.0, size=N),
            'normal_b': np.random.normal(loc=10.0, scale=10.0, size=N),
            'uniform_a': np.random.uniform(low=-2.0, high=10.0, size=N),
            'uniform_b': np.random.uniform(low=-20.0, high=100.0, size=N),
            'laplace': np.random.laplace(loc=0.0, scale=1.0, size=N)
        }

        list_distributions = copy.deepcopy(list(distributions.keys()))
        random.shuffle(list_distributions)
        key = list_distributions[index]
        return key, distributions[key]

    def time_function(self, BX):
        """
        Calculating the survival times based on the given distribution
        T = H^(-1)( -log(U)/risk_score ), where:
            * H is the cumulative baseline hazard function
                (H^(-1) is the inverse function)
            * U is a random variable uniform - Uni[0,1].

        The method is inspired by https://gist.github.com/jcrudy/10481743
        """

        # Calculating scale coefficient using the features
        num_samples = BX.shape[0]
        lambda_exp_BX = np.exp(BX) * self.alpha
        lambda_exp_BX = lambda_exp_BX.flatten()

        # Generating random uniform variables
        U = np.random.uniform(0, 1, num_samples)

        # Exponential
        if self.survival_distribution.lower().startswith('exp'):
            self.survival_distribution = 'Exponential'
            return - np.log(U) / (lambda_exp_BX)

        # Weibull
        elif self.survival_distribution.lower().startswith('wei'):
            self.survival_distribution = 'Weibull'
            return np.power(- np.log(U) / (lambda_exp_BX), 1. / self.beta)

        # Gompertz
        elif self.survival_distribution.lower().startswith('gom'):
            self.survival_distribution = 'Gompertz'
            return (1. / self.beta) * \
                   np.log(1 - self.beta * np.log(U) / (lambda_exp_BX))

        # Log-Logistic
        elif 'logistic' in self.survival_distribution.lower():
            self.survival_distribution = 'Log-Logistic'
            return np.power(U / (1. - U), 1. / self.beta) / (lambda_exp_BX)

        # Log-Normal
        elif 'normal' in self.survival_distribution.lower():
            self.survival_distribution = 'Log-Normal'
            W = np.random.normal(0, 1, num_samples)
            return lambda_exp_BX * np.exp(self.beta * W)

    def hazard_function(self, t, BX):
        """ Calculating the hazard function based on the given distribution """

        # Calculating scale coefficient using the features
        _lambda = self.alpha * np.exp(BX)

        # Exponential
        if self.survival_distribution.lower().startswith('exp'):
            return np.repeat(_lambda, len(t))

            # Weibull
        elif self.survival_distribution.lower().startswith('wei'):
            return _lambda * self.beta * np.power(t, self.beta - 1)

            # Gompertz
        elif self.survival_distribution.lower().startswith('gom'):
            return _lambda * np.exp(self.beta * t)

        # Log-Logistic
        elif self.survival_distribution.lower().endswith('logistic'):
            numerator = _lambda * self.beta * np.power((_lambda * t), self.beta - 1)
            denominator = (1 + np.power((_lambda * t), self.beta))
            return numerator / denominator

        # Log-Normal
        elif self.survival_distribution.lower().endswith('normal'):
            arg_normal = (np.log(t) - np.log(_lambda)) / self.beta
            numerator = (1. / (t * self.beta)) * scipy.stats.norm.pdf(arg_normal)
            denominator = 1. - scipy.stats.norm.cdf(arg_normal)
            return numerator / denominator

    def survival_function(self, t, BX):
        """
        Calculating the survival function based on the given
        distribution
        """

        # Calculating scale coefficient using the features
        _lambda = self.alpha * np.exp(BX)

        # Exponential
        if self.survival_distribution.lower().startswith('exp'):
            return np.exp(-t * _lambda)

        # Weibull
        elif self.survival_distribution.lower().startswith('wei'):
            return np.exp(-np.power(t, self.beta) * _lambda)

        # Gompertz
        elif self.survival_distribution.lower().startswith('gom'):
            return np.exp(-_lambda / self.beta * (np.exp(self.beta * t) - 1))

        # Log-Logistic
        elif self.survival_distribution.lower().endswith('logistic'):
            return 1. / (1. + np.power(_lambda * t, self.beta))

        # Log-Normal
        elif self.survival_distribution.lower().endswith('normal'):
            arg_cdf = (np.log(t) - np.log(_lambda)) / self.beta
            return 1. - scipy.stats.norm.cdf(arg_cdf)

    def risk_function(self, x_std):
        """ Calculating the risk function based on the given risk type """

        # Dot product
        risk = np.dot(x_std, self.feature_weights)

        # Choosing the type of risk
        if self.risk_type.lower() == 'linear':
            return risk.reshape(-1, 1)

        elif self.risk_type.lower() == 'square':
            risk = np.square(risk * self.risk_parameter)

        elif self.risk_type.lower() == 'gaussian':
            risk = np.square(risk)
            risk = np.exp(- risk * self.risk_parameter)

        return risk.reshape(-1, 1)

    def generate_data(self, num_samples=100, num_features=3,
                      feature_weights=None):
        """
        Generating a dataset of simulated survival times from a given
        distribution through the hazard function using the Cox model

        Parameters:
        -----------
        * `num_samples`: **int** *(default=100)* --
            Number of samples to generate

        * `num_features`: **int** *(default=3)* --
            Number of features to generate

        * `feature_weights`: **array-like** *(default=None)* --
            list of the coefficients of the underlying Cox-Model.
            The features linked to each coefficient are generated
            from random distribution from the following list:

            * binomial
            * chisquare
            * exponential
            * gamma
            * normal
            * uniform
            * laplace

            If None then feature_weights = [1.]*num_features

        Returns:
        --------
        * dataset: pandas.DataFrame
            dataset of simulated survival times, event status and features

        Example:
        --------
        from pysurvival.models.simulations import SimulationModel

        # Initializing the simulation model
        sim = SimulationModel( survival_distribution = 'gompertz',
                               risk_type = 'linear',
                               censored_parameter = 5.0,
                               alpha = 0.01,
                               beta = 5., )

        # Generating N Random samples
        N = 1000
        dataset = sim.generate_data(num_samples = N, num_features=5)

        # Showing a few data-points
        dataset.head()
        """

        # Data parameters
        self.num_variables = num_features
        if feature_weights is None:
            self.feature_weights = [1.] * self.num_variables
            feature_weights = self.feature_weights

        else:
            feature_weights = utils.check_data(feature_weights)
            if num_features != len(feature_weights):
                error = "The length of feature_weights ({}) "
                error += "and num_features ({}) are not the same."
                error = error.format(len(feature_weights), num_features)
                raise ValueError(error)
            self.feature_weights = feature_weights

        # Generating random features
        # Creating the features
        X = np.zeros((num_samples, self.num_variables))
        columns = []
        for i in range(self.num_variables):
            key, X[:, i] = self.random_data(num_samples)
            columns.append('x_' + str(i + 1))
        X_std = self.scaler.fit_transform(X)
        BX = self.risk_function(X_std)

        # Building the survival times
        T = self.time_function(BX)
        C = np.random.normal(loc=self.censored_parameter,
                             scale=5, size=num_samples)
        C = np.maximum(C, 0.)
        time = np.minimum(T, C)
        E = 1. * (T == time)

        # Building dataset
        self.features = columns
        self.dataset = pd.DataFrame(data=np.c_[X, time, E],
                                    columns=columns + ['time', 'event'])

        # Building the time axis and time buckets
        self.times = np.linspace(0., max(self.dataset['time']), self.bins)
        self.get_time_buckets()

        # Building baseline functions
        self.baseline_hazard = self.hazard_function(self.times, 0)
        self.baseline_survival = self.survival_function(self.times, 0)

        # Printing summary message
        message_to_print = "Number of data-points: {} - Number of events: {}"
        print(message_to_print.format(num_samples, sum(E)))

        return self.dataset

    def _predict(self, x, t=None, **kwargs):
        """
        Predicting the hazard, density and survival functions

        Parameters:
        -----------
            * x: pd.Dataframe or np.ndarray or list
                x is the testing dataset containing the features
                x should not be standardized before, the model
                will take care of it
            * t: float (default=None)
                Time at which hazard, density and survival functions
                should be calculated. If None, the method returns
                the functions for all times t.
        """

        # Convert x into the right format
        x = utils.check_data(x)

        # Scaling the dataset
        if x.ndim == 1:
            x = self.scaler.transform(x.reshape(1, -1))

        elif x.ndim == 2:
            x = self.scaler.transform(x)

        else:
            # Ensuring x has 2 dimensions
            if x.ndim == 1:
                x = np.reshape(x, (1, -1))

        # Calculating risk_score, hazard, density and survival
        BX = self.risk_function(x)
        hazard = self.hazard_function(self.times, BX.reshape(-1, 1))
        survival = self.survival_function(self.times, BX.reshape(-1, 1))
        density = (hazard * survival)

        if t is None:
            return hazard, density, survival
        else:
            min_abs_value = [abs(a_j_1 - t) for (a_j_1, a_j) in self.time_buckets]
            index = np.argmin(min_abs_value)
            return hazard[:, index], density[:, index], survival[:, index]

    def predict_risk(self, x, **kwargs):
        """
        Predicting the risk score function

        Parameters:
        -----------
            * x: pd.Dataframe or np.ndarray or list
                x is the testing dataset containing the features
                x should not be standardized before, the model
                will take care of it
        """

        # Convert x into the right format
        x = utils.check_data(x)

        # Scaling the dataset
        if x.ndim == 1:
            x = self.scaler.transform(x.reshape(1, -1))

        elif x.ndim == 2:
            x = self.scaler.transform(x)

        else:
            # Ensuring x has 2 dimensions
            if x.ndim == 1:
                x = np.reshape(x, (1, -1))

        # Calculating risk_score
        risk_score = self.risk_function(x)

        return risk_score

max_stars_repo_name: zmiimz/pysurvival

max_issues_repo_name: zmiimz/pysurvival

max_forks_repo_name: zmiimz/pysurvival

reason: import numpy,import scipy


text:
import os
import stat
import glob
import math
import subprocess
import re
import sys
import string
from decimal import Decimal
from astropy.io import fits
from astropy import wcs
from astropy.coordinates import Angle
import astropy.units as u
from dateutil import parser
import numpy
import warnings
from astropy.io.fits.verify import VerifyWarning

# MODIFY THESE FIELDS AS NEEDED!
# input path *with* ending forward slash
input_path = './'

# ignore warnings
warnings.filterwarnings('ignore', category=VerifyWarning, append=True)

print 'OBJECT' + ',' + 'ITIME' + ',' + 'COADDS' + \
    ',' + 'CYCLES' + ',' + 'FITS' + ',' + 'MJD'

# get a list of all FITS files in the input directory
fitsfiles = glob.glob(input_path+'*.fits')+glob.glob(input_path+'*.fit')
# loop through all qualifying files and perform plate-solving
for fitsfile in sorted(fitsfiles):
    d1 = fits.open('%s' % fitsfile)
    d1.close()
    h1 = d1[0].header
    try:
        object = str(h1['OBJECT'])
        itime = str(h1['ITIME'])
        coadds = str(h1['CO_ADDS'])
        irafname = str(h1['IRAFNAME'])
        cycles = str(h1['CYCLES'])
        mjd = str(h1['MJD_OBS'])
        print object + ',' + itime + ',' + coadds + \
            ',' + cycles + ',' + irafname + ',' + mjd
    except KeyError:
        pass

max_stars_repo_name: mcnowinski/various-and-sundry

max_issues_repo_name: mcnowinski/various-and-sundry

max_forks_repo_name: mcnowinski/various-and-sundry

reason: import numpy,import astropy,from astropy


text:
import boto3
import pytest
from botocore.stub import Stubber

from sagemaker_studio_sparkmagic_lib.emr import EMRCluster

def test_get_cluster_happy_case_non_kerberos():
    emr = boto3.client("emr")
    with Stubber(emr) as emr_stub:
        describe_cluster_response = {
            "Cluster": {"Id": "j-3DD9ZR01DAU14", "Name": "Mycluster"}
        }
        list_instances_response = {
            "Instances": [
                {
                    "Id": "j-3DD9ZR01DAU14",
                }
            ]
        }
        emr_stub.add_response("describe_cluster", describe_cluster_response)
        emr_stub.add_response("list_instances", list_instances_response)
        emr_cluster = EMRCluster(cluster_id="j-3DD9ZR01DAU14", emr=emr)

        assert emr_cluster._cluster == describe_cluster_response["Cluster"]
        assert emr_cluster._instances == list_instances_response["Instances"]
        assert emr_cluster._sec_conf is None
        assert not emr_cluster.is_krb_cluster

def test_get_cluster_happy_case_kerberos():
    emr = boto3.client("emr")
    with Stubber(emr) as emr_stub:
        describe_cluster_response = {
            "Cluster": {
                "Id": "j-3DD9ZR01DAU14",
                "Name": "Mycluster",
                "SecurityConfiguration": "kerb-security-config",
            }
        }
        list_instances_response = {
            "Instances": [
                {
                    "Id": "j-3DD9ZR01DAU14",
                }
            ]
        }
        describe_sec_conf_response = {
            "Name": "kerb-security-config",
            "SecurityConfiguration": "{}",
        }
        emr_stub.add_response("describe_cluster", describe_cluster_response)
        emr_stub.add_response("list_instances", list_instances_response)
        emr_stub.add_response(
            "describe_security_configuration", describe_sec_conf_response
        )
        emr_cluster = EMRCluster(cluster_id="j-3DD9ZR01DAU14", emr=emr)

        assert emr_cluster._cluster == describe_cluster_response["Cluster"]
        assert emr_cluster._instances == list_instances_response["Instances"]
        assert emr_cluster._sec_conf == {}

def test_get_bad_cluster():
    emr = boto3.client("emr")
    with Stubber(emr) as emr_stub:
        list_instances_response = {
            "Instances": [
                {
                    "Id": "j-3DD9ZR01DAU14",
                }
            ]
        }
        emr_stub.add_client_error("describe_cluster", service_error_code=400)
        emr_stub.add_response("list_instances", list_instances_response)
        with pytest.raises(ValueError) as e:
            EMRCluster(cluster_id="j-3DD9ZR01DAU14", emr=emr)

def test_cluster_dedicated_krb_cluster():
    emr = boto3.client("emr")
    with Stubber(emr) as emr_stub:
        describe_cluster_response = {
            "Cluster": {
                "Id": "j-3DD9ZR01DAU14",
                "Name": "Mycluster",
                "SecurityConfiguration": "kerb-security-config",
                "KerberosAttributes": {
                    "Realm": "KTEST.COM",
                    "KdcAdminPassword": "********",
                },
                "MasterPublicDnsName": "ec2-34-222-47-14.us-west-2.compute.amazonaws.com",
            }
        }
        list_instances_response = {
            "Instances": [
                {
                    "Id": "j-3DD9ZR01DAU14",
                    "Ec2InstanceId": "i-0736242069217a485",
                    "PublicDnsName": "ec2-34-222-47-14.us-west-2.compute.amazonaws.com",
                    "PublicIpAddress": "34.222.47.14",
                    "PrivateDnsName": "ip-172-31-1-113.us-west-2.compute.internal",
                    "PrivateIpAddress": "172.31.1.113",
                }
            ]
        }
        describe_sec_conf_response = {
            "Name": "kerb-security-config",
            "SecurityConfiguration": '{"EncryptionConfiguration": {"EnableInTransitEncryption": false, '
            '"EnableAtRestEncryption": false},"AuthenticationConfiguration": {'
            '"KerberosConfiguration": {"Provider": "ClusterDedicatedKdc", '
            '"ClusterDedicatedKdcConfiguration": {"TicketLifetimeInHours": 24 }}}}',
        }
        emr_stub.add_response("describe_cluster", describe_cluster_response)
        emr_stub.add_response("list_instances", list_instances_response)
        emr_stub.add_response(
            "describe_security_configuration", describe_sec_conf_response
        )
        emr_cluster = EMRCluster(cluster_id="j-3DD9ZR01DAU14", emr=emr)
        assert emr_cluster.is_krb_cluster
        assert (
            emr_cluster.primary_node_private_dns_name()
            == "ip-172-31-1-113.us-west-2.compute.internal"
        )
        krb_props = {
            "realms": {
                "KTEST.COM": {
                    "kdc": "ip-172-31-1-113.us-west-2.compute.internal:88",
                    "admin_server": "ip-172-31-1-113.us-west-2.compute.internal:749",
                    "default_domain": "us-west-2.compute.internal",
                }
            },
            "libdefaults": {"default_realm": "KTEST.COM", "ticket_lifetime": "24h"},
            "domain_realm": {
                "us-west-2.compute.internal": "KTEST.COM",
                ".us-west-2.compute.internal": "KTEST.COM",
            },
        }
        assert emr_cluster.get_krb_conf() == krb_props

def test_cross_realm_krb_cluster():
    emr = boto3.client("emr")
    with Stubber(emr) as emr_stub:
        describe_cluster_response = {
            "Cluster": {
                "Id": "j-3DD9ZR01DAU14",
                "Name": "Mycluster",
                "SecurityConfiguration": "kerb-security-config",
                "KerberosAttributes": {
                    "Realm": "EC2.INTERNAL",
                    "KdcAdminPassword": "********",
                },
                "MasterPublicDnsName": "ec2-34-222-47-14.us-west-2.compute.amazonaws.com",
            }
        }
        list_instances_response = {
            "Instances": [
                {
                    "Id": "j-3DD9ZR01DAU14",
                    "Ec2InstanceId": "i-0736242069217a485",
                    "PublicDnsName": "ec2-34-222-47-14.us-west-2.compute.amazonaws.com",
                    "PublicIpAddress": "34.222.47.14",
                    "PrivateDnsName": "ip-172-31-1-113.us-west-2.compute.internal",
                    "PrivateIpAddress": "172.31.1.113",
                }
            ]
        }
        describe_sec_conf_response = {
            "Name": "kerb-security-config",
            "SecurityConfiguration": '{"EncryptionConfiguration": {"EnableInTransitEncryption": false, '
            '"EnableAtRestEncryption": false},"AuthenticationConfiguration": {'
            '"KerberosConfiguration": {"Provider": "ClusterDedicatedKdc", '
            '"ClusterDedicatedKdcConfiguration": {"TicketLifetimeInHours": 24,'
            '"CrossRealmTrustConfiguration": {"Realm": "KOLLOJUN.NET","Domain": "kollojun.net",'
            ' "AdminServer": "kollojun.net", "KdcServer": "kollojun.net"}}}}}',
        }
        emr_stub.add_response("describe_cluster", describe_cluster_response)
        emr_stub.add_response("list_instances", list_instances_response)
        emr_stub.add_response(
            "describe_security_configuration", describe_sec_conf_response
        )
        emr_cluster = EMRCluster(cluster_id="j-3DD9ZR01DAU14", emr=emr)
        assert emr_cluster.is_krb_cluster
        assert (
            emr_cluster.primary_node_private_dns_name()
            == "ip-172-31-1-113.us-west-2.compute.internal"
        )
        krb_props = {
            "realms": {
                "EC2.INTERNAL": {
                    "kdc": "ip-172-31-1-113.us-west-2.compute.internal:88",
                    "admin_server": "ip-172-31-1-113.us-west-2.compute.internal:749",
                    "default_domain": "us-west-2.compute.internal",
                },
                "KOLLOJUN.NET": {
                    "kdc": "kollojun.net",
                    "admin_server": "kollojun.net",
                    "default_domain": "kollojun.net",
                },
            },
            "libdefaults": {"default_realm": "EC2.INTERNAL", "ticket_lifetime": "24h"},
            "domain_realm": {
                "us-west-2.compute.internal": "EC2.INTERNAL",
                ".us-west-2.compute.internal": "EC2.INTERNAL",
                ".kollojun.net": "KOLLOJUN.NET",
                "kollojun.net": "KOLLOJUN.NET",
            },
        }
        assert emr_cluster.get_krb_conf() == krb_props

max_stars_repo_name: knaresh/sagemaker-studio-sparkmagic-lib

max_issues_repo_name: knaresh/sagemaker-studio-sparkmagic-lib

max_forks_repo_name: knaresh/sagemaker-studio-sparkmagic-lib

reason: from sage


text:
from math import factorial
from itertools import chain
from itertools import combinations

import numpy as np

import coincidencetest
from coincidencetest._coincidencetest import compute_number_of_covers
from coincidencetest._coincidencetest import binom
from coincidencetest._coincidencetest import calculate_probability_of_multicoincidence
from coincidencetest import coincidencetest
from coincidencetest import coincidencetest_matrix

def _compute_number_of_covers(set_sizes: tuple=(), ambient_size: int=0,
                             strategy: str='binomial-formula'):
    """
    For testing purposes, adds support for the "recursion" strategy. This uses
    a recurrence formula deduced by removing one element from the ambient set, with
    base cases provided by classical partition counting formulas and surjective
    function counting formulas.
    """
    if strategy in ['binomial-formula', 'brute-force']:
        return compute_number_of_covers(set_sizes=set_sizes,
                                        ambient_size=ambient_size,
                                        strategy=strategy)
    if strategy == 'recursion':
        number_subsets = len(set_sizes)
        if all(size == 1 for size in set_sizes):
            if number_subsets == ambient_size:
                return factorial(ambient_size)
            if number_subsets < ambient_size:
                return 0
            if number_subsets > ambient_size:
                return number_of_surjective_functions(
                    source_size = sum(set_sizes),
                    target_size = ambient_size,
                )
        if sum(set_sizes) == ambient_size:
            return multinomial(set_sizes)
        if sum(set_sizes) < ambient_size:
            return 0
        if ambient_size == 0:
            return 0
        if any(size > ambient_size for size in set_sizes):
            return 0
        if len(set_sizes) == 1:
            size = set_sizes[0]
            if size == ambient_size:
                return 1
            if size < ambient_size:
                return 0
            raise ValueError('ambient_size %s too small?' % ambient_size)
        if len(set_sizes) == 0:
            return 0

        counts_by_strata = [
            _compute_number_of_covers(
                set_sizes = reduce_set_sizes(set_sizes=set_sizes, which=selected),
                ambient_size = ambient_size - 1,
                strategy = 'recursion',
            )
            for selected in powerset(number_subsets) if selected != ()
        ]
        return sum(counts_by_strata)
    return None

def multinomial(subset_sizes):
    """
    Computes a multinomial coefficient.

    Parameters
    ----------
    subset_sizes : tuple
        The integer parts of a partition of an integer.

    Returns
    -------
    coefficient: int
        The value.
    """
    if len(subset_sizes) == 1:
        return 1
    binomial_factor = binom(sum(subset_sizes), subset_sizes[-1])
    multinomial_factor = multinomial(subset_sizes[:-1])
    return binomial_factor * multinomial_factor

def stirling_second_kind(ambient_size: int=0, number_parts: int=0,
                         normalized: bool=False):
    """
    Computes an (unnormalized or ordinary) Stirling number of the second kind.
    Default is unnormalized, meaning that the division by the factorial of the
    ambient size is not performed.

    Parameters
    ----------
    ambient_size : int
        The ambient size whose partitions are to be counted.
    number_parts : int
        The number of parts for partitions to consider.
    normalized : bool
        Default False. If True, the ordinary Stirling number of the second kind is
        computed, i.e. the division by the factorial of the ambient size is
        performed.

    Returns
    -------
    stirling: int
        The value.
    """
    sign = lambda x: 1 if x % 2 == 0 else -1
    terms = [
        sign(i) * binom(number_parts, i) * pow(number_parts - i, ambient_size)
        for i in range(number_parts + 1)
    ]
    if normalized:
        return sum(terms) // factorial(number_parts)
    return sum(terms)

def number_of_surjective_functions(source_size: int=0, target_size: int=0):
    """
    Computes the number of surjective functions between two sets of given sizes.

    Parameters
    ----------
    source_size : int
        The size of the domain of the functions to be counted.
    target_size: int
        The size of the codomain of the functionsn to be counted.

    Returns
    -------
    count: int
        The value.
    """
    return stirling_second_kind(
        ambient_size = source_size,
        number_parts = target_size,
    )

def powerset(set_size):
    """
    Creates a list of all subsets of the set of integers {0, 1, ... `set_size - 1`}.

    Parameters
    ----------
    set_size : int
        The size of the base set.

    Returns
    -------
    subsets : list
        All subsets.
    """
    base_set = list(range(set_size))
    return list(chain.from_iterable(
        combinations(base_set, r) for r in range(len(base_set) + 1)
    ))

def reduce_set_sizes(set_sizes: tuple=(), which: list=None):
    """
    Convenience function to decrement specific members of a tuple of integers.

    Parameters
    ----------
    set_sizes : tuple
        The integers.
    which : list
        A list of the indices of the integers to be decremented.

    Returns
    -------
    reduced : tuple
        List of integers, with some decremented by 1. 0 values are omitted.
    """
    new_sizes = [
        set_sizes[i] - (1 if i in which else 0) for i in range(len(set_sizes))
    ]
    return tuple(s for s in new_sizes if s > 0)

class TestCountingFormulas:
    sample_cases = [
        ((2, 2), 3),
        ((3, 1), 4),
        ((3, 2), 4),
        ((3, 3, 4), 7),
        ((3, 3, 5, 2), 8),
    ]

    extreme_sample_cases = [
        [20, [101, 85, 75, 110, 99], 500],
        [50, [101, 85, 75, 110, 99], 500],
        [65, [101, 85, 75, 110, 99], 500],
        [50, [101, 185, 75, 210, 99], 1000],
        [50, [101, 185, 75, 210, 99], 5000],
        [50, [101, 185, 75, 210, 99], 10000],
        [120, [201, 215, 375, 210, 250], 10000],
    ]

    @staticmethod
    def test_manually_validated_cases():
        cases = {
            ((2, 2), 3) : 6,
        }
        for (set_sizes, ambient_size), expected_count in cases.items():
            for strategy in ['binomial-formula', 'brute-force', 'recursion']:
                assert(_compute_number_of_covers(
                    set_sizes = set_sizes,
                    ambient_size = ambient_size,
                    strategy = strategy,
                ) == expected_count)

    @staticmethod
    def test_consistent_with_previous_runs():
        cases = {
            ((3, 1), 4) : 4,
            ((3, 2), 4) : 12,
        }
        for (set_sizes, ambient_size), expected_count in cases.items():
            for strategy in ['binomial-formula', 'brute-force', 'recursion']:
                assert(_compute_number_of_covers(
                    set_sizes = set_sizes,
                    ambient_size = ambient_size,
                    strategy = strategy,
                ) == expected_count)

    @staticmethod
    def test_binomial_formula_matches_brute_force():
        for set_sizes, ambient_size in TestCountingFormulas.sample_cases:
            covers1 = _compute_number_of_covers(
                set_sizes = set_sizes,
                ambient_size = ambient_size,
                strategy = 'binomial-formula'
            )
            covers2 = _compute_number_of_covers(
                set_sizes = set_sizes,
                ambient_size = ambient_size,
                strategy = 'brute-force',
            )
            assert(covers1 == covers2)

    @staticmethod
    def test_binomial_formula_matches_recursion():
        for set_sizes, ambient_size in TestCountingFormulas.sample_cases:
            covers1 = _compute_number_of_covers(
                set_sizes = set_sizes,
                ambient_size = ambient_size,
                strategy = 'binomial-formula'
            )
            covers2 = _compute_number_of_covers(
                set_sizes = set_sizes,
                ambient_size = ambient_size,
                strategy = 'recursion',
            )
            assert(covers1 == covers2)

    @staticmethod
    def test_closed_formula_for_cdf():
        p_values = []
        for set_sizes, ambient_size in TestCountingFormulas.sample_cases:
            for I in range(1, min(set_sizes)+1):
                p1 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'closed-form',
                )
                p2 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'sum-distribution',
                )
                if p1 != 0:
                    denominator = p1
                else:
                    denominator = 1
                assert(abs(p1 - p2) / denominator < 0.00000000001)
                p_values.append((p1,p2))
        return p_values

    @staticmethod
    def test_direct_closed_formula_for_cdf():
        p_values = []
        for set_sizes, ambient_size in TestCountingFormulas.sample_cases:
            for I in range(1, min(set_sizes)+1):
                p1 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'closed-form-covers',
                )
                p2 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'sum-distribution',
                )
                if p1 != 0:
                    denominator = p1
                else:
                    denominator = 1
                assert(abs(p1 - p2) / denominator < 0.00000000001)
                p_values.append((p1,p2))
        return p_values

    @staticmethod
    def test_complementarity_optimization():
        p_values = []
        for set_sizes, ambient_size in TestCountingFormulas.sample_cases:
            for I in range(1, min(set_sizes)+1):
                p1 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'sum-distribution',
                )
                p2 = coincidencetest(
                    incidence_statistic = I,
                    frequencies = set_sizes,
                    number_samples = ambient_size,
                    strategy = 'sum-distribution',
                    use_complementary_values = True
                )
                if p1 != 0:
                    denominator = p1
                else:
                    denominator = 1
                assert(abs(p1 - p2) / denominator < 0.00000000001)
                p_values.append((p1,p2))
        return p_values

    @staticmethod
    def test_matrix_wrapper():
        A = np.array([
            [0, 0, 0],
            [1, 1, 1],
            [0, 0, 0],
            [0, 1, 0],
            [0, 0, 0],
            [1, 0, 0],
            [0, 0, 0],
            [0, 0, 0],
            [0, 0, 1],
            [0, 0, 0],
        ])
        p1 = coincidencetest_matrix(A)

        B = np.array([
            [0, 0, 0],
            [1, 1, 1],
            [0, 0, 0],
            [0, 1, 0],
            [0, 0, 0],
            [1, 0, 0],
            [0, 0, 0],
            [0, 0, 0],
            [0, 0, 1],
            [1, 0, 0],
        ])
        p2 = coincidencetest_matrix(B)

        case = (1, (2, 2, 2), 10)
        incidence_statistic, frequencies, number_samples = case
        p3 = coincidencetest(incidence_statistic, frequencies, number_samples)

        assert(p1 == p3)
        assert(p1 != p2)

class TestStirlingNumberCalc:
    @staticmethod
    def test_small():
        cases = {
            (3, 3) : 1,
            (4, 4) : 1,
            (5, 5) : 1,
            (3, 2) : 3,
        }
        for (ambient_size, number_parts), value in cases.items():
            assert(stirling_second_kind(
                ambient_size=ambient_size,
                number_parts=number_parts,
                normalized=True,
            ) == value)

    @staticmethod
    def test_larger():
        cases = {
            (8, 4) : 1701,
            (10, 3) : 9330,
            (10, 7) : 5880,
        }
        for (ambient_size, number_parts), value in cases.items():
            assert(stirling_second_kind(
                ambient_size=ambient_size,
                number_parts=number_parts,
                normalized=True,
            ) == value)

class TestExactProbabilityCalc:
    sample_cases = [
        ((3, 3, 3), 5),
        ((3, 4, 5), 7),
        ((5, 7, 10), 20),
        ((5, 5, 5), 80),
        ((10, 10, 10), 100),
    ]

    @staticmethod
    def do_test(set_sizes, ambient_size):
        intersection_cases = [
            (i, set_sizes, ambient_size) for i in range(min(set_sizes) + 1)
        ]
        outputs = {
            intersection_size :
            calculate_probability_of_multicoincidence(
                intersection_size = intersection_size,
                set_sizes = set_sizes,
                ambient_size = ambient_size,
            )
            for intersection_size, set_sizes, ambient_size in intersection_cases
        }
        total = sum(outputs.values())
        cdf = [
            total - sum([outputs[j] for j in range(0, i)]) for i in range(len(outputs))
        ]
        return outputs, cdf

    @staticmethod
    def test_exact_probability_of_intersection():
        for set_sizes, ambient_size in TestExactProbabilityCalc.sample_cases:
            TestExactProbabilityCalc.do_test(set_sizes, ambient_size)

class TestStatisticalSignificanceTest:
    sample_cases = {
        ((1, (2, 2, 2), 10)) : 0.07950617283950617,
    }

    @staticmethod
    def test_p_values_prior_calculation():
        cases = TestStatisticalSignificanceTest.sample_cases
        for (incidence_statistic, frequencies, number_samples), p_expected in cases.items():
            p = coincidencetest(incidence_statistic, frequencies, number_samples)
            assert(p == p_expected)

max_stars_repo_name: nadeemlab/coincidencetest

max_issues_repo_name: nadeemlab/coincidencetest

max_forks_repo_name: nadeemlab/coincidencetest


text:
import random
import numpy as np
from tkinter import *
from grid import CellGrid
from pedestrian import Pedestrian

def set_person(grid, row, col):
    """ Set a pedestrian in a specific cell in a specific grid """
    grid.draw_person()
    cell = grid.grid[row][col]
    cell.switch()
    cell.draw(grid.FILLED_COLOR_BG, grid.FILLED_COLOR_BORDER)

def set_target(grid, row, col):
    """ Set a target in a specific cell in a specific grid"""
    grid.draw_target()
    cell = grid.grid[row][col]
    cell.switch()
    cell.draw(grid.FILLED_COLOR_BG, grid.FILLED_COLOR_BORDER)

def set_obstacle(grid, row, col):
    """ Set an obstacle in a specific cell in a specific grid"""
    grid.draw_obstacle()
    cell = grid.grid[row][col]
    cell.switch()
    cell.draw(grid.FILLED_COLOR_BG, grid.FILLED_COLOR_BORDER)

def setup_task_1(app):
    """ Create environment for task 1 """
    grid = CellGrid(app, 50, 10)
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard

def setup_task_2(app):
    """ Create environment for task 2 """
    grid = CellGrid(app, 50, 10)
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard
    # draw person at (5,25)
    set_person(grid, 4, 24)
    # draw target at (25,25)
    set_target(grid, 24, 24)

def setup_task_3(app):
    """ Create environment for task 3 """
    grid = CellGrid(app, 50, 10)
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard
    # draw person at (10,10)
    set_person(grid, 9, 9)
    # draw person at (4,25)
    set_person(grid, 3, 24)
    # draw person at (46,25)
    set_person(grid, 45, 24)
    # draw person at (25,46)
    set_person(grid, 24, 45)
    # draw person at (20,6)
    set_person(grid, 19, 5)
    # draw target at (25,25)
    set_target(grid, 24, 24)

def setup_task_4(app):
    """ Create environment for task 4 """
    grid = CellGrid(app, 10, 50)
    grid.pack()
    set_target(grid, 0, 4)
    set_person(grid, 6, 4)
    for i in range(2, 7):
        set_obstacle(grid, 2, i)
    for i in range(3, 5):
        set_obstacle(grid, i, 2)
        set_obstacle(grid, i, 6)

def rimea_test_1(grid_size=40, screen_width=500):
    """ Create environment for RiMEA scenario 1 in task 5 """
    app = Tk()
    grid = CellGrid(app, grid_size, int(screen_width / grid_size))
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard

    # draw corridor (two lines of obstacles)
    for i in range(grid_size):
        set_obstacle(grid, (grid_size // 2) - 1, i)
        set_obstacle(grid, (grid_size // 2) + 2, i)

    # draw person at the beginning of corridor
    set_person(grid, (grid_size // 2), 0)

    # draw target at the end of the corridor
    set_target(grid, (grid_size // 2), grid_size - 1)
    set_target(grid, (grid_size // 2) + 1, grid_size - 1)

    app.mainloop()

def rimea_test_4(screen_width=500):
    """ Create environment for RiMEA scenario 4 in task 5 """
    grid_size = 80
    app = Tk()
    grid = CellGrid(app, grid_size, int(screen_width / grid_size), is_rimea_4=True)
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard

    # draw corridor (two lines of obstacles)
    corridor_width = 5
    for i in range(grid_size):
        set_obstacle(grid, (grid_size // 2) - corridor_width, i)
        set_obstacle(grid, (grid_size // 2) + corridor_width, i)

    # draw pedestrians at the beginning of corridor (specifying the spawning area)
    min_col = 0
    max_col = grid_size - 4
    min_row = (grid_size // 2) - corridor_width + 1
    max_row = (grid_size // 2) + corridor_width - 1
    possible_cells = []
    for i in range(min_row, max_row + 1):
        for j in range(min_col, max_col + 1):
            possible_cells.append((i, j))
    num_pedestrians = len(possible_cells) // 2
    print(f"PRODUCING {num_pedestrians} PEDESTRIANS IN AN AREA OF {len(possible_cells)} CELLS")
    while num_pedestrians > 0:
        candidate_cell = random.choice(possible_cells)
        possible_cells.remove(candidate_cell)
        set_person(grid, candidate_cell[0], candidate_cell[1])
        num_pedestrians -= 1
    set_person(grid, 39, 13)

    # draw target at the end of the corridor
    for i in range((grid_size // 2) - corridor_width + 1, (grid_size // 2) + corridor_width):
        set_target(grid, i, grid_size - 1)

    app.mainloop()

def rimea_test_6(screen_width=500):
    """ Create environment for RiMEA scenario 6 in task 5 """
    grid_size = 13
    num_pedestrians = 20
    app = Tk()
    grid = CellGrid(app, grid_size, int(screen_width / grid_size))
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard

    # draw corner (two lines of obstacles)
    # these lines looks rather complicated, but we could have simply hardcoded the coordinates of the obstacles,
    # this way it was a little easier to change the disposition and shape of the corner during testing.
    # Anyway this simply places 2 L-shaped lines of obstacles to create a corridor with a 90° turn as the one present
    # in RiMEA scenario 6
    for i in range(grid_size):
        if i in range(0, 8):
            set_obstacle(grid, i, grid_size - 1)
            set_obstacle(grid, i, grid_size - 4)
        elif i == 8:
            for j in range(grid_size):
                if j not in (grid_size - 2, grid_size - 3):
                    set_obstacle(grid, i, j)
        elif i in (9, 10, 11):
            set_obstacle(grid, i, grid_size - 1)
        else:
            for j in range(grid_size):
                set_obstacle(grid, i, j)

    # draw targets
    set_target(grid, 0, grid_size - 2)
    set_target(grid, 0, grid_size - 3)

    # pick pedestrians at random
    min_col, max_col, min_row, max_row = 0, 8, grid_size - 4, grid_size - 2  # boundaries for pedestrian spawning area
    possible_cells = []
    for i in range(min_row, max_row + 1):
        for j in range(min_col, max_col + 1):
            possible_cells.append((i, j))
    while num_pedestrians > 0:
        candidate_cell = random.choice(possible_cells)
        possible_cells.remove(candidate_cell)
        set_person(grid, candidate_cell[0], candidate_cell[1])
        num_pedestrians -= 1

    app.mainloop()

def rimea_test_7(num_pedestrians=50, screen_width=500):
    """ Create environment for RiMEA scenario 7 in task 5 """
    speed_list = sample_age_speed(num_pedestrians)
    app = Tk()
    grid = CellGrid(app, num_pedestrians, int(screen_width / num_pedestrians))
    grid.pack()
    grid.focus_set()  # to receive inputs form keyboard

    # draw person on first column
    for cell in range(num_pedestrians):
        set_person(grid, cell, 0)

    # draw target on last column
    for cell in range(num_pedestrians):
        set_target(grid, cell, num_pedestrians - 1)

    # create Pedestrian list with custom speeds
    grid.pedestrian_list = [
        Pedestrian(grid, cell, speed_list[i], num_pedestrians >= 100) for i, cell in enumerate(grid.pedestrian_cell_list)
    ]
    app.mainloop()
    print("Measured Average Speed: ", sum(grid.pedestrian_speeds) / len(grid.pedestrian_speeds))
    print("Expected Average Speed: ", sum(speed_list) / len(speed_list))

def sample_age_speed(num_samples):
    """
    given a number of samples to be sampled, returns an array of speeds for the Pedestrians to create
    after sampling their age (going from 3 to 80 years)
    :param num_samples: number of samples of speed to take
    :return: the list containing the speeds (rounded to the 2nd decimal)
    """
    lowest_age, highest_age = 3, 80
    first_range_speed = [0.6, 1.2]  # 3-10 years
    second_range_speed = [1.2, 1.6]  # 11-20 years
    third_range_speed = [1.6, 1.4]  # 21-50 years
    fourth_range_speed = [1.4, 1.1]  # 51-70 years
    fifth_range_speed = [1.1, 0.7]  # 71-80 years
    sampled_ages = np.random.randint(lowest_age, highest_age, num_samples)

    speed_list = []
    for age in sampled_ages:
        if 3 <= age <= 10:
            speed_list.append(np.random.uniform(first_range_speed[0], first_range_speed[1]))
        elif 11 <= age <= 20:
            speed_list.append(np.random.uniform(second_range_speed[0], second_range_speed[1]))
        elif 21 <= age <= 50:
            speed_list.append(np.random.uniform(third_range_speed[0], third_range_speed[1]))
        elif 51 <= age <= 70:
            speed_list.append(np.random.uniform(fourth_range_speed[0], fourth_range_speed[1]))
        else:
            speed_list.append(np.random.uniform(fifth_range_speed[0], fifth_range_speed[1]))
    return [round(speed, 2) for speed in speed_list]

max_stars_repo_name: AlexPasqua/MLCMS-exercises

max_issues_repo_name: AlexPasqua/MLCMS-exercises

max_forks_repo_name: AlexPasqua/MLCMS-exercises


text:
# Copyright (C) 2017-2019 Tormod Landet
# SPDX-License-Identifier: Apache-2.0

from __future__ import division, print_function
import sys
import numpy
from matplotlib import pyplot
from matplotlib.widgets import Slider

def read_iso_surfaces(filename):
    timesteps = []
    data = []

    with open(filename, 'rt') as f:
        description = f.readline()[1:].strip()
        value = float(f.readline().split()[-1])
        dim = int(f.readline().split()[-1])

        line = f.readline()
        while line:
            wds = line.split()
            try:
                time = float(wds[1])
                nsurf = int(wds[3])
            except Exception:
                break

            if nsurf == 0:
                timesteps.append(time)
                data.append([])
                line = f.readline()
                continue

            datalines = [f.readline() for _ in range(nsurf * 3)]
            if not datalines[-1]:
                break
            timesteps.append(time)
            data.append([])
            for i in range(nsurf):
                xvals = [float(v) for v in datalines[i * 3 + 0].split()]
                yvals = [float(v) for v in datalines[i * 3 + 1].split()]
                zvals = [float(v) for v in datalines[i * 3 + 2].split()]
                data[-1].append((xvals, yvals, zvals))

            line = f.readline()
    return description, value, dim, timesteps, data

def plotit(ax, contours, label):
    colour = get_colour(label)
    xvals = []
    yvals = []
    for contour in contours:
        if xvals:
            xvals.append(None)
            yvals.append(None)
        xvals.extend(contour[0])
        yvals.extend(contour[1])
    ax.plot(xvals, yvals, c=colour, label=label)

COLOURS = {}

def get_colour(label):
    available = 'bgrcmy'
    if label not in COLOURS:
        COLOURS[label] = available[len(COLOURS)]
    return COLOURS[label]

def main(filenames, equal_axes):
    all_data = []
    tmin = tmax = 0
    xmin, ymin, xmax, ymax = 1e100, 1e100, -1e100, -1e100
    for i, filename in enumerate(filenames):
        name = str(i + 1)
        if ':' in filename:
            filename, name = filename.split(':')
        print('Reading %s from file %s' % (name, filename))
        _description, _value, _dim, timesteps, data = read_iso_surfaces(filename)
        all_data.append((name, numpy.array(timesteps), data))

        tmax = max(tmax, timesteps[-1])
        for contours in data:
            for contour in contours:
                xmin = min(xmin, numpy.min(contour[0]))
                ymin = min(ymin, numpy.min(contour[1]))
                xmax = max(xmax, numpy.max(contour[0]))
                ymax = max(ymax, numpy.max(contour[1]))

    fig, ax = pyplot.subplots()
    pyplot.subplots_adjust(bottom=0.25)
    axcolor = '#a1b8dd'
    slider_ax = pyplot.axes([0.1, 0.1, 0.8, 0.03], facecolor=axcolor)
    slider = Slider(slider_ax, 'Time', tmin, tmax, valinit=tmin)

    xdiff = xmax - xmin
    ydiff = ymax - ymin
    xmin, xmax = xmin - 0.1 * xdiff, xmax + 0.1 * xdiff
    ymin, ymax = ymin - 0.1 * ydiff, ymax + 0.1 * ydiff
    ax.set_xlim(xmin, xmax)
    ax.set_ylim(ymin, ymax)

    def update(val):
        xmin, xmax = ax.get_xlim()
        ymin, ymax = ax.get_ylim()
        ax.clear()

        t = slider.val
        for name, timesteps, data in all_data:
            i = numpy.argmin(abs(timesteps - t))
            dt = timesteps[1] - timesteps[0]
            if abs(timesteps[i] - t) > 1.5 * dt:
                continue
            plotit(ax, data[i], name)

        ax.set_xlim(xmin, xmax)
        ax.set_ylim(ymin, ymax)

        if equal_axes:
            ax.set_aspect('equal')

        if len(filenames) > 1:
            ax.legend(loc='lower right')
        fig.canvas.draw_idle()

    slider.on_changed(update)
    slider.set_val(tmin)
    pyplot.show()

if __name__ == '__main__':
    filenames = sys.argv[1:]

    equal_axes = True
    if '--nonequal' in filenames:
        equal_axes = False
        filenames.remove('--nonequal')

    main(filenames, equal_axes)

max_stars_repo_name: TormodLandet/Ocellaris

max_issues_repo_name: TormodLandet/Ocellaris

max_forks_repo_name: TormodLandet/Ocellaris


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

"""
Train a decoder-based model using L2HMC (or HMC) as a posterior sampler
"""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import time, sys, string, os

import tensorflow as tf
import numpy as np

from utils.func_utils import accept, jacobian, autocovariance, get_log_likelihood, get_data,\
    var_from_scope, binarize, normal_kl, binarize_and_shuffle
from utils.distributions import Gaussian, GMM, GaussianFunnel, gen_ring
from utils.layers import Linear, Parallel, Sequential, Zip, ScaleTanh
from utils.dynamics import Dynamics
from utils.sampler import propose, tf_accept, chain_operator
from utils.losses import get_loss, loss_mixed

FLAGS = tf.app.flags.FLAGS
tf.app.flags.DEFINE_string('hparams', '', 'Comma sep list of name=value')
tf.app.flags.DEFINE_string('exp_id', '', 'exp_id')

DEFAULT_HPARAMS = tf.contrib.training.HParams(
    learning_rate=0.001,
    epoch=100,
    leapfrogs=5,
    MH=5,
    optimizer='adam',
    batch_size=512,
    latent_dim=50,
    update_sampler_every=1,
    eval_samples_every=1,
    random_lf_composition=0,
    stop_gradient=False,
    hmc=False,
    eps=0.1,
    energy_scale=0.,
)

# hardcode the loss
LOSS = 'mixed'

OPTIMIZERS = {
    'adam': tf.train.AdamOptimizer,
    'rmsprop': tf.train.RMSPropOptimizer,
    'nesterov': tf.train.MomentumOptimizer,
    'sgd': tf.train.GradientDescentOptimizer,
}

def main(_):
    hps = DEFAULT_HPARAMS
    print(FLAGS.hparams)
    hps.parse(FLAGS.hparams)

    # hack for logdir
    hps_values = hps.values()
    del(hps_values['epoch'])
    del(hps_values['eval_samples_every'])

    train_folder = string.join(
        [
            str(k)+'='+str(hps_values[k])
            for k in hps_values
        ],
        ',',
    )

    logdir = 'logs/%s/%s' % (FLAGS.exp_id, train_folder)

    print('Saving logs to %s' % logdir)

    float_x_train, float_x_test = get_data()
    N = float_x_train.shape[0]

    with tf.variable_scope('encoder'):
        encoder = Sequential([
            Linear(784, 1024, scope='encoder_1'),
            tf.nn.softplus,
            Linear(1024, 1024, scope='encoder_2'),
            tf.nn.softplus,
            Parallel([
                Linear(1024, hps.latent_dim, scope='encoder_mean'),
                Linear(1024, hps.latent_dim, scope='encoder_std'),
            ])
        ])

    with tf.variable_scope('decoder'):
        decoder = Sequential([
            Linear(hps.latent_dim, 1024, scope='decoder_1'),
            tf.nn.softplus,
            Linear(1024, 1024, scope='decoder_2'),
            tf.nn.softplus,
            Linear(1024, 784, scope='decoder_3', factor=0.01)
        ])

    # Setting up the VAE

    inp = tf.placeholder(tf.float32, shape=(None, 784))
    mu, log_sigma = encoder(inp)
    noise = tf.random_normal(tf.shape(mu))
    latent_q = mu + noise * tf.exp(log_sigma)
    logits = decoder(latent_q)

    # Setting up sampler
    def energy(z, aux=None):
        logits = decoder(z)
        log_posterior = -tf.reduce_sum(tf.nn.sigmoid_cross_entropy_with_logits(labels=aux, logits=logits), axis=1)
        log_prior = -0.5 * tf.reduce_sum(tf.square(z), axis=1)
        return (-log_posterior - log_prior)
    energy_stop_grad = lambda z, aux=None: energy(tf.stop_gradient(z), aux=None)
    sampler_loss = 0.

    with tf.variable_scope('sampler'):
        size1 = 200
        size2 = 200

        encoder_sampler = Sequential([
            Linear(784, 512, scope='encoder_1'),
            tf.nn.softplus,
            Linear(512, 512, scope='encoder_2'),
            tf.nn.softplus,
            Linear(512, size1, scope='encoder_3'),
        ])

        def net_factory(x_dim, scope, factor):
            with tf.variable_scope(scope):
                net = Sequential([
                    Zip([
                        Linear(hps.latent_dim, size1, scope='embed_1', factor=0.33),
                        Linear(hps.latent_dim, size1, scope='embed_2', factor=factor * 0.33),
                        Linear(2, size1, scope='embed_3', factor=0.33),
                        encoder_sampler,
                    ]),
                    sum,
                    tf.nn.relu,
                    Linear(size1, size2, scope='linear_1'),
                    tf.nn.relu,
                    Parallel([
                        Sequential([
                            Linear(size2, hps.latent_dim, scope='linear_s', factor=0.01),
                            ScaleTanh(hps.latent_dim, scope='scale_s')
                        ]),
                        Linear(size2, hps.latent_dim, scope='linear_t', factor=0.01),
                        Sequential([
                            Linear(size2, hps.latent_dim, scope='linear_f', factor=0.01),
                            ScaleTanh(hps.latent_dim, scope='scale_f'),
                        ])
                    ])
                ])
            return net

        dynamics = Dynamics(
            hps.latent_dim,
            energy,
            T=hps.leapfrogs,
            eps=hps.eps,
            hmc=hps.hmc,
            net_factory=net_factory,
            eps_trainable=True,
            use_temperature=False,
        )

    init_x = tf.stop_gradient(latent_q)
    init_v = tf.random_normal(tf.shape(init_x))

    for t in range(hps.MH):
        inverse_term = 0.
        other_term = 0.
        energy_loss = 0.

        if hps.stop_gradient:
            init_x = tf.stop_gradient(init_x)

        if hps.random_lf_composition > 0:
            nb_steps = tf.random_uniform((), minval=1, maxval=hps.random_lf_composition, dtype=tf.int32)

            final_x, _, px, MH = chain_operator(init_x, dynamics, nb_steps, aux=inp, do_mh_step=True)

            energy_loss = 0.

        else:
            inverse_term = 0.
            other_term = 0.

            final_x, _, px, MH = propose(init_x, dynamics, aux=inp, do_mh_step=True)

            #sampler_loss += 1.0 / hps.MH * loss_mixed(latent, Lx, px, scale=tf.stop_gradient(tf.exp(log_sigma)))

        # distance
        v = tf.square(final_x - init_x) / (tf.stop_gradient(tf.exp(2 * log_sigma)) + 1e-4)
        v = tf.reduce_sum(v, 1) * px + 1e-4

        # energy

        energy_diff = tf.square(energy(final_x, aux=inp) - energy(init_x, aux=inp)) * px + 1e-4

        inverse_term += 1.0 / hps.MH * tf.reduce_mean(1.0 / v)
        other_term -= 1.0 / hps.MH * tf.reduce_mean(v)
        energy_loss += 1.0 / hps.MH * (tf.reduce_mean(1.0 / energy_diff) - tf.reduce_mean(energy_diff))

        init_x = MH[0]

    latent_T = init_x

    sampler_loss = inverse_term + other_term + hps.energy_scale * energy_loss

    logits_T = decoder(tf.stop_gradient(latent_T))
    partition = tf.constant(np.sqrt((2 * np.pi) ** hps.latent_dim), dtype=tf.float32)
    prior_probs = tf.log(partition) + \
        0.5 * tf.reduce_sum(tf.square(tf.stop_gradient(latent_T)), axis=1)
    posterior_probs = tf.reduce_sum(tf.nn.sigmoid_cross_entropy_with_logits(labels=inp, logits=logits_T), axis=1)

    likelihood = tf.reduce_mean(prior_probs+posterior_probs, axis=0)

    kl = normal_kl(mu, tf.exp(log_sigma), 0., 1.)
    bce = tf.reduce_sum(tf.nn.sigmoid_cross_entropy_with_logits(labels=inp, logits=logits), axis=1)
    elbo = tf.check_numerics(tf.reduce_mean(kl+bce), 'elbo NaN')

    batch_per_epoch = N / hps.batch_size

    # Setting up train ops

    global_step = tf.Variable(0., trainable=False)
    # learning_rate = tf.train.exponential_decay(
    #     hps.learning_rate,
    #     global_step,
    #     750,
    #     0.96,
    #     staircase=True
    # )

    learning_rate = tf.train.piecewise_constant(global_step, [batch_per_epoch * 500.], [1e-3, 1e-4])

    opt_sampler = tf.train.AdamOptimizer(learning_rate)
    opt = tf.train.AdamOptimizer(learning_rate)

    elbo_train_op = opt.minimize(elbo, var_list=var_from_scope('encoder'))
    if not hps.hmc:
        gradients, variables = zip(*opt_sampler.compute_gradients(sampler_loss, var_list=var_from_scope('sampler')))
        gradients, global_norm = tf.clip_by_global_norm(gradients, 5.0)
        sampler_train_op = opt_sampler.apply_gradients(zip(gradients, variables))
        # sampler_train_op = opt_sampler.minimize(sampler_loss, var_list=var_from_scope('sampler'), global_step=global_step)
    else:
        sampler_train_op = tf.no_op()
    decoder_train_op = opt.minimize(likelihood, var_list=var_from_scope('decoder'), global_step=global_step)

    # if not hps.hmc:
    #    tf.summary.scalar('sampler_grad_norm', global_norm)

    tf.summary.scalar('inverse_term', inverse_term)
    tf.summary.scalar('other_term', other_term)
    tf.summary.scalar('energy_loss', energy_loss)
    tf.summary.scalar('sampler_loss', sampler_loss)
    tf.summary.scalar('log_prob', likelihood)
    tf.summary.scalar('elbo', elbo)
    tf.summary.scalar('p_accept', tf.reduce_mean(px))

    loss_summaries = tf.summary.merge_all()

    # For sample generation
    z_eval = tf.placeholder(tf.float32, shape=(None, hps.latent_dim))
    x_eval = tf.nn.sigmoid(decoder(z_eval))

    samples_summary = tf.summary.image(
        'samples',
        tf.reshape(x_eval, (-1, 28, 28, 1)),
        64,
    )

    saver = tf.train.Saver()
    writer = tf.summary.FileWriter(logdir)

    sess = tf.Session()
    sess.run(tf.global_variables_initializer())

    counter = 0

    # For graph restore
    tf.add_to_collection('inp', inp)
    tf.add_to_collection('latent_q', latent_q)
    tf.add_to_collection('latent_T', latent_T)
    tf.add_to_collection('logits_T', logits_T)
    tf.add_to_collection('z_eval', z_eval)
    tf.add_to_collection('x_eval', x_eval)

    time0 = time.time()
    for e in range(hps.epoch):
        x_train = binarize_and_shuffle(float_x_train)

        for t in range(batch_per_epoch):
            start = t * hps.batch_size
            end = start + hps.batch_size

            batch = x_train[start:end, :]

            fetches = [
                elbo, sampler_loss, likelihood, loss_summaries, \
                global_step, elbo_train_op, decoder_train_op, learning_rate
            ]

            if t % hps.update_sampler_every == 0:
                fetches += [sampler_train_op]

            fetched = sess.run(fetches, {inp: batch})

            if t % 50 == 0:
                print 'Step:%d::%d/%d::ELBO: %.3e::Loss sampler: %.3e:: Log prob: %.3e:: Lr: %g:: Time: %.2e' \
                    % (fetched[4], t, batch_per_epoch, fetched[0], fetched[1], fetched[2], fetched[-2], time.time()-time0)
                time0 = time.time()

            writer.add_summary(fetched[3], global_step=counter)
            counter += 1
        if e % hps.eval_samples_every == 0:
            saver.save(sess, '%s/model.ckpt' % logdir)
            samples_summary_ = sess.run(samples_summary, {z_eval: np.random.randn(64, hps.latent_dim)})
            writer.add_summary(samples_summary_, global_step=(e / hps.eval_samples_every))

    for AS in [64, 256, 1024, 4096, 8192]:
        cmd = 'python eval_vae.py --path "%s/" --split %s --anneal_steps %d'
        print 'Train fold evaluation. AS steps: %d' % AS
        os.system(cmd % (logdir, 'train', AS))

        print 'Test fold evaluation. AS steps: %d' % AS
        os.system(cmd % (logdir, 'test', AS))

    print 'Sampler eval'
    os.system('python eval_sampler.py --path "%s"' % logdir)

if __name__ == '__main__':
    tf.app.run(main)


text:
# ------------------------------------------------------------------------------
# This code is base on
# CornerNet (https://github.com/princeton-vl/CornerNet)
# Copyright (c) 2018, University of Michigan
# Licensed under the BSD 3-Clause License
# ------------------------------------------------------------------------------

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import numpy as np
import torch
import torch.nn as nn
from models.networks.large_hourglass import convolution, residual
from models.networks.large_hourglass import make_layer, make_layer_revr, make_pool_layer, make_unpool_layer
from models.networks.large_hourglass import make_cnv_layer, make_kp_layer, make_merge_layer, make_inter_layer
from models.networks.large_hourglass import kp_module
from models.networks.large_hourglass import exkp

class oneshot_exkp(nn.Module):
    def __init__(
            self, n, nstack, dims, modules, heads, pre=None, cnv_dim=256,
            make_tl_layer=None, make_br_layer=None,
            make_cnv_layer=make_cnv_layer, make_heat_layer=make_kp_layer,
            make_tag_layer=make_kp_layer, make_regr_layer=make_kp_layer,
            make_up_layer=make_layer, make_low_layer=make_layer,
            make_hg_layer=make_layer, make_hg_layer_revr=make_layer_revr,
            make_pool_layer=make_pool_layer, make_unpool_layer=make_unpool_layer,
            make_merge_layer=make_merge_layer, make_inter_layer=make_inter_layer,
            kp_layer=residual
    ):
        super(oneshot_exkp, self).__init__()

        self.nstack = nstack
        self.heads = heads

        curr_dim = dims[0]

        self.pre = nn.Sequential(
            convolution(7, 3, 128, stride=2),
            residual(3, 128, 256, stride=2)
        ) if pre is None else pre

        self.kps = nn.ModuleList([
            kp_module(
                n, dims, modules, layer=kp_layer,
                make_up_layer=make_up_layer,
                make_low_layer=make_low_layer,
                make_hg_layer=make_hg_layer,
                make_hg_layer_revr=make_hg_layer_revr,
                make_pool_layer=make_pool_layer,
                make_unpool_layer=make_unpool_layer,
                make_merge_layer=make_merge_layer
            ) for _ in range(nstack)
        ])
        self.cnvs = nn.ModuleList([
            make_cnv_layer(curr_dim, cnv_dim) for _ in range(nstack)
        ])

        self.inters = nn.ModuleList([
            make_inter_layer(curr_dim) for _ in range(nstack - 1)
        ])

        self.inters_ = nn.ModuleList([
            nn.Sequential(
                nn.Conv2d(curr_dim, curr_dim, (1, 1), bias=False),
                nn.BatchNorm2d(curr_dim)
            ) for _ in range(nstack - 1)
        ])
        self.cnvs_ = nn.ModuleList([
            nn.Sequential(
                nn.Conv2d(cnv_dim, curr_dim, (1, 1), bias=False),
                nn.BatchNorm2d(curr_dim)
            ) for _ in range(nstack - 1)
        ])

        self.match = match_block

        ## keypoint heatmaps
        for head in heads.keys():
            if 'hm' in head:
                module = nn.ModuleList([
                    make_heat_layer(
                        cnv_dim, curr_dim, heads[head]) for _ in range(nstack)
                ])
                self.__setattr__(head, module)
                for heat in self.__getattr__(head):
                    heat[-1].bias.data.fill_(-2.19)
            else:
                module = nn.ModuleList([
                    make_regr_layer(
                        cnv_dim, curr_dim, heads[head]) for _ in range(nstack)
                ])
                self.__setattr__(head, module)

        self.relu = nn.ReLU(inplace=True)

    def forward(self, image, ref_image):
        # print('image shape', image.shape)
        inter = self.pre(image)
        ref_inter = self.pre(ref_image)
        outs = []

        for ind in range(self.nstack):
            kp_, cnv_ = self.kps[ind], self.cnvs[ind]
            kp = kp_(inter)
            cnv = cnv_(kp)
            ref_kp = kp_(ref_inter)
            ref_cnv = cnv_(ref_kp)

            match = self.match(cnv, ref_cnv)

            out = {}
            for head in self.heads:
                layer = self.__getattr__(head)[ind]
                y = layer(match)
                out[head] = y

            outs.append(out)
            if ind < self.nstack - 1:
                inter = self.inters_[ind](inter) + self.cnvs_[ind](cnv)
                inter = self.relu(inter)
                inter = self.inters[ind](inter)
                ref_inter = self.inters_[ind](ref_inter) + self.cnvs_[ind](ref_cnv)
                ref_inter = self.relu(ref_inter)
                ref_inter = self.inters[ind](ref_inter)
        return outs

def make_hg_layer(kernel, dim0, dim1, mod, layer=convolution, **kwargs):
    layers = [layer(kernel, dim0, dim1, stride=2)]
    layers += [layer(kernel, dim1, dim1) for _ in range(mod - 1)]
    return nn.Sequential(*layers)

class HourglassNet(exkp):
    def __init__(self, heads, num_stacks=2):
        n = 5
        dims = [256, 256, 384, 384, 384, 512]
        modules = [2, 2, 2, 2, 2, 4]

        super(HourglassNet, self).__init__(
            n, num_stacks, dims, modules, heads,
            make_tl_layer=None,
            make_br_layer=None,
            make_pool_layer=make_pool_layer,
            make_hg_layer=make_hg_layer,
            kp_layer=residual, cnv_dim=256
        )

def get_oneshot_hourglass_net(num_layers, heads, head_conv):
    model = HourglassNet(heads, 2)
    return model

class match_block(nn.Module):
    def __init__(self, inplanes):
        super(match_block, self).__init__()

        self.sub_sample = False

        self.in_channels = inplanes
        self.inter_channels = None

        if self.inter_channels is None:
            self.inter_channels = self.in_channels // 2
            if self.inter_channels == 0:
                self.inter_channels = 1

        conv_nd = nn.Conv2d
        max_pool_layer = nn.MaxPool2d(kernel_size=(2, 2))
        bn = nn.BatchNorm2d

        self.g = conv_nd(in_channels=self.in_channels, out_channels=self.inter_channels,
                         kernel_size=1, stride=1, padding=0)

        self.W = nn.Sequential(
            conv_nd(in_channels=self.inter_channels, out_channels=self.in_channels,
                    kernel_size=1, stride=1, padding=0),
            bn(self.in_channels)
        )
        nn.init.constant_(self.W[1].weight, 0)
        nn.init.constant_(self.W[1].bias, 0)

        self.Q = nn.Sequential(
            conv_nd(in_channels=self.inter_channels, out_channels=self.in_channels,
                    kernel_size=1, stride=1, padding=0),
            bn(self.in_channels)
        )
        nn.init.constant_(self.Q[1].weight, 0)
        nn.init.constant_(self.Q[1].bias, 0)

        self.theta = conv_nd(in_channels=self.in_channels, out_channels=self.inter_channels,
                             kernel_size=1, stride=1, padding=0)
        self.phi = conv_nd(in_channels=self.in_channels, out_channels=self.inter_channels,
                           kernel_size=1, stride=1, padding=0)

        self.concat_project = nn.Sequential(
            nn.Conv2d(self.inter_channels * 2, 1, 1, 1, 0, bias=False),
            nn.ReLU()
        )

        # self.ChannelGate = ChannelGate(self.in_channels)
        self.globalAvgPool = nn.AdaptiveAvgPool2d(1)

    def forward(self, detect, aim):

        batch_size, channels, height_a, width_a = aim.shape
        batch_size, channels, height_d, width_d = detect.shape

        #####################################find aim image similar object ####################################################

        d_x = self.g(detect).view(batch_size, self.inter_channels, -1)
        d_x = d_x.permute(0, 2, 1).contiguous()

        a_x = self.g(aim).view(batch_size, self.inter_channels, -1)
        a_x = a_x.permute(0, 2, 1).contiguous()

        theta_x = self.theta(aim).view(batch_size, self.inter_channels, -1)
        theta_x = theta_x.permute(0, 2, 1)

        phi_x = self.phi(detect).view(batch_size, self.inter_channels, -1)

        f = torch.matmul(theta_x, phi_x)  # (aim,detect)

        # N = f.size(-1)
        # f_div_C = f / N

        f = f.permute(0, 2, 1).contiguous()
        N = f.size(-1)
        fi_div_C = f / N  # detect,aim

        # non_aim = torch.matmul(f_div_C, d_x)
        # non_aim = non_aim.permute(0, 2, 1).contiguous()
        # non_aim = non_aim.view(batch_size, self.inter_channels, height_a, width_a)
        # non_aim = self.W(non_aim)
        # non_aim = non_aim + aim

        non_det = torch.matmul(fi_div_C, a_x)
        non_det = non_det.permute(0, 2, 1).contiguous()
        non_det = non_det.view(batch_size, self.inter_channels, height_d, width_d)
        non_det = self.Q(non_det)
        non_det = non_det + detect

        ##################################### Response in chaneel weight ####################################################

        # c_weight = self.ChannelGate(non_aim)
        # act_aim = non_aim * c_weight
        # act_det = non_det * c_weight

        return non_det

max_stars_repo_name: IronSublimate/CenterNet-IS-old

max_issues_repo_name: IronSublimate/CenterNet-IS-old

max_forks_repo_name: IronSublimate/CenterNet-IS-old