text:
''''
Author: Gurkirt Singh
Date: 20:08:2016

Main purpose of this script to help produce framelevel annotations for DALY dataset
This script helps to extract frames from video and store useful info about videos.

'''
import os as os
import numpy as np
import json,shutil,pickle

actions = ['ApplyingMakeUpOnLips', 'BrushingTeeth', 'CleaningFloor', 'CleaningWindows', 'Drinking', 'FoldingTextile', 'Ironing', 'Phoning', 'PlayingHarmonica', 'TakingPhotosOrVideos']

basedir = '../'; # you chaneg to the place whre you videos and images should be.

def extractframes(vids): # take all .mp4 videos and extract frames using ffmpeg
    for vid in vids:
        vidfile = basedir+'videos/'+vid+'.mp4'
        imgdir = basedir+'images/'+vid+'/'
        if not os.path.isdir(imgdir):
            os.mkdir(imgdir)
        cmd = 'ffmpeg -i {} -qscale:v 5 -vf decimate=-0:99999 {}%05d.jpg'.format(vidfile,imgdir); #-vsync 0
        print cmd
        os.system(cmd)
            # PNG format is very storage heavy so I choose jpg.
            # images will be generated in JPG format with quality scale = 5; you can adjust according to you liking
            # In appearence it doen't look that deblurred as opposed to default settings by ffmpeg
            # @v 5 images will take alomst 145GB

def saveVidInfo(vids):
    vidinfo = dict();
    count = 0;
    for vid in vids:
        vidfile = basedir+'videos/'+vid+'.mp4'
        imgdir = basedir+'images/'+vid+'/'
        imglist = os.listdir(imgdir);
        imglist = [i for i in imglist if i.endswith('.jpg')];
        cmd  = 'ffprobe -v quiet -print_format json  -show_streams -count_frames {} >>temp{}.json'.format(vidfile,'0')
        # -count_frames option takes the most of the time to run ffprobe
        print cmd
        os.system(cmd)
        with open('temp{}.json'.format(count)) as f:
            ffdata = json.load(f)

        if ffdata["streams"][0]['codec_type'] == 'video':
            ffdata = ffdata["streams"][0]
        else:
            ffdata = ffdata["streams"][1]

        # ffdata = ffdata["streams"][0]
        os.remove('temp{}.json'.format('0'))
        vinfo = dict();
        vinfo['duration']=float(ffdata['duration']);
        vinfo['nbframes']=len(imglist) #int(ffdata['nb_frames']);
        vinfo['ffprobeNbFrames']=int(ffdata['nb_frames']); #ffprobeNbFrames and nbframes are not always same (in case of variable fps vidoes)
        vinfo['frame_rate_info'] = ffdata['r_frame_rate']+'/'+ffdata['avg_frame_rate'] # frame rate info provided by ffprobe doen't match with ffmpeg number of frames always
        vinfo['fps'] = float(vinfo['nbframes'])/vinfo['duration']; # compute fps
        vidinfo[vid] = vinfo;
        print vinfo

    #with open('../hfiles/videoInfo.json','w') as f:
    #     json.dump(vidinfo,f)

if __name__ == '__main__':
    #downloaded = os.listdir(basedir+'videos') # get list of file
    #downloaded = [d for d in downloaded if d.endswith('.mp4')]  # keep only .mp4 files.
    #print 'number of videos downloded are ', len(downloaded)
    with open('../hfiles/video_ids.txt') as f:
        vids = f.readlines(); # read video ids
    vids = [d.rstrip('\n') for d in vids]  # remove \n
    with open('../hfiles/daly1.1.0.pkl') as f:
        daly = pickle.load(f);
    with open('../hfiles/videoInfo.json') as f:
        jdaly = json.load(f);
    metadata = daly['metadata'];
    count =0;
    tempvids = [];
    for vid in vids:

        imgdir = basedir+'rgb-images/'+vid+'/'
        #flowdir = basedir+'motion-images/'+vid+'/'
        if not os.path.isdir(imgdir):
             os.mkdir(imgdir)
        #if not os.path.isdir(flowdir):
        #     os.mkdir(flowdir)
        imglist = os.listdir(imgdir)
        imglist = [i for i in imglist if i.endswith('.jpg')];
        num_frames = len(imglist);
        nbframes_ffmpeg = metadata[vid+'.mp4']['nbframes_ffmpeg']
        duration =  metadata[vid+'.mp4']['duration']
        #duration = metadata[vid+'.mp4']['duration']
        jduration = jdaly[vid]['duration']
        #duration = metadata[vid+'.mp4']['nbframes_ffmpeg']
        diffs = abs(metadata[vid+'.mp4']['nbframes_ffmpeg']  - num_frames);
        if abs(duration-jduration)>0.2:
            #shutil.rmtree(imgdir)
            #shutil.rmtree(flowdir)
            tempvids.append(vid)
            print vid, '.mp4 nbframes_ffmpeg ', nbframes_ffmpeg,  ' duration ', metadata[vid+'.mp4']['duration'] , ' fps ', metadata[vid+'.mp4']['fps'] , ' number of extracted frames ',num_frames,  ' duration ',jduration,' fps ', jdaly[vid]['fps']
            count +=1;
    print 'diff in ', count
    # for v in tempvids:
    #     cmd = "wget --user privateuser --password pltbd21aocolnq http://pascal.inrialpes.fr/data2/daly_cache/cache/"+v+".mp4 -O /mnt/jupiter-beta/DALY/videos/"+v+".mp4"
    #     os.system(cmd)
    # extractframes(tempvids)

max_stars_repo_name: gurkirt/daly-prep

max_issues_repo_name: gurkirt/daly-prep

max_forks_repo_name: gurkirt/daly-prep


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

import tensorflow as tf
import numpy as np
import os
import re
import json
import nltk
from itertools import chain
from collections import Counter
from nltk.stem.wordnet import WordNetLemmatizer

from bayou.models.low_level_evidences.utils import CONFIG_ENCODER, CONFIG_INFER, C0, UNK

class Evidence(object):

    def init_config(self, evidence, chars_vocab):
        for attr in CONFIG_ENCODER + (CONFIG_INFER if chars_vocab else []):
            self.__setattr__(attr, evidence[attr])

    def dump_config(self):
        js = {attr: self.__getattribute__(attr) for attr in CONFIG_ENCODER + CONFIG_INFER}
        return js

    @staticmethod
    def read_config(js, chars_vocab):
        evidences = []
        for evidence in js:
            name = evidence['name']
            if name == 'apicalls':
                e = APICalls()
            elif name == 'types':
                e = Types()
            elif name == 'keywords':
                e = Keywords()
            else:
                raise TypeError('Invalid evidence name: {}'.format(name))
            e.init_config(evidence, chars_vocab)
            evidences.append(e)
        return evidences

    def read_data_point(self, program):
        raise NotImplementedError('read_data() has not been implemented')

    def set_chars_vocab(self, data):
        raise NotImplementedError('set_chars_vocab() has not been implemented')

    def wrangle(self, data):
        raise NotImplementedError('wrangle() has not been implemented')

    def placeholder(self, config):
        raise NotImplementedError('placeholder() has not been implemented')

    def exists(self, inputs):
        raise NotImplementedError('exists() has not been implemented')

    def init_sigma(self, config):
        raise NotImplementedError('init_sigma() has not been implemented')

    def encode(self, inputs, config):
        raise NotImplementedError('encode() has not been implemented')

    def evidence_loss(self, psi, encoding, config):
        raise NotImplementedError('evidence_loss() has not been implemented')

class APICalls(Evidence):

    def read_data_point(self, program):
        apicalls = program['apicalls'] if 'apicalls' in program else []
        return list(set(apicalls))

    def set_chars_vocab(self, data):
        counts = Counter([c for apicalls in data for c in apicalls])
        self.chars = sorted(counts.keys(), key=lambda w: counts[w], reverse=True)
        self.vocab = dict(zip(self.chars, range(len(self.chars))))
        self.vocab_size = len(self.vocab)

    def wrangle(self, data):
        wrangled = np.zeros((len(data), 1, self.vocab_size), dtype=np.int32)
        for i, apicalls in enumerate(data):
            for c in apicalls:
                if c in self.vocab:
                    wrangled[i, 0, self.vocab[c]] = 1
        return wrangled

    def placeholder(self, config):
        return tf.placeholder(tf.float32, [config.batch_size, 1, self.vocab_size])

    def exists(self, inputs):
        i = tf.reduce_sum(inputs, axis=2)
        return tf.not_equal(tf.count_nonzero(i, axis=1), 0)

    def init_sigma(self, config):
        with tf.variable_scope('apicalls'):
            self.sigma = tf.get_variable('sigma', [])

    def encode(self, inputs, config):
        with tf.variable_scope('apicalls'):
            latent_encoding = tf.zeros([config.batch_size, config.latent_size])
            inp = tf.slice(inputs, [0, 0, 0], [config.batch_size, 1, self.vocab_size])
            inp = tf.reshape(inp, [-1, self.vocab_size])
            encoding = tf.layers.dense(inp, self.units, activation=tf.nn.tanh)
            for i in range(self.num_layers - 1):
                encoding = tf.layers.dense(encoding, self.units, activation=tf.nn.tanh)
            w = tf.get_variable('w', [self.units, config.latent_size])
            b = tf.get_variable('b', [config.latent_size])
            latent_encoding += tf.nn.xw_plus_b(encoding, w, b)
            return latent_encoding

    def evidence_loss(self, psi, encoding, config):
        sigma_sq = tf.square(self.sigma)
        loss = 0.5 * (config.latent_size * tf.log(2 * np.pi * sigma_sq + 1e-10)
                      + tf.square(encoding - psi) / sigma_sq)
        return loss

    @staticmethod
    def from_call(callnode):
        call = callnode['_call']
        call = re.sub('^\$.*\$', '', call)  # get rid of predicates
        name = call.split('(')[0].split('.')[-1]
        name = name.split('<')[0]  # remove generics from call name
        return [name] if name[0].islower() else []  # Java convention

class Types(Evidence):

    def read_data_point(self, program):
        types = program['types'] if 'types' in program else []
        return list(set(types))

    def set_chars_vocab(self, data):
        counts = Counter([t for types in data for t in types])
        self.chars = sorted(counts.keys(), key=lambda w: counts[w], reverse=True)
        self.vocab = dict(zip(self.chars, range(len(self.chars))))
        self.vocab_size = len(self.vocab)

    def wrangle(self, data):
        wrangled = np.zeros((len(data), 1, self.vocab_size), dtype=np.int32)
        for i, types in enumerate(data):
            for t in types:
                if t in self.vocab:
                    wrangled[i, 0, self.vocab[t]] = 1
        return wrangled

    def placeholder(self, config):
        return tf.placeholder(tf.float32, [config.batch_size, 1, self.vocab_size])

    def exists(self, inputs):
        i = tf.reduce_sum(inputs, axis=2)
        return tf.not_equal(tf.count_nonzero(i, axis=1), 0)

    def init_sigma(self, config):
        with tf.variable_scope('types'):
            self.sigma = tf.get_variable('sigma', [])

    def encode(self, inputs, config):
        with tf.variable_scope('types'):
            latent_encoding = tf.zeros([config.batch_size, config.latent_size])
            inp = tf.slice(inputs, [0, 0, 0], [config.batch_size, 1, self.vocab_size])
            inp = tf.reshape(inp, [-1, self.vocab_size])
            encoding = tf.layers.dense(inp, self.units, activation=tf.nn.tanh)
            for i in range(self.num_layers - 1):
                encoding = tf.layers.dense(encoding, self.units, activation=tf.nn.tanh)
            w = tf.get_variable('w', [self.units, config.latent_size])
            b = tf.get_variable('b', [config.latent_size])
            latent_encoding += tf.nn.xw_plus_b(encoding, w, b)
            return latent_encoding

    def evidence_loss(self, psi, encoding, config):
        sigma_sq = tf.square(self.sigma)
        loss = 0.5 * (config.latent_size * tf.log(2 * np.pi * sigma_sq + 1e-10)
                      + tf.square(encoding - psi) / sigma_sq)
        return loss

    @staticmethod
    def get_types_re(s):
        patt = re.compile('java[x]?\.(\w*)\.(\w*)(\.([A-Z]\w*))*')
        types = [match.group(4) if match.group(4) is not None else match.group(2)
                 for match in re.finditer(patt, s)]
        primitives = {
            'byte': 'Byte',
            'short': 'Short',
            'int': 'Integer',
            'long': 'Long',
            'float': 'Float',
            'double': 'Double',
            'boolean': 'Boolean',
            'char': 'Character'
        }

        for p in primitives:
            if s == p or re.search('\W{}'.format(p), s):
                types.append(primitives[p])
        return list(set(types))

    @staticmethod
    def from_call(callnode):
        call = callnode['_call']
        types = Types.get_types_re(call)

        if '_throws' in callnode:
            for throw in callnode['_throws']:
                types += Types.get_types_re(throw)

        if '_returns' in callnode:
            types += Types.get_types_re(callnode['_returns'])

        return list(set(types))

class Keywords(Evidence):

    def __init__(self):
        nltk.download('wordnet')
        self.lemmatizer = WordNetLemmatizer()

    STOP_WORDS = {  # CoreNLP English stop words
        "'ll", "'s", "'m", "a", "about", "above", "after", "again", "against", "all", "am", "an", "and",
        "any", "are", "aren't", "as", "at", "be", "because", "been", "before", "being", "below", "between",
        "both", "but", "by", "can", "can't", "cannot", "could", "couldn't", "did", "didn't", "do", "does",
        "doesn't", "doing", "don't", "down", "during", "each", "few", "for", "from", "further", "had",
        "hadn't", "has", "hasn't", "have", "haven't", "having", "he", "he'd", "he'll", "he's", "her",
        "here", "here's", "hers", "herself", "him", "himself", "his", "how", "how's", "i", "i'd", "i'll",
        "i'm", "i've", "if", "in", "into", "is", "isn't", "it", "it's", "its", "itself", "let's", "me",
        "more", "most", "mustn't", "my", "myself", "no", "nor", "not", "of", "off", "on", "once", "only",
        "or", "other", "ought", "our", "ours", "ourselves", "out", "over", "own", "same", "shan't", "she",
        "she'd", "she'll", "she's", "should", "shouldn't", "so", "some", "such", "than", "that", "that's",
        "the", "their", "theirs", "them", "themselves", "then", "there", "there's", "these", "they",
        "they'd", "they'll", "they're", "they've", "this", "those", "through", "to", "too", "under",
        "until", "up", "very", "was", "wasn't", "we", "we'd", "we'll", "we're", "we've", "were", "weren't",
        "what", "what's", "when", "when's", "where", "where's", "which", "while", "who", "who's", "whom",
        "why", "why's", "with", "won't", "would", "wouldn't", "you", "you'd", "you'll", "you're", "you've",
        "your", "yours", "yourself", "yourselves", "return", "arent", "cant", "couldnt", "didnt", "doesnt",
        "dont", "hadnt", "hasnt", "havent", "hes", "heres", "hows", "im", "isnt", "its", "lets", "mustnt",
        "shant", "shes", "shouldnt", "thats", "theres", "theyll", "theyre", "theyve", "wasnt", "were",
        "werent", "whats", "whens", "wheres", "whos", "whys", "wont", "wouldnt", "youd", "youll", "youre",
        "youve"
    }

    def lemmatize(self, word):
        w = self.lemmatizer.lemmatize(word, 'v')
        return self.lemmatizer.lemmatize(w, 'n')

    def read_data_point(self, program):
        keywords = [self.lemmatize(k) for k in program['keywords']] if 'keywords' in program else []
        return list(set(keywords))

    def set_chars_vocab(self, data):
        counts = Counter([c for keywords in data for c in keywords])
        self.chars = sorted(counts.keys(), key=lambda w: counts[w], reverse=True)
        self.vocab = dict(zip(self.chars, range(len(self.chars))))
        self.vocab_size = len(self.vocab)

    def wrangle(self, data):
        wrangled = np.zeros((len(data), 1, self.vocab_size), dtype=np.int32)
        for i, keywords in enumerate(data):
            for k in keywords:
                if k in self.vocab and k not in Keywords.STOP_WORDS:
                    wrangled[i, 0, self.vocab[k]] = 1
        return wrangled

    def placeholder(self, config):
        return tf.placeholder(tf.float32, [config.batch_size, 1, self.vocab_size])

    def exists(self, inputs):
        i = tf.reduce_sum(inputs, axis=2)
        return tf.not_equal(tf.count_nonzero(i, axis=1), 0)

    def init_sigma(self, config):
        with tf.variable_scope('keywords'):
            self.sigma = tf.get_variable('sigma', [])

    def encode(self, inputs, config):
        with tf.variable_scope('keywords'):
            latent_encoding = tf.zeros([config.batch_size, config.latent_size])
            inp = tf.slice(inputs, [0, 0, 0], [config.batch_size, 1, self.vocab_size])
            inp = tf.reshape(inp, [-1, self.vocab_size])
            encoding = tf.layers.dense(inp, self.units, activation=tf.nn.tanh)
            for i in range(self.num_layers - 1):
                encoding = tf.layers.dense(encoding, self.units, activation=tf.nn.tanh)
            w = tf.get_variable('w', [self.units, config.latent_size])
            b = tf.get_variable('b', [config.latent_size])
            latent_encoding += tf.nn.xw_plus_b(encoding, w, b)
            return latent_encoding

    def evidence_loss(self, psi, encoding, config):
        sigma_sq = tf.square(self.sigma)
        loss = 0.5 * (config.latent_size * tf.log(2 * np.pi * sigma_sq + 1e-10)
                      + tf.square(encoding - psi) / sigma_sq)
        return loss

    @staticmethod
    def split_camel(s):
        s = re.sub('(.)([A-Z][a-z]+)', r'\1#\2', s)  # UC followed by LC
        s = re.sub('([a-z0-9])([A-Z])', r'\1#\2', s)  # LC followed by UC
        return s.split('#')

    @staticmethod
    def from_call(callnode):
        call = callnode['_call']
        call = re.sub('^\$.*\$', '', call)  # get rid of predicates
        qualified = call.split('(')[0]
        qualified = re.sub('<.*>', '', qualified).split('.')  # remove generics for keywords

        # add qualified names (java, util, xml, etc.), API calls and types
        keywords = list(chain.from_iterable([Keywords.split_camel(s) for s in qualified if s not in ['java', 'javax']])) + \
            list(chain.from_iterable([Keywords.split_camel(c) for c in APICalls.from_call(callnode)])) + \
            list(chain.from_iterable([Keywords.split_camel(t) for t in Types.from_call(callnode)]))

        # convert to lower case, omit stop words and take the set
        return list(set([k.lower() for k in keywords if k.lower() not in Keywords.STOP_WORDS]))

# TODO: handle Javadoc with word2vec
class Javadoc(Evidence):

    def read_data_point(self, program, infer=False):
        javadoc = program['javadoc'] if 'javadoc' in program else None
        if not javadoc:
            javadoc = UNK
        try:  # do not consider non-ASCII javadoc
            javadoc.encode('ascii')
        except UnicodeEncodeError:
            javadoc = UNK
        javadoc = javadoc.split()
        return javadoc

    def set_dicts(self, data):
        if self.pretrained_embed:
            save_dir = os.path.join(self.save_dir, 'embed_' + self.name)
            with open(os.path.join(save_dir, 'config.json')) as f:
                js = json.load(f)
            self.chars = js['chars']
        else:
            self.chars = [C0] + list(set([w for point in data for w in point]))
        self.vocab = dict(zip(self.chars, range(len(self.chars))))
        self.vocab_size = len(self.vocab)

ext: py

reason: import numpy


text:
# coding: utf8
from copy import copy
from array import array
from math import *

import numpy
from math import *

def model_snowaccumulation(tsmax = 0.0,
         tmax = 0.0,
         trmax = 0.0,
         precip = 0.0):
    """
     - Name: SnowAccumulation -Version: 1.0, -Time step: 1
     - Description:
                 * Title: snowfall accumulation  calculation
                 * Author: STICS
                 * Reference: doi:http://dx.doi.org/10.1016/j.agrformet.2014.05.002
                 * Institution: INRA
                 * Abstract: It simulates the depth of snow cover and recalculate weather data
     - inputs:
                 * name: tsmax
                               ** description : maximum daily air temperature (tmax) below which all precipitation is assumed to be snow
                               ** inputtype : parameter
                               ** parametercategory : constant
                               ** datatype : DOUBLE
                               ** default : 0.0
                               ** min : 0.0
                               ** max : 1000
                               ** unit : degC
                               ** uri :
                 * name: tmax
                               ** description : current maximum air temperature
                               ** inputtype : variable
                               ** variablecategory : auxiliary
                               ** datatype : DOUBLE
                               ** default : 0.0
                               ** min : 0.0
                               ** max : 5000.0
                               ** unit : degC
                               ** uri :
                 * name: trmax
                               ** description : tmax above which all precipitation is assumed to be rain
                               ** inputtype : parameter
                               ** parametercategory : constant
                               ** datatype : DOUBLE
                               ** default : 0.0
                               ** min : 0.0
                               ** max : 5000.0
                               ** unit : degC
                               ** uri :
                 * name: precip
                               ** description : current precipitation
                               ** inputtype : variable
                               ** variablecategory : auxiliary
                               ** datatype : DOUBLE
                               ** default : 0.0
                               ** min : 0.0
                               ** max : 5000.0
                               ** unit : mmW
                               ** uri :
     - outputs:
                 * name: Snowaccu
                               ** description : snowfall accumulation
                               ** variablecategory : rate
                               ** datatype : DOUBLE
                               ** min : 0.0
                               ** max : 500.0
                               ** unit : mmW/d
                               ** uri :
    """

    fs = 0.0
    if tmax < tsmax:
        fs = 1.0
    if tmax >= tsmax and tmax <= trmax:
        fs = (trmax - tmax) / (trmax - tsmax)
    Snowaccu = fs * precip
    return Snowaccu


text:
# -*- coding: utf-8 -*-
"""
Runs many queries and keeps track of some results
"""
import logging
import sys
import textwrap
import numpy as np  # NOQA
import utool as ut
from wbia.expt import experiment_helpers
from wbia.expt import test_result

print, rrr, profile = ut.inject2(__name__)
logger = logging.getLogger('wbia')

NOMEMORY = ut.get_argflag('--nomemory')
TESTRES_VERBOSITY = 2 - (2 * ut.QUIET)
NOCACHE_TESTRES = ut.get_argflag(('--nocache-testres', '--nocache-big'), False)
USE_BIG_TEST_CACHE = (
    not ut.get_argflag(('--no-use-testcache', '--nocache-test'))
    and ut.USE_CACHE
    and not NOCACHE_TESTRES
)
USE_BIG_TEST_CACHE = False
TEST_INFO = True

# dont actually query. Just print labels and stuff
DRY_RUN = ut.get_argflag(('--dryrun', '--dry'))

def run_expt(
    ibs,
    acfg_name_list,
    test_cfg_name_list,
    use_cache=None,
    qaid_override=None,
    daid_override=None,
    initial_aids=None,
):
    r"""
    Loops over annot configs.

    Try and use this function as a starting point to clean up this module.
    The code is getting too untenable.

    CommandLine:
        python -m wbia.expt.harness run_expt --acfginfo
        python -m wbia.expt.harness run_expt --pcfginfo
        python -m wbia.expt.harness run_expt

    Ignore:
        test_cfg_name_list = [p]

    Example:
        >>> # SLOW_DOCTEST
        >>> from wbia.expt.harness import *  # NOQA
        >>> import wbia
        >>> ibs = wbia.opendb(defaultdb='PZ_MTEST')
        >>> default_acfgstrs = ['ctrl:qsize=20,dpername=1,dsize=10',
        >>>                     'ctrl:qsize=20,dpername=10,dsize=20']
        >>> acfg_name_list = default_acfgstrs
        >>> test_cfg_name_list = ['default:proot=smk', 'default']
        >>> #test_cfg_name_list = ['custom', 'custom:fg_on=False']
        >>> use_cache = False
        >>> testres_list = run_expt(ibs, acfg_name_list, test_cfg_name_list, use_cache)
    """
    logger.info('[harn] run_expt')
    # Generate list of database annotation configurations
    if len(acfg_name_list) == 0:
        raise ValueError('must give acfg name list')

    acfg_list, expanded_aids_list = experiment_helpers.get_annotcfg_list(
        ibs,
        acfg_name_list,
        qaid_override=qaid_override,
        daid_override=daid_override,
        initial_aids=initial_aids,
        use_cache=use_cache,
    )

    # Generate list of query pipeline param configs
    cfgdict_list, pipecfg_list = experiment_helpers.get_pipecfg_list(
        test_cfg_name_list, ibs=ibs
    )

    cfgx2_lbl = experiment_helpers.get_varied_pipecfg_lbls(cfgdict_list)
    # NOTE: Can specify --pcfginfo or --acfginfo

    if ut.NOT_QUIET:
        ut.colorprint(
            textwrap.dedent(
                """

        [harn]================
        [harn] harness.test_configurations2()"""
            ).strip(),
            'white',
        )
        msg = '[harn] Running %s using %s and %s' % (
            ut.quantstr('test', len(acfg_list) * len(cfgdict_list)),
            ut.quantstr('pipeline config', len(cfgdict_list)),
            ut.quantstr('annot config', len(acfg_list)),
        )
        ut.colorprint(msg, 'white')

    testres_list = []

    nAcfg = len(acfg_list)

    testnameid = ibs.get_dbname() + ' ' + str(test_cfg_name_list) + str(acfg_name_list)
    lbl = '[harn] TEST_CFG ' + str(test_cfg_name_list) + str(acfg_name_list)
    expanded_aids_iter = ut.ProgIter(
        expanded_aids_list,
        lbl='annot config',
        freq=1,
        autoadjust=False,
        enabled=ut.NOT_QUIET,
    )

    for acfgx, (qaids, daids) in enumerate(expanded_aids_iter):
        assert len(qaids) != 0, '[harness] No query annots specified'
        assert len(daids) != 0, '[harness] No database annotas specified'
        acfg = acfg_list[acfgx]
        if ut.NOT_QUIET:
            ut.colorprint('\n---Annot config testnameid=%r' % (testnameid,), 'brightcyan')
        subindexer_partial = ut.ProgPartial(
            parent_index=acfgx, parent_length=nAcfg, enabled=ut.NOT_QUIET
        )
        testres_ = make_single_testres(
            ibs,
            qaids,
            daids,
            pipecfg_list,
            cfgx2_lbl,
            cfgdict_list,
            lbl,
            testnameid,
            use_cache=use_cache,
            subindexer_partial=subindexer_partial,
        )
        if DRY_RUN:
            continue
        testres_.acfg = acfg
        testres_.test_cfg_name_list = test_cfg_name_list
        testres_list.append(testres_)
    if DRY_RUN:
        logger.info('DRYRUN: Cannot continue past run_expt')
        sys.exit(0)

    testres = test_result.combine_testres_list(ibs, testres_list)
    # testres.print_results()
    logger.info('Returning Test Result')
    return testres

@profile
def make_single_testres(
    ibs,
    qaids,
    daids,
    pipecfg_list,
    cfgx2_lbl,
    cfgdict_list,
    lbl,
    testnameid,
    use_cache=None,
    subindexer_partial=ut.ProgIter,
):
    """
    CommandLine:
        python -m wbia run_expt
    """
    cfgslice = None
    if cfgslice is not None:
        pipecfg_list = pipecfg_list[cfgslice]

    dbname = ibs.get_dbname()

    # if ut.NOT_QUIET:
    #     logger.info('[harn] Make single testres')

    cfgx2_qreq_ = [
        ibs.new_query_request(qaids, daids, verbose=False, query_cfg=pipe_cfg)
        for pipe_cfg in ut.ProgIter(pipecfg_list, lbl='Building qreq_', enabled=False)
    ]

    if use_cache is None:
        use_cache = USE_BIG_TEST_CACHE

    if use_cache:
        try:
            bt_cachedir = ut.ensuredir((ibs.get_cachedir(), 'BULK_TEST_CACHE2'))
            cfgstr_list = [qreq_.get_cfgstr(with_input=True) for qreq_ in cfgx2_qreq_]
            bt_cachestr = ut.hashstr_arr27(cfgstr_list, ibs.get_dbname() + '_cfgs')
            bt_cachename = 'BULKTESTCACHE2_v2'
            testres = ut.load_cache(bt_cachedir, bt_cachename, bt_cachestr)
            testres.cfgdict_list = cfgdict_list
            testres.cfgx2_lbl = cfgx2_lbl  # hack override
        except IOError:
            pass
        else:
            if ut.NOT_QUIET:
                ut.colorprint(
                    '[harn] single testres cache hit... returning', 'brightcyan'
                )
            return testres

    if ibs.table_cache:
        # HACK
        prev_feat_cfgstr = None

    cfgx2_cmsinfo = []
    cfgiter = subindexer_partial(
        range(len(cfgx2_qreq_)), lbl='pipe config', freq=1, adjust=False
    )
    # Run each pipeline configuration
    for cfgx in cfgiter:
        qreq_ = cfgx2_qreq_[cfgx]
        cprint = ut.colorprint
        cprint('testnameid=%r' % (testnameid,), 'green')
        cprint(
            'annot_cfgstr = %s' % (qreq_.get_cfgstr(with_input=True, with_pipe=False),),
            'yellow',
        )
        cprint('pipe_cfgstr= %s' % (qreq_.get_cfgstr(with_data=False),), 'brightcyan')

        cprint('pipe_hashstr = %s' % (qreq_.get_pipe_hashid(),), 'cyan')
        if DRY_RUN:
            continue

        indent_prefix = '[%s cfg %d/%d]' % (
            dbname,
            # cfgiter.count (doesnt work when quiet)
            (cfgiter.parent_index * cfgiter.length) + cfgx,
            cfgiter.length * cfgiter.parent_length,
        )

        with ut.Indenter(indent_prefix):
            # Run the test / read cache
            _need_compute = True
            if use_cache:
                # smaller cache for individual configuration runs
                st_cfgstr = qreq_.get_cfgstr(with_input=True)
                st_cachedir = ut.unixjoin(bt_cachedir, 'small_tests')
                st_cachename = 'smalltest'
                ut.ensuredir(st_cachedir)
                try:
                    cmsinfo = ut.load_cache(st_cachedir, st_cachename, st_cfgstr)
                except IOError:
                    _need_compute = True
                else:
                    _need_compute = False
            if _need_compute:
                assert not ibs.table_cache
                if ibs.table_cache:
                    if len(
                        prev_feat_cfgstr is not None
                        and prev_feat_cfgstr != qreq_.qparams.feat_cfgstr
                    ):
                        # Clear features to preserve memory
                        ibs.clear_table_cache()
                        # qreq_.ibs.print_cachestats_str()
                cm_list = qreq_.execute()
                cmsinfo = test_result.build_cmsinfo(cm_list, qreq_)
                # record previous feature configuration
                if ibs.table_cache:
                    prev_feat_cfgstr = qreq_.qparams.feat_cfgstr
                if use_cache:
                    ut.save_cache(st_cachedir, st_cachename, st_cfgstr, cmsinfo)
        if not NOMEMORY:
            # Store the results
            cfgx2_cmsinfo.append(cmsinfo)
        else:
            cfgx2_qreq_[cfgx] = None
    if ut.NOT_QUIET:
        ut.colorprint('[harn] Completed running test configurations', 'white')
    if DRY_RUN:
        logger.info('ran tests dryrun mode.')
        return
    if NOMEMORY:
        logger.info('ran tests in memory savings mode. Cannot Print. exiting')
        return
    # Store all pipeline config results in a test result object
    testres = test_result.TestResult(pipecfg_list, cfgx2_lbl, cfgx2_cmsinfo, cfgx2_qreq_)
    testres.testnameid = testnameid
    testres.lbl = lbl
    testres.cfgdict_list = cfgdict_list
    testres.aidcfg = None
    if use_cache:
        try:
            ut.save_cache(bt_cachedir, bt_cachename, bt_cachestr, testres)
        except Exception as ex:
            ut.printex(ex, 'error saving testres cache', iswarning=True)
            if ut.SUPER_STRICT:
                raise
    return testres

max_stars_repo_name: dylanirion/wildbook-ia

max_issues_repo_name: solomonkimunyu/wildbook-ia

max_forks_repo_name: solomonkimunyu/wildbook-ia


text:
from __future__ import annotations

import ast
from typing import Any, Callable, Optional, cast

import pytest

from dataframe_expressions import (DataFrame, ast_DataFrame, exclusive_class,
                                   render)

from .utils_for_testing import reset_var_counter  # NOQA

class multi_leaf_object(DataFrame):
    def __init__(self, df: DataFrame):
        DataFrame.__init__(self, expr=ast_DataFrame(df))

    @property
    def x1(self) -> DataFrame:
        return self.x_new_1

class leaf_object(DataFrame):
    def __init__(self, df: DataFrame):
        DataFrame.__init__(self, expr=ast_DataFrame(df))

    @property
    def x2(self) -> DataFrame:
        return self.x1

def test_collection_object():
    df = DataFrame()
    mlo = multi_leaf_object(df)
    df1 = mlo.x1

    expr, _ = render(df1)
    assert ast.dump(expr) == "Attribute(value=ast_DataFrame(), attr='x_new_1', ctx=Load())"

def test_collection_subtract():
    df = DataFrame()
    mlo1 = multi_leaf_object(df.m1)
    mlo2 = multi_leaf_object(df.m2)
    df1 = mlo1-mlo2

    expr, _ = render(df1)
    assert ast.dump(expr) == "BinOp(left=Attribute(value=ast_DataFrame(), attr='m1', " \
        "ctx=Load()), op=Sub(), right=Attribute(value=ast_DataFrame(), attr='m2', ctx=Load()))"

def test_collection_object_other():
    df = DataFrame()
    mlo = multi_leaf_object(df)
    df1 = mlo.x12

    expr, _ = render(df1)
    assert ast.dump(expr) == "Attribute(value=ast_DataFrame(), attr='x12', ctx=Load())"

def test_collection_nested():
    df = DataFrame()
    mlo = multi_leaf_object(df)
    ml1 = leaf_object(mlo)
    df1 = ml1.x2

    expr, _ = render(df1)
    assert ast.dump(expr) == "Attribute(value=ast_DataFrame(), attr='x_new_1', ctx=Load())"

@exclusive_class
class multi_leaf_object_excl(DataFrame):
    def __init__(self, df: DataFrame):
        DataFrame.__init__(self, expr=ast_DataFrame(df))
        self.__p = df

    @property
    def x1(self) -> DataFrame:
        return self.__p.x_new_1

@exclusive_class
class leaf_object_excl(DataFrame):
    def __init__(self, df: DataFrame):
        DataFrame.__init__(self, expr=ast_DataFrame(df))
        self.__p = df

    @property
    def x2(self) -> DataFrame:
        return self.__p.x1

def test_collection_object_excl():
    df = DataFrame()
    mlo = multi_leaf_object_excl(df)
    df1 = mlo.x1

    expr, _ = render(df1)
    assert ast.dump(expr) == "Attribute(value=ast_DataFrame(), attr='x_new_1', ctx=Load())"

def test_collection_subtract_excl():
    df = DataFrame()
    mlo1 = multi_leaf_object_excl(df.m1)
    mlo2 = multi_leaf_object_excl(df.m2)
    with pytest.raises(Exception) as e:
        mlo1-mlo2

    assert "operator" in str(e.value)
    assert "multi_leaf_object" in str(e.value)

def test_collection_object_other_excl():
    with pytest.raises(Exception) as e:
        df = DataFrame()
        mlo = multi_leaf_object_excl(df)
        df1 = mlo.x12

        expr, _ = render(df1)

    assert 'No such attribute' in str(e.value)

def test_collection_nested_excl():
    df = DataFrame()
    mlo = multi_leaf_object_excl(df)
    ml1 = leaf_object_excl(mlo)
    df1 = ml1.x2

    expr, _ = render(df1)
    assert ast.dump(expr) == "Attribute(value=ast_DataFrame(), attr='x_new_1', ctx=Load())"

class op_base:
    def render(self, f: Callable[[vec], Any]) -> Any:
        assert False, 'not implemented'

class op_bin(op_base):
    def __init__(self, a: op_base, b: op_base):
        self._a = a
        self._b = b

    def render(self, f: Callable[[vec], Any]) -> Any:
        return self._a.render(f) + self._b.render(f)

class op_vec(op_base):
    def __init__(self, a: vec):
        self._a = a

    def render(self, f: Callable[[vec], Any]) -> Any:
        return f(self._a)

class vec(DataFrame):
    def __init__(self, df: Optional[DataFrame], compound: Optional[op_base] = None) -> None:
        DataFrame.__init__(self, expr=ast_DataFrame(df))
        self._ref: op_base = compound if compound is not None else op_vec(self)

    @property
    def xy(self) -> DataFrame:
        from numpy import sqrt
        bx = self._ref.render(lambda v: v.x)
        by = self._ref.render(lambda v: v.y)
        return sqrt(bx*bx + by*by)

    def __add__(self, other: vec) -> vec:
        'Do the addition'
        return vec(None, op_bin(self._ref, other._ref))

def test_xy():
    df = DataFrame()
    v = vec(df)
    df1 = v.xy

    expr, _ = render(df1)
    assert isinstance(expr, ast.Call)
    assert isinstance(expr.func, ast.Name)
    assert expr.func.id == 'sqrt'
    assert len(expr.args) == 1
    assert isinstance(expr.args[0], ast.BinOp)
    assert isinstance(cast(ast.BinOp, expr.args[0]).op, ast.Add)

def test_add_xy():
    df = DataFrame()
    x1 = vec(df.x)
    x2 = vec(df.y)
    df1 = (x1 + x2).xy

    expr, _ = render(df1)
    assert isinstance(expr, ast.Call)
    assert isinstance(expr.func, ast.Name)
    assert expr.func.id == 'sqrt'
    assert isinstance(expr.args[0], ast.BinOp)
    assert isinstance(cast(ast.BinOp, expr.args[0]).op, ast.Add)

    x_component2 = cast(ast.BinOp, expr.args[0]).left
    assert isinstance(x_component2, ast.BinOp)
    assert isinstance(x_component2.op, ast.Mult)

    x_component = x_component2.left
    assert isinstance(x_component, ast.BinOp)
    assert ast.dump(x_component) == "BinOp(left=Attribute(value=Attribute(value=ast_DataFrame()," \
        " attr='x', ctx=Load()), attr='x', ctx=Load()), " \
        "op=Add(), " \
        "right=Attribute(value=Attribute(value=ast_DataFrame(), attr='y', ctx=Load()), " \
        "attr='x', ctx=Load()))"

reason: from numpy


text:
# SPDX-License-Identifier: BSD-3-Clause
# Copyright (c) 2022 Scipp contributors (https://github.com/scipp)
# @author Neil Vaytet

from .figure1d import PlotFigure1d
import numpy as np

class PlotProfile(PlotFigure1d):
    """
    Class for 1 dimensional profile plots, that are displayed as 1d slices into
    a higher dimensional data.

    This is using the `PlotFigure1d` extensively which can display 1d lines and
    keep or remove lines for comparison of spectra.
    In addition, it has a `slice_area` indicator to show the current range
    occupied by the slice that is displayed in the main subplot as a grey
    rectangle (using `axvspan`).

    Finally, it shows or hide the profile line depending on whether the mouse
    cursor is inside the main plot axes or not (`toggle_hover_visibility`).
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.current_visible_state = False
        self.slice_area = None
        self.visible = False
        self._xstart = 1
        self._xend = 2

    def _to_widget(self):
        """
        When using the inline backend, the figure is rendered as an image, so
        if we want to hide the profile plot, we need to hide the entire widget
        and not just the canvas, as in toggle_view().
        """
        widg = super()._to_widget()
        if not self.is_widget():
            widg.layout.display = None if self.visible else 'none'
        return widg

    def update_axes(self, *args, **kwargs):
        """
        Upon axes update, we reset the slice area indicator.
        """
        super().update_axes(*args, legend_labels=False, **kwargs)
        # PlotFigure1d.update_axes clears self.ax
        self.slice_area = self.ax.axvspan(self._xstart,
                                          self._xend,
                                          alpha=0.5,
                                          color='lightgrey')

    def toggle_hover_visibility(self, value):
        """
        If the mouse moves off the image, we hide the profile. If it moves
        back onto the image, we show the profile.
        """
        for line in self._lines.values():
            line.data.set_visible(value)
            # Need to get the 3rd element of the errorbar container, which
            # contains the vertical errorbars, and then the first element of
            # that because it is a tuple itself.
            if line.error is not None:
                line.error[2][0].set_visible(value)
            for ml in line.masks.values():
                ml.set_visible(value)
                ml.set_gid("onaxes" if value else "offaxes")
        if value != self.current_visible_state:
            if value:
                self.rescale_to_data()
            self.current_visible_state = value

    def set_slice_area(self, xstart, xend):
        """
        When the data slice is updated (position or thickness), we update the
        slice area indicator.
        """
        self._xstart = xstart
        self._xend = xend
        if self.slice_area is not None:
            new_xy = np.array([[xstart, 0.0], [xstart, 1.0], [xend, 1.0], [xend, 0.0],
                               [xstart, 0.0]])
            self.slice_area.set_xy(new_xy)
            self.draw()

    def toggle_view(self, visible=True):
        """
        Show or hide profile view.
        """
        if self.is_widget():
            self.fig.canvas.layout.display = None if visible else 'none'
            self.toolbar.set_visible(visible)
        self.visible = visible

    def is_visible(self):
        """
        Get visible state of profile view.
        """
        return self.visible

    def connect(self):
        """
        For the profile, we connect the log buttons of the toolbar directly to
        callbacks local to `PlotProfile`, since all we need to do is toggle
        the scale on the matplotlib axes.
        """
        if self.toolbar is not None:
            self.toolbar.set_resampling_mode_display(False)
            self.toolbar.connect(controller=self)

    def toggle_dim_scale(self, dim):
        """
        Toggle x-axis scale from toolbar button signal.
        """
        def toggle(change):
            self.ax.set_xscale("log" if change['new'] else "linear")
            self.draw()

        return toggle

    def toggle_norm(self, owner):
        """
        Toggle y-axis scale from toolbar button signal.
        """
        self.ax.set_yscale("log" if owner.value else "linear")
        self.draw()

max_stars_repo_name: mlund/scipp

max_issues_repo_name: mlund/scipp

max_forks_repo_name: mlund/scipp


text:
import astropy.units as u
import numpy as np

from einsteinpy import constant
from einsteinpy.ijit import jit

_c = constant.c.value

def cartesian_to_spherical_fast(
    x, y, z, v_x=None, v_y=None, v_z=None, velocities_provided=False
):
    if velocities_provided:
        return cartesian_to_spherical(x, y, z, v_x, v_y, v_z)
    return cartesian_to_spherical_novel(x, y, z)

@jit
def cartesian_to_spherical(x, y, z, v_x, v_y, v_z):
    """
    Utility function (jitted) to convert cartesian to spherical.
    This function should eventually result in Coordinate Transformation Graph!
    """
    hxy = np.hypot(x, y)
    r = np.hypot(hxy, z)
    theta = np.arctan2(hxy, z)
    phi = np.arctan2(y, x)
    n1 = x ** 2 + y ** 2
    n2 = n1 + z ** 2
    v_r = (x * v_x + y * v_y + z * v_z) / np.sqrt(n2)
    v_t = (z * (x * v_x + y * v_y) - n1 * v_z) / (n2 * np.sqrt(n1))
    v_p = -1 * (v_x * y - x * v_y) / n1

    return r, theta, phi, v_r, v_t, v_p

@jit
def cartesian_to_spherical_novel(x, y, z):
    """
    Utility function (jitted) to convert cartesian to spherical.
    This function should eventually result in Coordinate Transformation Graph!
    """
    hxy = np.hypot(x, y)
    r = np.hypot(hxy, z)
    theta = np.arctan2(hxy, z)
    phi = np.arctan2(y, x)

    return r, theta, phi

def cartesian_to_bl_fast(
    x, y, z, a, v_x=None, v_y=None, v_z=None, velocities_provided=False
):
    if velocities_provided:
        return cartesian_to_bl(x, y, z, a, v_x, v_y, v_z)
    return cartesian_to_bl_novel(x, y, z, a)

@jit
def cartesian_to_bl(x, y, z, a, v_x, v_y, v_z):
    """
    Utility function (jitted) to convert cartesian to boyer lindquist.
    This function should eventually result in Coordinate Transformation Graph!
    """
    w = (x ** 2 + y ** 2 + z ** 2) - (a ** 2)
    r = np.sqrt(0.5 * (w + np.sqrt((w ** 2) + (4 * (a ** 2) * (z ** 2)))))
    theta = np.arccos(z / r)
    phi = np.arctan2(y, x)
    dw_dt = 2 * (x * v_x + y * v_y + z * v_z)
    v_r = (1 / (2 * r)) * (
        (dw_dt / 2)
        + (
            (w * dw_dt + 4 * (a ** 2) * z * v_z)
            / (2 * np.sqrt((w ** 2) + (4 * (a ** 2) * (z ** 2))))
        )
    )
    v_t = (-1 / np.sqrt(1 - np.square(z / r))) * ((v_z * r - v_r * z) / (r ** 2))
    v_p = (1 / (1 + np.square(y / x))) * ((v_y * x - v_x * y) / (x ** 2))

    return r, theta, phi, v_r, v_t, v_p, a

@jit
def cartesian_to_bl_novel(x, y, z, a):
    """
    Utility function (jitted) to convert cartesian to boyer lindquist.
    This function should eventually result in Coordinate Transformation Graph!
    """
    w = (x ** 2 + y ** 2 + z ** 2) - (a ** 2)
    r = np.sqrt(0.5 * (w + np.sqrt((w ** 2) + (4 * (a ** 2) * (z ** 2)))))
    theta = np.arccos(z / r)
    phi = np.arctan2(y, x)

    return r, theta, phi, a

def spherical_to_cartesian_fast(
    r, t, p, v_r=None, v_t=None, v_p=None, velocities_provided=False
):
    if velocities_provided:
        return spherical_to_cartesian(r, t, p, v_r, v_t, v_p)
    return spherical_to_cartesian_novel(r, t, p)

@jit
def spherical_to_cartesian(r, t, p, v_r, v_t, v_p):
    """
    Utility function (jitted) to convert spherical to cartesian.
    This function should eventually result in Coordinate Transformation Graph!
    """
    x = r * np.cos(p) * np.sin(t)
    y = r * np.sin(p) * np.sin(t)
    z = r * np.cos(t)
    v_x = (
        np.sin(t) * np.cos(p) * v_r
        - r * np.sin(t) * np.sin(p) * v_p
        + r * np.cos(t) * np.cos(p) * v_t
    )
    v_y = (
        np.sin(t) * np.sin(p) * v_r
        + r * np.cos(t) * np.sin(p) * v_t
        + r * np.sin(t) * np.cos(p) * v_p
    )
    v_z = np.cos(t) * v_r - r * np.sin(t) * v_t
    return x, y, z, v_x, v_y, v_z

@jit
def spherical_to_cartesian_novel(r, t, p):
    """
    Utility function (jitted) to convert spherical to cartesian.
    This function should eventually result in Coordinate Transformation Graph!
    """
    x = r * np.cos(p) * np.sin(t)
    y = r * np.sin(p) * np.sin(t)
    z = r * np.cos(t)
    return x, y, z

def bl_to_cartesian_fast(
    r, t, p, a, v_r=None, v_t=None, v_p=None, velocities_provided=False
):
    if velocities_provided:
        return bl_to_cartesian(r, t, p, a, v_r, v_t, v_p)
    return bl_to_cartesian_novel(r, t, p, a)

@jit
def bl_to_cartesian(r, t, p, a, v_r, v_t, v_p):
    """
    Utility function (jitted) to convert bl to cartesian.
    This function should eventually result in Coordinate Transformation Graph!
    """
    xa = np.sqrt(r ** 2 + a ** 2)
    sin_norm = xa * np.sin(t)
    x = sin_norm * np.cos(p)
    y = sin_norm * np.sin(p)
    z = r * np.cos(t)
    v_x = (
        (r * v_r * np.sin(t) * np.cos(p) / xa)
        + (xa * np.cos(t) * np.cos(p) * v_t)
        - (xa * np.sin(t) * np.sin(p) * v_p)
    )
    v_y = (
        (r * v_r * np.sin(t) * np.sin(p) / xa)
        + (xa * np.cos(t) * np.sin(p) * v_t)
        + (xa * np.sin(t) * np.cos(p) * v_p)
    )
    v_z = (v_r * np.cos(t)) - (r * np.sin(t) * v_t)
    return x, y, z, v_x, v_y, v_z

@jit
def bl_to_cartesian_novel(r, t, p, a):
    """
    Utility function (jitted) to convert bl to cartesian.
    This function should eventually result in Coordinate Transformation Graph!
    """
    xa = np.sqrt(r ** 2 + a ** 2)
    sin_norm = xa * np.sin(t)
    x = sin_norm * np.cos(p)
    y = sin_norm * np.sin(p)
    z = r * np.cos(t)
    return x, y, z

@jit
def lorentz_factor(v_vec):
    """
    Returns the Lorentz Factor, ``gamma``

    Parameters
    ----------
    v_vec : ~numpy.ndarray
        Velocity 3-Vector

    Returns
    -------
    gamma : float
        Lorentz Factor

    """
    v_norm2 = v_vec.dot(v_vec)
    gamma = 1 / np.sqrt(1 - v_norm2 / _c ** 2)

    return gamma

def v_t(g_cov_mat, v_vec, time_like=True):
    """
    Utility function to return Timelike component of 4-Velocity
    Assumes a (+, -, -, -) Metric Signature

    Parameters
    ----------
    g_cov_mat : ~numpy.ndarray
        Matrix, containing Covariant Metric \
        Tensor values, in same coordinates as ``v_vec``
        Numpy array of shape (4,4)
    v_vec : ~numpy.ndarray
        Velocity 3-Vector
    time_like : bool, optional
        To determine, if the 4-Velocity is for a Time-like or \
        a Null-like Geodesic
        Defaults to ``True``

    Returns
    -------
    v_t : float
        Timelike component of 4-Velocity

    """
    u1, u2, u3 = v_vec
    g = g_cov_mat
    # Factor to add to coefficient, C
    fac = -1 if time_like else 0
    # Defining coefficients for quadratic equation
    A = g[0, 0]
    B = 2 * (g[0, 1] * u1 + g[0, 2] * u2 + g[0, 3] * u3)
    C = (
        (g[1, 1] * u1 ** 2 + g[2, 2] * u2 ** 2 + g[3, 3] * u3 ** 2)
        + 2 * u1 * (g[1, 2] * u2 + g[1, 3] * u3)
        + 2 * u2 * g[2, 3] * u3
        + fac
    )
    D = (B ** 2) - (4 * A * C)

    v_t = (-B + np.sqrt(D)) / (2 * A)

    return v_t * u.one

def four_position(t, x_vec):
    """
    Utility function to return 4-Position

    Parameters
    ----------
    t : float
        Coordinate Time
    x_vec : ~numpy.ndarray
        Position 3-Vector

    Returns
    -------
    x_4vec : ~numpy.ndarray
        Position 4-Vector

    """
    x_4vec = np.append([_c * t], x_vec)

    return x_4vec

def four_velocity(g_cov_mat, v_vec, time_like=True):
    """
    Utility function to return 4-Velocity

    Parameters
    ----------
    g_cov_mat : ~numpy.ndarray
        Matrix, containing Covariant Metric \
        Tensor values, in same coordinates as ``v_vec``
        Numpy array of shape (4,4)
    v_vec : ~numpy.ndarray
        Velocity 3-Vector
    time_like : bool, optional
        To determine, if the 4-Velocity is for a Time-like or \
        a Null-like Geodesic
        Defaults to ``True``

    Returns
    -------
    v_4vec : ~numpy.ndarray
        Velocity 4-Vector

    """
    v_4vec = np.append(v_t(g_cov_mat, v_vec, time_like), v_vec)

    return v_4vec

def stacked_vec(g_cov_mat, t, x_vec, v_vec, time_like=True):
    """
    Packages 4-Position and 4-Velocity into a Length-8 vector

    Parameters
    ----------
    g_cov_mat : ~numpy.ndarray
        Matrix, containing Covariant Metric \
        Tensor values, in same coordinates as \
        ``x_vec`` or ``v_vec``
        Numpy array of shape (4,4)
    t : float
        Coordinate Time
    x_vec : ~numpy.ndarray
        Position 3-Vector
    v_vec : ~numpy.ndarray
        Velocity 3-Vector
    time_like : bool, optional
        To determine, if the 4-Velocity is for a Time-like or \
        a Null-like Geodesic
        Defaults to ``True``

    Returns
    -------
    stacked_vec : ~numpy.ndarray
        Length-8 Vector of form [x0, x1, x2, x3, v0, v1, v2, v3]

    """
    x_4vec = four_position(t, x_vec)
    v_4vec = four_velocity(g_cov_mat, v_vec, time_like)

    stacked_vec = np.hstack((x_4vec, v_4vec))

    return stacked_vec

reason: import numpy,import astropy


text:
#!/usr/bin/env python

from __future__ import division
import numpy as np
from scipy.linalg import hankel
import scipy.io as sio
import matplotlib.pyplot as plt

from ..tools import *

def cum2est(y, maxlag, nsamp, overlap, flag):
  """
  CUM2EST Covariance function.
  Should be involed via "CUMEST" for proper parameter checks.
  Parameters:
           y: input data vector (column)
      maxlag: maximum lag to be computed
    samp_seg: samples per segment (<=0 means no segmentation)
     overlap: percentage overlap of segments
        flag: 'biased', biased estimates are computed
              'unbiased', unbiased estimates are computed.

  Output:
       y_cum: estimated covariance,
              C2(m)  -maxlag <= m <= maxlag
  """

  (n1, n2) = shape(y, 2)
  N = n1*n2
  overlap  = np.fix(overlap/100 * nsamp)
  nrecord  = np.fix((N - overlap)/(nsamp - overlap))
  nadvance = nsamp - overlap

  y_cum    = np.zeros([maxlag+1, 1])
  ind = np.arange(nsamp)
  y = y.ravel(order='F')

  for i in xrange(nrecord):
    x = y[ind]
    x = x - np.mean(x)

    for k in xrange(maxlag+1):
      y_cum[k] = y_cum[k] + np.dot(x[0:nsamp-k].T, x[k:nsamp])

    ind = ind + int(nadvance)

  if flag == 'biased':
    y_cum = y_cum / (nsamp * nrecord)
  else:
    y_cum = y_cum / (nrecord * (nsamp-np.matrix(range(maxlag+1)).T))
    y_cum = np.asarray(y_cum)

  if maxlag > 0:
    y_cum = make_arr([np.conj(y_cum[maxlag:0:-1]), y_cum], axis=0)

  return y_cum

max_stars_repo_name: tamphi/spectrum

max_issues_repo_name: tamphi/spectrum

max_forks_repo_name: tamphi/spectrum

reason: import numpy,import scipy,from scipy


text:
from .. import settings
from .. import logging as logg
from .utils import norm
from .transition_matrix import transition_matrix

from scipy.sparse import issparse
import numpy as np
import warnings

def quiver_autoscale(X_emb, V_emb):
    import matplotlib.pyplot as pl
    scale_factor = np.abs(X_emb).max()  # just so that it handles very large values
    fig, ax = pl.subplots()
    Q = ax.quiver(X_emb[:, 0] / scale_factor, X_emb[:, 1] / scale_factor,
                  V_emb[:, 0], V_emb[:, 1], angles='xy', scale_units='xy', scale=None)
    Q._init()
    fig.clf()
    #fig.close()
    return Q.scale / scale_factor

def velocity_embedding(data, basis=None, vkey='velocity', scale=10, self_transitions=True, use_negative_cosines=True,
                       direct_pca_projection=None, retain_scale=False, autoscale=True, all_comps=True, T=None, copy=False):
    """Computes the single cell velocities in the embedding

    Arguments
    ---------
    data: :class:`~anndata.AnnData`
        Annotated data matrix.
    basis: `str` (default: `'tsne'`)
        Which embedding to use.
    vkey: `str` (default: `'velocity'`)
        Name of velocity estimates to be used.
    scale: `int` (default: 10)
        Scale parameter of gaussian kernel for transition matrix.
    self_transitions: `bool` (default: `True`)
        Whether to allow self transitions, based on the confidences of transitioning to neighboring cell.
    use_negative_cosines: `bool` (default: `True`)
        Whether to use not only positive, but also negative cosines and use those transitions to the opposite way.
    direct_pca_projection: `bool` (default: `None`)
        Whether to directly project the velocities into PCA space, thus skipping velocity graph.
    retain_scale: `bool` (default: `False`)
        Whether to retain scale from high dimensional space in embedding.
    autoscale: `bool` (default: `True`)
        Whether to scale the embedded velocities by a scalar multiplier,
        which simply ensures that the arrows in the embedding are properly scaled.
    all_comps: `bool` (default: `True`)
        Whether to compute the velocities on all embedding components or just the first two.
    T: `csr_matrix` (default: `None`)
        Allows the user to directly pass a transition matrix.

    Returns
    -------
    Returns or updates `adata` with the attributes
    velocity_basis: `.obsm`
        coordinates of velocity projection on embedding
    """
    adata = data.copy() if copy else data

    if basis is None:
        keys = [key for key in ['pca', 'tsne', 'umap'] if 'X_' + key in adata.obsm.keys()]
        if len(keys) > 0: basis = 'pca' if direct_pca_projection else keys[-1]
        else: raise ValueError('No basis specified')

    if 'X_' + basis not in adata.obsm_keys():
        raise ValueError('You need compute the embedding first.')

    if direct_pca_projection and 'pca' in basis:
        logg.warn(
            'Directly projecting velocities into PCA space is only for exploratory analysis on principal components.\n'
            '         It does not reflect the actual velocity field from high dimensional gene expression space.\n'
            '         To visualize velocities, consider using the velocity graph setting `direct_pca_projection=False`.\n')

    logg.info('computing velocity embedding', r=True)

    if direct_pca_projection and 'pca' in basis:
        V = adata.layers[vkey]
        PCs = adata.varm['PCs'] if all_comps else adata.varm['PCs'][:, :2]

        if vkey + '_genes' in adata.var.keys():
            vgenes = np.array(adata.var[vkey + '_genes'], dtype=bool)
            V = V[:, vgenes]
            PCs = PCs[vgenes]
        nans = np.isnan(V.sum(0))
        if np.any(nans):
            V = V[:, ~nans]
            PCs = PCs[~nans]

        X_emb = adata.obsm['X_' + basis]
        V_emb = (V - V.mean(0)).dot(PCs)

    else:
        X_emb = adata.obsm['X_' + basis] if all_comps else adata.obsm['X_' + basis][:, :2]
        V_emb = np.zeros(X_emb.shape)

        T = transition_matrix(adata, vkey=vkey, scale=scale, self_transitions=self_transitions,
                              use_negative_cosines=use_negative_cosines) if T is None else T
        T.setdiag(0)
        T.eliminate_zeros()

        densify = adata.n_obs < 1e4
        TA = T.A if densify else None

        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            for i in range(adata.n_obs):
                indices = T[i].indices
                dX = X_emb[indices] - X_emb[i, None]  # shape (n_neighbors, 2)
                if not retain_scale: dX /= norm(dX)[:, None]
                dX[np.isnan(dX)] = 0  # zero diff in a steady-state
                probs = TA[i, indices] if densify else T[i].data
                V_emb[i] = probs.dot(dX) - probs.mean() * dX.sum(0)  # probs.sum() / len(indices)

        if retain_scale:
            delta = T.dot(adata.X) - adata.X
            if issparse(delta): delta = delta.A
            cos_proj = (adata.layers[vkey] * delta).sum(1) / norm(delta)
            V_emb *= np.clip(cos_proj[:, None] * 10, 0, 1)

    if autoscale: V_emb /= (3 * quiver_autoscale(X_emb, V_emb))

    vkey += '_' + basis
    adata.obsm[vkey] = V_emb

    logg.info('    finished', time=True, end=' ' if settings.verbosity > 2 else '\n')
    logg.hint(
        'added\n'
        '    \'' + vkey + '\', embedded velocity vectors (adata.obsm)')

    return adata if copy else None

max_stars_repo_name: fidelram/scvelo

max_issues_repo_name: fidelram/scvelo

max_forks_repo_name: fidelram/scvelo

reason: import numpy,from scipy


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

"""
Provides functions to build a large network modeling a carrier topology.
This is the target parameter optimization topology for the mapping algorithm.
The topology is based on WP3 - Service Provider Scenario for Optimization.ppt
by Telecom Italia (UNIFY SVN repo)
Parameter names are also based on the .ppt file.

"""
import logging
import math
import random
import string

import networkx as nx

from generator import NFFG

# Aggregation links (100Gbps) Connecting Distribution nodes to Aggregation Nodes
aggr_link = {'bandwidth': 1000, 'delay': 0.2}

log = logging.getLogger("TopoConstruct")
logging.basicConfig(level=logging.WARN,
                      format='%(levelname)s:%(name)s:%(message)s')

max_portids = {}
def add_port(obj, increment_port_ids=False):
  # WARNING! this function is not thread safe!!
  global max_portids
  if not increment_port_ids:
    port = obj.add_port()
  else:
    if obj in max_portids:
      max_portids[obj] += 1
      port = obj.add_port(id=max_portids[obj])
    else:
      max_portids[obj] = 1
      port = obj.add_port(id=1)
  return port

def getGenForName(prefix):
   number = 0
   while True:
     yield prefix+str(number)
     number += 1

prefixes = {}
def getName(prefix):
  # WARNING! this function is not thread safe!!
  global prefixes
  while True:
    if prefix in prefixes:
      return prefixes[prefix].next()
    else:
      prefixes[prefix] = getGenForName(prefix)
      return prefixes[prefix].next()

def addRedundantPairedConnection(nffg, an0, an1, bn0, bn1, p, linkres):
  """
  Connects A-s to B-s and B-s to A-s with undirected links with linkres.
  """
  nffg.add_undirected_link(add_port(an0, p), add_port(bn0, p), **linkres)
  nffg.add_undirected_link(add_port(an0, p), add_port(bn1, p), **linkres)
  nffg.add_undirected_link(add_port(an1, p), add_port(bn0, p), **linkres)
  nffg.add_undirected_link(add_port(an1, p), add_port(bn1, p), **linkres)

def index_gen():
  while True:
    yield int(math.floor(random.random() * 100000))

def gen_params(l):
  while True:
    yield l[next(index_gen()) % len(l)]

def addRetailOrBusinessPart(nffg, an0, an1, p, popn, BNAS_PE,
                            Cpb, access_bw, part="R"):
  """
  Retail and Business part inside one PoP is structurally the same.
  """
  log.debug("Adding %s part for %s..."%(part, popn))
  # add Distribution Nodes (100Gbps switching capacity)
  dnres = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
           'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  dn0 = None
  dn1 = None
  if BNAS_PE > 0:
    dn0 = nffg.add_infra(id=getName("dn"),
                                **dnres)
    dn1 = nffg.add_infra(id=getName("dn"),
                                **dnres)
    addRedundantPairedConnection(nffg, an0, an1, dn0, dn1, p, aggr_link)

  # add BNAS or PE (10Gbps switching capacity)
  # and connecting SAPs towards Retail Clients (links with BCT bandwidth)
  for i in range(0, BNAS_PE):
    log.debug("Adding switch %s for %s part..."%(i, part))
    bnas_pe_res = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
               'bandwidth': 100, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
    if part == "R":
      bnas_pe = nffg.add_infra(id=getName("bnas"),
                               **bnas_pe_res)
    elif part == "B":
      bnas_pe = nffg.add_infra(id=getName("pe"),
                               **bnas_pe_res)
    else:
      raise Exception("Invalid part identifier given for CarrierTopoBuilder"
                      ".addRetailOrBusinessPart")
    distr_link = {'bandwidth': 100, 'delay': 0.2}

    #add Distribution Links towards Distribution Nodes
    nffg.add_undirected_link(add_port(dn0, p), add_port(bnas_pe, p),
                             **distr_link)
    nffg.add_undirected_link(add_port(dn1, p), add_port(bnas_pe, p),
                             **distr_link)

    # add clients to current BNAS or PE
    log.debug("Connecting %s SAPs to switch %s of %s part."%(Cpb, i, part))
    for j in range(0, Cpb):
      nameid = getName("sap")
      sap = nffg.add_sap(id=nameid, name=nameid)
      access_link = {'bandwidth': access_bw, 'delay': 0.5}
      nffg.add_undirected_link(add_port(bnas_pe, p), add_port(sap, p),
                               **access_link)

def addCassis(nffg, fi0, fi1, p, cluster_id, chassis_id, popn,
              SE, NF_types, SE_cores, SE_mem, SE_storage, CL_bw, CH_links):
  log.debug("Adding Chassis no.%s with %s Servers for Cluster no.%s of %s."%
            (chassis_id,SE,cluster_id,popn))
  fabricext_res = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
                   'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  fe0 = nffg.add_infra(id=getName("fe"), **fabricext_res)
  fe1 = nffg.add_infra(id=getName("fe"), **fabricext_res)
  # add links connecting the Fabric Interconnects and Fabric Extenders
  for i in range(0, CH_links/2):
      nffg.add_undirected_link(add_port(fi0, p), add_port(fe0, p),
                               bandwidth=float(CL_bw)/CH_links, delay=0.2)
      nffg.add_undirected_link(add_port(fi1, p), add_port(fe1, p),
                               bandwidth=float(CL_bw)/CH_links, delay=0.2)

  # add servers and connect them to Fabric Extenders
  for s in range(0, SE):
    server_res = {'cpu': next(gen_params(SE_cores)),
                  'mem': next(gen_params(SE_mem)),
                  'storage': next(gen_params(SE_storage)),
                  'delay': 0.5, 'bandwidth': 1000,
                  'infra_type': NFFG.TYPE_INFRA_EE}
    server = nffg.add_infra(id=getName("host"),
                            **server_res)
    # add supported types
    server.add_supported_type(random.sample(NF_types,
                                     (next(index_gen()) % len(NF_types)) + 1))
    # connect servers to Fabric Extenders with 10Gbps links
    server_link = {'bandwidth': 100, 'delay': 0.2}
    nffg.add_undirected_link(add_port(server, p), add_port(fe0, p), **server_link)
    nffg.add_undirected_link(add_port(server, p), add_port(fe1, p), **server_link)

def addCloudNFVPart(nffg, an0, an1, p, popn, CL, CH, SE, SAN_bw, SAN_storage,
                    NF_types, SE_cores, SE_mem, SE_storage, CL_bw, CH_links):
  log.debug("Adding Cloud/NFV part for %s."%popn)
  dnres = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
           'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  dn0 = nffg.add_infra(id=getName("dn"),
                       **dnres)
  dn1 = nffg.add_infra(id=getName("dn"),
                       **dnres)
  addRedundantPairedConnection(nffg, an0, an1, dn0, dn1, p, aggr_link)

  # add Server Clusters
  for i in range(0, CL):
    log.debug("Adding Cluster no.%s to Could/NFV part of %s"%(i, popn))
    fi_res = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
              'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
    fabric_interconnect0 = nffg.add_infra(id=getName("fi"), **fi_res)
    fabric_interconnect1 = nffg.add_infra(id=getName("fi"), **fi_res)
    addRedundantPairedConnection(nffg, an0, an1, fabric_interconnect0,
                                 fabric_interconnect1, p, aggr_link)

    # NOTE: SAN can't host any VNFs now!!
    # SAN is an Infra with big storage (internal bw should be big: e.g. 1Tbps)
    san_res = {'cpu': 0, 'mem': 0, 'storage': SAN_storage, 'delay': 0.1,
              'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_EE}
    san = nffg.add_infra(id=getName("san"), **san_res)
    # connect SAN to Fabric Interconnects
    nffg.add_undirected_link(add_port(san, p), add_port(fabric_interconnect0, p),
                             bandwidth=SAN_bw, delay=0.1)
    nffg.add_undirected_link(add_port(san, p), add_port(fabric_interconnect1, p),
                             bandwidth=SAN_bw, delay=0.1)
    # add Chassis
    for j in range(0, CH):
      addCassis(nffg, fabric_interconnect0, fabric_interconnect1, p, i, j, popn,
                SE, NF_types, SE_cores, SE_mem, SE_storage, CL_bw, CH_links)

def addPoP(nffg, popcnt, backbonenode0, backbonenode1, p,
           BNAS, RCpb, RCT,
           PE, BCpb, BCT,
           CL, CH, SE, SAN_bw, SAN_storage, NF_types,
           SE_cores, SE_mem, SE_storage, CL_bw, CH_links):
  """
  Create one PoP which consists of three domains:
    - Cloud/NFV services
    - Retail Edge
    - Business Edge
  Backbone nodes where the Aggregation Nodes should be connected.
  BNAS: number of BNAS nodes (~2-10)
  RCpb: number of Retail Clients per BNAS box (~40k)
  RCT: traffic per Retail Clients (0.1-0.2 Mbps)
  PE: number of PE nodes per PoP (~2-8)
  BCpb: number of business clients oer PE box (~4k)
  BCT: traffic per Business Clients (0.1-0.2 Mbps)
  CL: number of clusters in Cloud/NFV part (?)
  CH: number of Chassis per cluster (~8-40)
  SE: number of Servers per chassis (~8)
  SAN_bw: Cluster bandwith to SAN (160Gbps - 1.6Tbps)
  SAN_storage: storage of one SAN (?)
  NF_types: list of supported NF types on the servers
  SE_cores: list of numbers of cores per server (~8-16)
  SE_mem: list of memory capacities per server (~32000MB - 64000MB)
  SE_storage: list of storage capacities per server (~300GB - 1500GB)
  CL_bw: cluster bandwidth to servers per Chassis (~40Gbps - 160Gbps)
  CH_links: number of uplinks per Chassis (~4-16)
    NOTE: Link bw from Fabric Extender to Fabric Interc. equals
    CL_bw/CH_links (~10Gbps).
  """
  popn = "PoP"+str(popcnt)

  log.debug("Adding PoP %s..."%popcnt)
  # add Aggregation Nodes (1Tbps switching capacity)
  anres = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
           'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  an0 = nffg.add_infra(id=getName("an"), **anres)
  an1 = nffg.add_infra(id=getName("an"), **anres)

  # add uplinks to the Backbone
  nffg.add_undirected_link(add_port(an0, p), add_port(backbonenode0, p),
                           bandwidth=1000, delay=0.1)
  nffg.add_undirected_link(add_port(an1, p), add_port(backbonenode1, p),
                           bandwidth=1000, delay=0.1)

  addRetailOrBusinessPart(nffg, an0, an1, p, popn, BNAS, RCpb, RCT)
  addRetailOrBusinessPart(nffg, an0, an1, p, popn, PE, BCpb, BCT,
                          part="B")

  if CL > 0:
      addCloudNFVPart(nffg, an0, an1, p, popn, CL, CH, SE, SAN_bw, SAN_storage,
                      NF_types, SE_cores, SE_mem, SE_storage, CL_bw, CH_links)

  return

def getCarrierTopo(params, increment_port_ids=False):
  """
  Construct the core network and add PoPs with their parameters.
  params is a list of dictionaries with PoP data:
    'Retail': (BNAS, RCpb, RCT)
    'Business': (PE, BCpb, BCT)
    'CloudNFV': (CL,CH,SE,SAN_bw,SAN_sto,NF_types,SE_cores,SE_mem,SE_sto,
                 CL_bw, CH_links)
  WARNING: using this function with increment_port_ids=True this function is not
  thread safe, because it uses global variable then!
  """
  # This initializes the random generator always to the same value, so the
  # returned index sequence, and thus the network parameters will be generated
  # always the same (we want a fixed network environment)
  # The generated identifiers are still different between genereations, but
  # those does not influence the mapping process
  random.seed(0)
  popcnt = 0
  nffg = NFFG(id="CarrierTopo")
  p = increment_port_ids
  backbone_res =  {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
                   'bandwidth': 10000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  bn0 = nffg.add_infra(id=getName("bn"), **backbone_res)
  bn1 = nffg.add_infra(id=getName("bn"), **backbone_res)
  bn2 = nffg.add_infra(id=getName("bn"), **backbone_res)
  bn3 = nffg.add_infra(id=getName("bn"), **backbone_res)

  nffg.add_undirected_link(add_port(bn0, p), add_port(bn1, p), bandwidth=1000,
                           delay=10)
  nffg.add_undirected_link(add_port(bn1, p), add_port(bn2, p), bandwidth=1000,
                           delay=10)
  nffg.add_undirected_link(add_port(bn2, p), add_port(bn3, p), bandwidth=1000,
                           delay=10)
  nffg.add_undirected_link(add_port(bn3, p), add_port(bn0, p), bandwidth=1000,
                           delay=10)
  backbones = (bn0, bn1, bn2, bn3)
  bnlen = len(backbones)
  for popdata in params:
    tmp = []
    tmp.extend(popdata['Retail'])
    tmp.extend(popdata['Business'])
    tmp.extend(popdata['CloudNFV'])
    addPoP(nffg, popcnt, backbones[popcnt%bnlen], backbones[(popcnt+1)%bnlen],
           p, *tmp)
    popcnt += 1
  """
  #                      BNAS,RCpb,  RCT, PE,BCpb, BCT, CL,CH,SE, SAN_bw,
  addPoP(nffg, bn2, bn3, 2,   10000, 0.2, 2, 4000, 0.2, 2, 8, 8,  160000,
       # SAN_sto,NF_types,  SE_cores,  SE_mem,        SE_sto,  CL_bw, CH_links
         100000, ['A','B'], [8,12,16], [32000,64000], [150],   40000, 4)
  #                      BNAS, RCpb,  RCT, PE,BCpb, BCT, CL,CH, SE, SAN_bw,
  addPoP(nffg, bn1, bn2, 10,   40000, 0.2, 8, 4000, 0.2, 4, 40, 8,  160000,
       # SAN_sto,NF_types,             SE_cores,  SE_mem,        SE_sto,
         100000, ['A','B','C','D','E'],[8,12,16], [32000,64000], [150,200],
       # CL_bw, CH_links
         80000, 8)
  """
  log.debug("Carrier topology construction finished!")
  return nffg

def getMediumTopo():
  """
  Constructs a medium sized topology for worst case presentation, if the bigger
  would take too long to finish. Its size is around 12 460 nodes
  (~4500 with SAP cutting)
  """
  topoparams = []
  # params of one PoP
  # 'Retail': (BNAS, RCpb, RCT)
  # 'Business': (PE, BCpb, BCT)
  # 'CloudNFV': (CL,CH,SE,SAN_bw,SAN_sto,NF_types,SE_cores,SE_mem,SE_sto,
  #              CL_bw, CH_links)
  topoparams.append({'Retail': (6, 250, 0.2), 'Business': (6, 200, 0.2),
                     'CloudNFV': (4, 40, 8,  160000, 100000, ['A','B','C'],
                                 [4,8,16],  [32000, 64000], [100,150], 40000, 4)})
  topoparams.append({'Retail': (6, 100, 0.2), 'Business': (4, 200, 0.2),
                     'CloudNFV': (4, 20, 8,  160000, 100000, ['A','B'],
                                  [8,12,16], [32000,64000], [150], 40000, 4)})
  topoparams.append({'Retail': (3, 400, 0.2), 'Business': (8, 100, 0.2),
                    'CloudNFV': (4, 30, 8,  160000, 100000, ['B', 'C'],
                                [4,8,12,16], [32000,64000], [200], 40000, 4)})
  topoparams.append({'Retail': (10, 100, 0.2), 'Business': (8, 150, 0.2),
                    'CloudNFV': (4, 40, 8,  160000, 100000, ['B', 'C'],
                                [4,8,12,16], [32000,64000], [200], 40000, 4)})
  return getCarrierTopo(topoparams), topoparams

def getSmallTopo():
  """
  Constructs a small topology which is structurally similar to carrier topology,
  but could be executed fast enough for testing.
  """
  topoparams = []
  # params of one PoP
  # 'Retail': (BNAS, RCpb, RCT)
  # 'Business': (PE, BCpb, BCT)
  # 'CloudNFV': (CL,CH,SE,SAN_bw,SAN_sto,NF_types,SE_cores,SE_mem,SE_sto,
  #              CL_bw, CH_links)
  topoparams.append({'Retail': (2, 250, 0.2), 'Business': (2, 100, 0.2),
                     'CloudNFV': (2, 4, 8,  160000, 100000, ['A','B','C'],
                                  [4,8,16],  [32000], [100,150],   40000, 4)})
  topoparams.append({'Retail': (2, 250, 0.2), 'Business': (2, 150, 0.2),
                     'CloudNFV': (2, 2, 8,  160000, 100000, ['A','B'],
                                  [8,12,16], [32000,64000], [150], 40000, 4)})
  # topoparams.append({'Retail': (2, 4000, 0.2), 'Business': (8, 2000, 0.2),
  #                    'CloudNFV': (2, 40, 8,  160000, 100000, ['B', 'C'],
  # [4,8,12,16], [32000,64000], [200], 40000, 4)})
  return getCarrierTopo(topoparams), topoparams

def getMicroTopo():
  topoparams = []
  topoparams.append({'Retail': (2, 50, 0.2), 'Business': (2, 30, 0.2),
                     'CloudNFV': (2, 2, 4,  160000, 100000, ['A','B'],
                                  [8,12,16], [32000,64000], [150], 40000, 4)})
  topoparams.append({'Retail': (2, 50, 0.2), 'Business': (2, 30, 0.2),
                     'CloudNFV': (2, 2, 4,  160000, 100000, ['A','B', 'C'],
                                  [8,12,16], [32000,64000], [150], 40000, 4)})
  return getCarrierTopo(topoparams), topoparams

def getNanoTopo():
  topoparams = []
  topoparams.append({'Retail': (1, 2, 10), 'Business': (1, 2, 10),
                     'CloudNFV': (1, 2, 4,  1000, 100000, ['A', 'B', 'C'],
                                  [8,12,16], [32000,64000], [150], 4000, 4)})
  """
  topoparams.append({'Retail': (2, 5, 0.2), 'Business': (2, 5, 0.2),
                     'CloudNFV': (1, 2, 4,  160000, 100000, ['A','B', 'C'],
                                  [8,12,16], [32000,64000], [150], 4000, 4)})
  """
  return getCarrierTopo(topoparams, increment_port_ids=True), topoparams

def getPicoTopo():
  """
  Not carrier style topo. Few nodes with big resources.
  """
  random.seed(0)
  nffg = NFFG(id="SmallExampleTopo")
  switch = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
            'bandwidth': 100, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  sw = nffg.add_infra(id = getName("sw"), **switch)
  infra = {'cpu': 90, 'mem': 32000, 'storage': 200, 'delay': 1.0,
           'bandwidth': 1000, 'infra_type': NFFG.TYPE_INFRA_EE}
  linkres = {'bandwidth': 100, 'delay': 0.5}

  inf1 = nffg.add_infra(id = getName("infra"), **infra)
  inf0 = inf1
  inf1.add_supported_type(list(string.ascii_uppercase)[:10])
  for i in range(0,4):
    if i == 3:
      inf2 = inf0
    else:
      inf2 = nffg.add_infra(id = getName("infra"), **infra)
      inf2.add_supported_type(list(string.ascii_uppercase)[:10])
    nameid = getName("sap")
    sap = nffg.add_sap(id = nameid, name = nameid)
    # add links

    nffg.add_undirected_link(sw.add_port(), inf2.add_port(), **linkres)
    nffg.add_undirected_link(inf1.add_port(), inf2.add_port(), **linkres)
    nffg.add_undirected_link(inf2.add_port(), sap.add_port(id=1), **linkres)
    inf1 = inf2

  return nffg

def getSNDlib_dfn_gwin(gwin_path = "dfn-gwin.gml", save_to_file=False,
                       gen_sap_names=False,
                       abc_nf_type_num=10, edge_computing=False):
  """
  Topology taken from SNDlib, dfn-gwin.
  """
  random.seed(0)
  gwin = nx.read_gml(gwin_path)
  nffg = NFFG(id="dfn-gwin")
  nf_types = list(string.ascii_uppercase)[:abc_nf_type_num]
  switch = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
            'bandwidth': 40000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  infrares = {'cpu': 400, 'mem': 320000, 'storage': 1500, 'delay': 1.0,
           'bandwidth': 40000, 'infra_type': NFFG.TYPE_INFRA_EE}
  corelinkres = {'bandwidth': 10000, 'delay': 1.0}
  aggrlinkres = {'bandwidth': 1000, 'delay': 5.0}
  acclinkres = {'bandwidth': 100, 'delay': 1.0}
  gwinnodes = []
  for n in  gwin.nodes_iter():
    gwinnodes.append(n.rstrip('.'))
  # get topology from dfn-gwin
  for n in gwinnodes:
    nffg.add_infra(id=n, **switch)
  for i,j in gwin.edges_iter():
    nffg.add_undirected_link(nffg.network.node[i.rstrip('.')].add_port(),
                             nffg.network.node[j.rstrip('.')].add_port(),
                             **corelinkres)

  if not edge_computing:
    # add cloud nodes to 6 random nodes.
    nodeset1 = random.sample(gwinnodes, 3)
    nodeset1.extend(random.sample(gwinnodes, 3))
    for n in nodeset1:
      infra = nffg.add_infra(id=getName(n+"Host"), **infrares)
      infra.add_supported_type(random.sample(nf_types, 6))
      nffg.add_undirected_link(nffg.network.node[n].add_port(), infra.add_port(),
                               **corelinkres)

  nodeset2 = random.sample(gwinnodes, 3)
  nodeset2.extend(random.sample(gwinnodes, 3))
  # add access switches to 6 random nodes
  for n in nodeset2:
    sw = nffg.add_infra(id=getName(n+"Sw"), **switch)
    nffg.add_undirected_link(nffg.network.node[n].add_port(), sw.add_port(),
                             **aggrlinkres)
    for i in xrange(0,random.randint(3,4)):
      if gen_sap_names:
        nameid = getName("sap")
      else:
        nameid = getName(n+"SAP")
      sap = nffg.add_sap(id=nameid, name=nameid)
      nffg.add_undirected_link(sap.add_port(id=1), sw.add_port(), **acclinkres)

  if edge_computing:
    # add hosts to the 6 access switches available
    for n in filter(lambda i: "Sw" in i.id, [i for i in nffg.infras]):
      infra = nffg.add_infra(id=getName(n.id + "Host"), **infrares)
      infra.add_supported_type(random.sample(nf_types, 6))
      nffg.add_undirected_link(nffg.network.node[n.id].add_port(),
                               infra.add_port(),
                               **corelinkres)

  # save it to file
  if save_to_file:
    augmented_gwin = nx.MultiDiGraph()
    augmented_gwin.add_nodes_from(nffg.network.nodes_iter())
    augmented_gwin.add_edges_from(nffg.network.edges_iter())
    nx.write_gml(augmented_gwin, "augmented-dfn-gwin.gml")

  return nffg

def getFatTreeTopo(abc_nf_type_num=10, save_to_file=False):
  """
  Constructs a data center fat tree topology, with similar values as
  SNDLib gwin.
  :param gen_sap_names:
  :param abc_nf_type_num:
  :param save_to_file:
  :return:
  """
  random.seed(1)
  nf_types = list(string.ascii_uppercase)[:abc_nf_type_num]
  nffg = NFFG(id="FatTree")
  switch = {'cpu': 0, 'mem': 0, 'storage': 0, 'delay': 0.5,
            'bandwidth': 40000, 'infra_type': NFFG.TYPE_INFRA_SDN_SW}
  infrares = {'cpu': 400, 'mem': 320000, 'storage': 1500, 'delay': 1.0,
              'bandwidth': 40000, 'infra_type': NFFG.TYPE_INFRA_EE}
  corelinkres = {'bandwidth': 10000, 'delay': 1.0}
  aggrlinkres = {'bandwidth': 1000, 'delay': 5.0}
  acclinkres = {'bandwidth': 100, 'delay': 1.0}

  # add core switches
  core_sw = []
  for i in nx.complete_graph(4):
    core_sw.append(nffg.add_infra(id='CoreSw'+str(i), **switch))
  for i,j in nx.complete_graph(4).edges():
    nffg.add_undirected_link(core_sw[i].add_port(id=getName('p'+core_sw[i].id)),
                             core_sw[j].add_port(id=getName('p'+core_sw[j].id)),
                             **corelinkres)

  # add aggregation switches
  aggr_sw = []
  for i in xrange(0,8):
    aggr_sw.append(nffg.add_infra(id='AggrSw'+str(i), **switch))
  for sw_c in core_sw:
    for sw_a in random.sample(aggr_sw, 4):
      nffg.add_undirected_link(sw_c.add_port(id=getName('p'+sw_c.id)),
                               sw_a.add_port(id=getName('p'+sw_a.id)),
                               **aggrlinkres)

  # add sap nodes with one connecting switch each
  for i in xrange(0,19):
    nameid = getName('SAP')
    sap = nffg.add_sap(id=nameid, name=nameid)
    access_sw = nffg.add_infra(id='AccessSw'+str(i), **switch)
    # connect the SAP to its Switch to eliminate SAP link bottleneck.
    # WARNING: in some NFFG environments SAPs should only have one port, in
    # others this port must have ID 1.
    nffg.add_undirected_link(sap.add_port(id=1),
                             access_sw.add_port(id=getName('p'+access_sw.id)),
                             **corelinkres)
    for sw_a in random.sample(aggr_sw, 6):
      nffg.add_undirected_link(access_sw.add_port(id=getName('p'+access_sw.id)),
                               sw_a.add_port(id=getName('p'+sw_a.id)),
                                             **acclinkres)

  # add hosts: 2 connected to core and 4 connected to aggregation switches
  for number, sw_list in zip((2, 4), (core_sw, aggr_sw)):
    for i, sw in zip(xrange(0,number), random.sample(sw_list, number)):
      infra = nffg.add_infra(id=getName('Host'), **infrares)
      infra.add_supported_type(random.sample(nf_types, 6))
      nffg.add_undirected_link(sw.add_port(id=getName('p'+sw.id)),
                               infra.add_port(id=getName('p'+infra.id)),
                               **corelinkres)

  if save_to_file:
    fat_tree = nx.MultiDiGraph()
    fat_tree.add_nodes_from(nffg.network.nodes_iter())
    fat_tree.add_edges_from(nffg.network.edges_iter())
    nx.write_gml(fat_tree, "fat-tree-data-center.gml")

  return nffg

if __name__ == '__main__':
  topoparams = []
  # params of one PoP
  # 'Retail': (BNAS, RCpb, RCT)
  # 'Business': (PE, BCpb, BCT)
  # 'CloudNFV': (CL,CH,SE,SAN_bw,SAN_sto,NF_types,SE_cores,SE_mem,SE_sto,
  #              CL_bw, CH_links)
  # print getSNDlib_dfn_gwin().dump() NOT WORKING.
  topoparams.append({'Retail': (2, 10000, 0.2), 'Business': (2, 8000, 0.2),
                     'CloudNFV': (2, 8, 8,  160000, 100000, ['A','B','C'],
                                  [4,8,16],  [32000], [100,150],   40000, 4)})
  topoparams.append({'Retail': (2, 10000, 0.2), 'Business': (4, 4000, 0.2),
                     'CloudNFV': (2, 8, 8,  160000, 100000, ['A','B'],
                                  [8,12,16], [32000,64000], [150], 40000, 4)})
  # topoparams.append({'Retail': (2, 20000, 0.2), 'Business': (8, 4000, 0.2),
  #                    'CloudNFV': (2, 40, 8,  160000, 100000, ['B', 'C'],
  #                                 [4,8,12,16], [32000,64000], [200], 40000, 4)})
  # topo = getCarrierTopo(topoparams)
  # print topo.dump()

  # with open("augmented-dfn-gwin.nffg", "w") as f:
  #   f.write(getSNDlib_dfn_gwin(abc_nf_type_num=10, gen_sap_names=False,
  #                              save_to_file=True, edge_computing=True).dump())

  with open("fat-tree-data-center.nffg", "w") as f:
    f.write(getFatTreeTopo(save_to_file=True).dump())

max_stars_repo_name: hsnlab/mapping

reason: import networkx


text:
# 参考：https://www.cnblogs.com/xxhbdk/p/9177423.html
import numpy as np
import matplotlib.pyplot as plt

# 建立“蚂蚁”类
class Ant(object):
    def __init__(self, path):
        self.path = path  # 蚂蚁当前迭代整体路径
        self.length = self.calc_length(path)  # 蚂蚁当前迭代整体路径长度

    def calc_length(self, path_):  # path=[A, B, C, D, A]注意路径闭环
        length_ = 0
        for i in range(len(path_) - 1):
            delta = (path_[i].x - path_[i + 1].x, path_[i].y - path_[i + 1].y)
            length_ += np.linalg.norm(delta)
        return length_

    @staticmethod
    def calc_len(A, B):  # 静态方法，计算城市A与城市B之间的距离
        return np.linalg.norm((A.x - B.x, A.y - B.y))

# 建立“城市”类
class City(object):
    def __init__(self, x, y):
        self.x = x
        self.y = y

# 建立“路径”类
class Path(object):
    def __init__(self, A):  # A为起始城市
        self.path = [A, A]

    def add_path(self, B):  # 追加路径信息，方便计算整体路径长度
        self.path.append(B)
        self.path[-1], self.path[-2] = self.path[-2], self.path[-1]

# 构建“蚁群算法”的主体
class ACO(object):
    def __init__(self, ant_num=50, maxIter=50, alpha=1, beta=5, rho=0.1, Q=1):
        self.ants_num = ant_num  # 蚂蚁个数
        self.maxIter = maxIter  # 蚁群最大迭代次数
        self.alpha = alpha  # 信息启发式因子
        self.beta = beta  # 期望启发式因子
        self.rho = rho  # 信息素挥发速度
        self.Q = Q  # 信息素强度
        ###########################
        self.deal_data('coordinates.dat')  # 提取所有城市的坐标信息
        ###########################
        self.path_seed = np.zeros(self.ants_num).astype(int)  # 记录一次迭代过程中每个蚂蚁的初始城市下标
        self.ants_info = np.zeros((self.maxIter, self.ants_num))  # 记录每次迭代后所有蚂蚁的路径长度信息
        self.best_path = np.zeros(self.maxIter)  # 记录每次迭代后整个蚁群的“历史”最短路径长度
        ###########################
        self.solve()  # 完成算法的迭代更新
        self.display()  # 数据可视化展示

    def deal_data(self, filename):
        with open(filename, 'rt') as f:
            temp_list = list(line.split() for line in f)  # 临时存储提取出来的坐标信息
        self.cities_num = len(temp_list)  # 1. 获取城市个数
        self.cities = list(City(float(item[0]), float(item[1])) for item in temp_list)  # 2. 构建城市列表
        self.city_dist_mat = np.zeros((self.cities_num, self.cities_num))  # 3. 构建城市距离矩阵
        for i in range(self.cities_num):
            A = self.cities[i]
            for j in range(i, self.cities_num):
                B = self.cities[j]
                self.city_dist_mat[i][j] = self.city_dist_mat[j][i] = Ant.calc_len(A, B)
        self.phero_mat = np.ones((self.cities_num, self.cities_num))  # 4. 初始化信息素矩阵
        # self.phero_upper_bound = self.phero_mat.max() * 1.2                              ###信息素浓度上限
        self.eta_mat = 1 / (self.city_dist_mat + np.diag([np.inf] * self.cities_num))  # 5. 初始化启发函数矩阵

    def solve(self):
        iterNum = 0  # 当前迭代次数
        while iterNum < self.maxIter:
            print(iterNum)
            self.random_seed()  # 使整个蚁群产生随机的起始点
            delta_phero_mat = np.zeros((self.cities_num, self.cities_num))  # 初始化每次迭代后信息素矩阵的增量
            ##########################################################################
            for i in range(self.ants_num):
                city_index1 = self.path_seed[i]  # 每只蚂蚁访问的第一个城市下标
                ant_path = Path(self.cities[city_index1])  # 记录每只蚂蚁访问过的城市
                tabu = [city_index1]  # 记录每只蚂蚁访问过的城市下标，禁忌城市下标列表
                non_tabu = list(set(range(self.cities_num)) - set(tabu))
                for j in range(self.cities_num - 1):  # 对余下的城市进行访问
                    up_proba = np.zeros(self.cities_num - len(tabu))  # 初始化状态迁移概率的分子
                    for k in range(self.cities_num - len(tabu)):
                        up_proba[k] = np.power(self.phero_mat[city_index1][non_tabu[k]], self.alpha) * \
                                      np.power(self.eta_mat[city_index1][non_tabu[k]], self.beta)
                    proba = up_proba / sum(up_proba)  # 每条可能子路径上的状态迁移概率
                    while True:  # 提取出下一个城市的下标
                        random_num = np.random.rand()
                        index_need = np.where(proba > random_num)[0]
                        if len(index_need) > 0:
                            city_index2 = non_tabu[index_need[0]]
                            break
                    ant_path.add_path(self.cities[city_index2])
                    tabu.append(city_index2)
                    non_tabu = list(set(range(self.cities_num)) - set(tabu))
                    city_index1 = city_index2
                self.ants_info[iterNum][i] = Ant(ant_path.path).length
                if iterNum == 0 and i == 0:  # 完成对最佳路径城市的记录
                    self.best_cities = ant_path.path
                else:
                    if self.ants_info[iterNum][i] < Ant(self.best_cities).length: self.best_cities = ant_path.path
                tabu.append(tabu[0])  # 每次迭代完成后，使禁忌城市下标列表形成完整闭环
                for l in range(self.cities_num):
                    delta_phero_mat[tabu[l]][tabu[l + 1]] += self.Q / self.ants_info[iterNum][i]

            self.best_path[iterNum] = Ant(self.best_cities).length

            self.update_phero_mat(delta_phero_mat)  # 更新信息素矩阵
            iterNum += 1

    def update_phero_mat(self, delta):
        self.phero_mat = (1 - self.rho) * self.phero_mat + delta
        # self.phero_mat = np.where(self.phero_mat > self.phero_upper_bound, self.phero_upper_bound, self.phero_mat) # 判断是否超过浓度上限

    def random_seed(self):  # 产生随机的起始点下表，尽量保证所有蚂蚁的起始点不同
        if self.ants_num <= self.cities_num:  # 蚂蚁数 <= 城市数
            self.path_seed[:] = np.random.permutation(range(self.cities_num))[:self.ants_num]
        else:  # 蚂蚁数 > 城市数
            self.path_seed[:self.cities_num] = np.random.permutation(range(self.cities_num))
            temp_index = self.cities_num
            while temp_index + self.cities_num <= self.ants_num:
                self.path_seed[temp_index:temp_index + self.cities_num] = np.random.permutation(range(self.cities_num))
                temp_index += self.cities_num
            temp_left = self.ants_num % self.cities_num
            if temp_left != 0:
                self.path_seed[temp_index:] = np.random.permutation(range(self.cities_num))[:temp_left]

    def display(self):  # 数据可视化展示
        plt.figure(figsize=(6, 10))
        plt.subplot(211)
        plt.plot(self.ants_info, 'g.')
        plt.plot(self.best_path, 'r-', label='history_best')
        plt.xlabel('Iteration')
        plt.ylabel('length')
        plt.legend()
        plt.subplot(212)
        plt.plot(list(city.x for city in self.best_cities), list(city.y for city in self.best_cities), 'g-')
        plt.plot(list(city.x for city in self.best_cities), list(city.y for city in self.best_cities), 'r.')
        plt.xlabel('x')
        plt.ylabel('y')
        plt.savefig('ACO.png', dpi=500)
        plt.show()
        plt.close()

if __name__ == '__main__':
    ACO()


text:
import cv2
import torch
import numpy as np
import BboxToolkit as bt
PI = np.pi

def regular_theta(theta, mode='180', start=-PI/2):
    assert mode in ['360', '180']
    cycle = 2 * PI if mode == '360' else PI

    theta = theta - start
    theta = theta % cycle
    return theta + start

def mintheta_obb(rbox):
    x, y, w, h, theta, cls = np.split(rbox, 6, axis=-1)
    theta1 = regular_theta(theta)
    theta2 = regular_theta(theta + PI / 2)
    abs_theta1 = np.abs(theta1)
    abs_theta2 = np.abs(theta2)

    w_regular = np.where(abs_theta1 < abs_theta2, w, h)
    h_regular = np.where(abs_theta1 < abs_theta2, h, w)
    theta_regular = np.where(abs_theta1 < abs_theta2, theta1, theta2)
    # theta_regular = np.where(theta_regular == -PI / 4, PI / 4, theta_regular)
    return np.concatenate([x, y, w_regular, h_regular, theta_regular, cls], axis=-1)

# def regular_obb(obboxes):
#     x, y, w, h, theta = obboxes.unbind(dim=-1)
#     w_regular = torch.where(w > h, w, h)
#     h_regular = torch.where(w > h, h, w)
#     theta_regular = torch.where(w > h, theta, theta+PI/2)
#     theta_regular = regular_theta(theta_regular)
#     return torch.stack([x, y, w_regular, h_regular, theta_regular], dim=-1)

# def mintheta_obb(obboxes):
#     ''''
#     norm the theta to (-45, 45), and its w, h need to be exchanged
#     '''

#     x, y, w, h, theta = obboxes.unbind(dim=-1)
#     theta1 = regular_theta(theta)
#     theta2 = regular_theta(theta + pi/2)
#     abs_theta1 = torch.abs(theta1)
#     abs_theta2 = torch.abs(theta2)

#     w_regular = torch.where(abs_theta1 < abs_theta2, w, h)
#     h_regular = torch.where(abs_theta1 < abs_theta2, h, w)
#     theta_regular = torch.where(abs_theta1 < abs_theta2, theta1, theta2)

#     obboxes = torch.stack([x, y, w_regular, h_regular, theta_regular], dim=-1)

#     return obboxes
def poly2obb_np(rbox):
    if rbox.shape[-1] == 9:
        res = np.empty((*rbox.shape[:-1], 6))
        res[..., 5] = rbox[..., 8]
        rbox = rbox[..., :8].reshape(1, -1, 2).astype(np.float32)
    elif rbox.shape[-1] == 8:
        res = np.empty([*rbox.shape[:-1], 5])
        rbox = rbox.reshape(1, -1, 2).astype(np.float32)
    else:
        raise NotImplementedError
    (x, y), (w, h), angle = cv2.minAreaRect(rbox)
    if w >= h:
        angle = -angle
    else:
        w, h = h, w
        angle = -90 - angle
    theta = angle / 180 * PI
    res[..., 0] = x
    res[..., 1] = y
    res[..., 2] = w
    res[..., 3] = h
    res[..., 4] = theta

    return res[None, :]

def poly2obb(polys):
    polys_np = polys.detach().cpu().numpy()

    order = polys_np.shape[:-1]
    num_points = polys_np.shape[-1] // 2
    polys_np = polys_np.reshape(-1, num_points, 2)
    polys_np = polys_np.astype(np.float32)

    obboxes = []
    for poly in polys_np:
        (x, y), (w, h), angle = cv2.minAreaRect(poly)
        if w >= h:
            angle = -angle
        else:
            w, h = h, w
            angle = -90 - angle
        theta = angle / 180 * PI
        obboxes.append([x, y, w, h, theta])

    if not obboxes:
        obboxes = np.zeros((0, 5))
    else:
        obboxes = np.array(obboxes)

    obboxes = obboxes.reshape(*order, 5)
    return polys.new_tensor(obboxes)

# def rectpoly2obb(polys):
#     theta = torch.atan2(-(polys[..., 3] - polys[..., 1]),
#                         polys[..., 2] - polys[..., 0])
#     Cos, Sin = torch.cos(theta), torch.sin(theta)
#     Matrix = torch.stack([Cos, -Sin, Sin, Cos], dim=-1)
#     Matrix = Matrix.view(*Matrix.shape[:-1], 2, 2)

#     x = polys[..., 0::2].mean(-1)
#     y = polys[..., 1::2].mean(-1)
#     center = torch.stack([x, y], dim=-1).unsqueeze(-2)
#     center_polys = polys.view(*polys.shape[:-1], 4, 2) - center
#     rotate_polys = torch.matmul(center_polys, Matrix.transpose(-1, -2))

#     xmin, _ = torch.min(rotate_polys[..., :, 0], dim=-1)
#     xmax, _ = torch.max(rotate_polys[..., :, 0], dim=-1)
#     ymin, _ = torch.min(rotate_polys[..., :, 1], dim=-1)
#     ymax, _ = torch.max(rotate_polys[..., :, 1], dim=-1)
#     w = xmax - xmin
#     h = ymax - ymin

#     obboxes = torch.stack([x, y, w, h, theta], dim=-1)
#     return regular_obb(obboxes)

def poly2hbb(polys):
    polys = polys.view(*polys.shape[:-1], polys.size(-1)//2, 2)
    lt_point = torch.min(polys, dim=-2)[0]
    rb_point = torch.max(polys, dim=-2)[0]
    return torch.cat([lt_point, rb_point], dim=-1)

def obb2poly(obboxes):
    center, w, h, theta = torch.split(obboxes, [2, 1, 1, 1], dim=-1)
    Cos, Sin = torch.cos(theta), torch.sin(theta)

    vector1 = torch.cat(
        [w/2 * Cos, -w/2 * Sin], dim=-1)
    vector2 = torch.cat(
        [-h/2 * Sin, -h/2 * Cos], dim=-1)

    point1 = center + vector1 + vector2
    point2 = center + vector1 - vector2
    point3 = center - vector1 - vector2
    point4 = center - vector1 + vector2
    return torch.cat(
        [point1, point2, point3, point4], dim=-1)

def obb2hbb(obboxes, mode="xyxy"):
    center, w, h, theta = torch.split(obboxes, [2, 1, 1, 1], dim=-1)
    Cos, Sin = torch.cos(theta), torch.sin(theta)
    x_bias = torch.abs(w/2 * Cos) + torch.abs(h/2 * Sin)
    y_bias = torch.abs(w/2 * Sin) + torch.abs(h/2 * Cos)
    bias = torch.cat([x_bias, y_bias], dim=-1)
    if mode == "xyxy":
        obboxes = torch.cat([center-bias, center+bias], dim=-1)
    elif mode == "cxcywh":
        obboxes = torch.cat([center, 2 * bias.abs()])
    return obboxes

def hbb2poly(hbboxes):
    l, t, r, b = hbboxes.unbind(-1)
    return torch.stack([l, t, r, t, r, b, l ,b], dim=-1)

def hbb2obb(hbboxes):
    x = (hbboxes[..., 0] + hbboxes[..., 2]) * 0.5
    y = (hbboxes[..., 1] + hbboxes[..., 3]) * 0.5
    w = hbboxes[..., 2] - hbboxes[..., 0]
    h = hbboxes[..., 3] - hbboxes[..., 1]
    theta = x.new_zeros(*x.shape)

    obboxes1 = torch.stack([x, y, w, h, theta], dim=-1)
    obboxes2 = torch.stack([x, y, h, w, theta-PI/2], dim=-1)
    obboxes = torch.where((w >= h)[..., None], obboxes1, obboxes2)
    return obboxes

_type_func_map = {
    ('poly', 'obb'): poly2obb,
    ('poly', 'hbb'): poly2hbb,
    ('obb', 'poly'): obb2poly,
    ('obb', 'hbb'): obb2hbb,
    ('hbb', 'poly'): hbb2poly,
    ('hbb', 'obb'): hbb2obb
}

def get_bbox_type(bboxes, with_score=False):
    dim = bboxes.size(-1)
    if with_score:
        dim -= 1

    if dim == 4:
        return 'hbb'
    if dim == 5:
        return 'obb'
    if dim  == 8:
        return 'poly'
    return 'notype'

def bbox2type(bboxes, to_type):
    assert to_type in ['hbb', 'obb', 'poly']

    ori_type = get_bbox_type(bboxes)
    if ori_type == 'notype':
        raise ValueError('Not a bbox type')
    if ori_type == to_type:
        return bboxes
    trans_func = _type_func_map[(ori_type, to_type)]
    return trans_func(bboxes)