text:
import keras,os
from keras.models import Model
from keras.layers import Dense, Conv2D, MaxPool2D, Flatten, Input
from keras.preprocessing.image import ImageDataGenerator
from keras.optimizers import Adam
import numpy as np
import h5py
from keras.applications.vgg16 import VGG16

def getVGGModel():
    input_tensor = Input(shape=(224,224,3))
    vgg_model = VGG16(weights='imagenet', include_top=False, input_tensor=input_tensor)
    x = vgg_model.get_layer('block5_pool').output

    x = Flatten()(x)
    x = Dense(4096, activation='relu')(x)
    x = Dense(4096, activation='relu')(x)
    x = Dense(5, activation='softmax')(x)
    cus_model = Model(input=vgg_model.input, output=x)
    return cus_model

model = getVGGModel()

for layer in model.layers[:19]:
    layer.trainable = False

opt = Adam(lr=0.00001, beta_1=0.9, beta_2=0.999)
model.compile(loss='categorical_crossentropy',
              optimizer=opt,
              metrics=['accuracy'])
print(model.summary())

datagen = ImageDataGenerator()

train_it = datagen.flow_from_directory('Dataset/Train/', target_size=(224,224), class_mode='categorical', batch_size=16, shuffle=True)
val_it = datagen.flow_from_directory('Dataset/Validation/', target_size=(224,224), class_mode='categorical', batch_size=16, shuffle=True)

model.fit_generator(train_it,
                    steps_per_epoch=325,
                    validation_data=val_it,
                    validation_steps=82,
                    epochs=5)

max_stars_repo_name: zhaozhang/DesignSafe

max_issues_repo_name: zhaozhang/DesignSafe

max_forks_repo_name: zhaozhang/DesignSafe


text:
import calendar
import numpy as np
import pandas as pd
import re
import scipy.interpolate as interp
import urllib
import warnings
from datetime import datetime as dt,timedelta

from ..plot import Plot

from ..utils import *

try:
    import cartopy.feature as cfeature
    from cartopy import crs as ccrs
    from cartopy.mpl.gridliner import LONGITUDE_FORMATTER, LATITUDE_FORMATTER
except:
    warnings.warn("Warning: Cartopy is not installed in your python environment. Plotting functions will not work.")

try:
    import matplotlib as mlib
    import matplotlib.lines as mlines
    import matplotlib.patheffects as path_effects
    import matplotlib.pyplot as plt
    import matplotlib.colors as col
    import matplotlib.ticker as mticker
    import matplotlib.patches as mpatches

except:
    warnings.warn("Warning: Matplotlib is not installed in your python environment. Plotting functions will not work.")

class RainPlot(Plot):

    def __init__(self):

        self.use_credit = True

    def plot_storm(self,storm,grid,levels,cmap,domain="dynamic",plot_all_dots=False,ax=None,track_labels=False,save_path=None,prop={},map_prop={}):

        r"""
        Creates a plot of a single storm track.

        Parameters
        ----------
        storm : str, tuple or dict
            Requested storm. Can be either string of storm ID (e.g., "AL052019"), tuple with storm name and year (e.g., ("Matthew",2016)), or a dict entry.
        domain : str
            Domain for the plot. Can be one of the following:
            "dynamic" - default. Dynamically focuses the domain using the storm track(s) plotted.
            "north_atlantic" - North Atlantic Ocean basin
            "pacific" - East/Central Pacific Ocean basin
            "lonW/lonE/latS/latN" - Custom plot domain
        plot_all_dots : bool
            Whether to plot dots for all observations along the track. If false, dots will be plotted every 6 hours. Default is false.
        ax : axes
            Instance of axes to plot on. If none, one will be generated. Default is none.
        prop : dict
            Property of storm track lines.
        map_prop : dict
            Property of cartopy map.
        """

        #Set default properties
        default_prop={'dots':True,'fillcolor':'category','cmap':None,'levels':None,'linecolor':'k','linewidth':1.0,'ms':7.5}
        default_map_prop={'res':'m','land_color':'#FBF5EA','ocean_color':'#EDFBFF','linewidth':0.5,'linecolor':'k','figsize':(14,9),'dpi':200}

        #Initialize plot
        prop = self.add_prop(prop,default_prop)
        map_prop = self.add_prop(map_prop,default_map_prop)
        self.plot_init(ax,map_prop,plot_geography=False)

        #--------------------------------------------------------------------------------------

        #get resolution corresponding to string in prop
        res = self.check_res(map_prop['res'])

        #fill oceans if specified
        self.ax.set_facecolor(map_prop['ocean_color'])
        ocean_mask = self.ax.add_feature(cfeature.OCEAN.with_scale(res),facecolor=map_prop['ocean_color'],edgecolor='face',zorder=3)
        lake_mask = self.ax.add_feature(cfeature.LAKES.with_scale(res),facecolor=map_prop['ocean_color'],edgecolor='face',zorder=1)
        continent_mask = self.ax.add_feature(cfeature.LAND.with_scale(res),facecolor=map_prop['land_color'],edgecolor='face',zorder=0)

        #draw geography
        states = self.ax.add_feature(cfeature.STATES.with_scale(res),linewidths=map_prop['linewidth'],
                                     linestyle='solid',edgecolor=map_prop['linecolor'],zorder=4)
        countries = self.ax.add_feature(cfeature.BORDERS.with_scale(res),linewidths=map_prop['linewidth'],
                                        linestyle='solid',edgecolor=map_prop['linecolor'],zorder=4)
        coastlines = self.ax.add_feature(cfeature.COASTLINE.with_scale(res),linewidths=map_prop['linewidth'],
                                         linestyle='solid',edgecolor=map_prop['linecolor'],zorder=4)

        #--------------------------------------------------------------------------------------

        #Retrieve grid coordinates and values
        plot_grid = True
        if isinstance(grid,dict):
            if 'lat' in grid.keys():
                grid_lat = grid['lat']
                grid_lon = grid['lon']
                grid_val = grid['grid']
            else:
                ms = grid['ms']
                mec = grid['mec']
                mew = grid['mew']
                minimum_threshold = grid['minimum_threshold']
                plot_grid = False
        else:
            grid_lat = grid.lat.values
            grid_lon = grid.lon.values
            grid_val = grid.values

        #Determine levels and colormap
        if levels is None:
            levels = [1,2,3,4,6,8,10,12,16,20]
        if cmap is None:
            cmap = plt.cm.YlGn
        norm = col.BoundaryNorm(levels,cmap.N)

        #Contour fill grid if requested
        if plot_grid:
            self.ax.contourf(grid_lon, grid_lat, grid, levels, cmap=cmap, norm=norm, zorder=2)

        #Plot dots if requested
        else:

            #Iterate over sorted data (so highest totals show up on top)
            iter_df = storm.rain.sort_values('Total')
            for _,row in iter_df.iterrows():

                #Retrieve rain total and determine color
                rain_value = row['Total']
                if rain_value < minimum_threshold: continue
                color = self.rgb(cmap(norm(rain_value),bytes=True)[:-1])

                #Specify additional kwargs
                ms_kwargs = {}
                if mec is not None: ms_kwargs = {'mec':mec,'mew':mew}
                self.ax.plot(row['Lon'],row['Lat'],'o',ms=ms,color=color,**ms_kwargs,transform=ccrs.PlateCarree())

        #Produce colorbar
        cs = plt.cm.ScalarMappable(cmap=cmap, norm=norm)
        cs.set_array([])
        self.fig.colorbar(cs,ax=self.ax)

        #--------------------------------------------------------------------------------------

        #Keep record of lat/lon coordinate extrema
        max_lat = np.nanmax(storm.rain['Lat'].values)
        min_lat = np.nanmin(storm.rain['Lat'].values)
        max_lon = np.nanmax(storm.rain['Lon'].values)
        min_lon = np.nanmin(storm.rain['Lon'].values)

        #Get storm data
        storm_data = storm.dict

        #Retrieve storm data
        lats = storm_data['lat']
        lons = storm_data['lon']
        vmax = storm_data['vmax']
        styp = storm_data['type']
        sdate = storm_data['date']

        #Account for cases crossing dateline
        if self.proj.proj4_params['lon_0'] == 180.0:
            new_lons = np.array(lons)
            new_lons[new_lons<0] = new_lons[new_lons<0]+360.0
            lons = new_lons.tolist()

        #Force dynamic_tropical to tropical if an invest
        invest_bool = False
        if 'invest' in storm_data.keys() and storm_data['invest']:
            invest_bool = True
            if domain == 'dynamic_tropical': domain = 'dynamic'

        #Add to coordinate extrema
        if domain == 'dynamic_tropical':
            type_array = np.array(storm_data['type'])
            idx = np.where((type_array == 'SD') | (type_array == 'SS') | (type_array == 'TD') | (type_array == 'TS') | (type_array == 'HU'))
            use_lats = (np.array(storm_data['lat'])[idx]).tolist()
            use_lons = (np.array(lons)[idx]).tolist()
        else:
            use_lats = storm_data['lat']
            use_lons = np.copy(lons).tolist()

        #Iterate over storm data to plot
        for i,(i_lat,i_lon,i_vmax,i_mslp,i_date,i_type) in enumerate(zip(storm_data['lat'],lons,storm_data['vmax'],storm_data['mslp'],storm_data['date'],storm_data['type'])):

            #Determine line color, with SSHWS scale used as default
            if prop['linecolor'] == 'category':
                segmented_colors = True
                line_color = get_colors_sshws(np.nan_to_num(i_vmax))

            #Use user-defined colormap if another storm variable
            elif isinstance(prop['linecolor'],str) == True and prop['linecolor'] in ['vmax','mslp']:
                segmented_colors = True
                color_variable = storm_data[prop['linecolor']]
                if prop['levels'] is None: #Auto-determine color levels if needed
                    prop['levels'] = [np.nanmin(color_variable),np.nanmax(color_variable)]
                cmap,levels = get_cmap_levels(prop['linecolor'],prop['cmap'],prop['levels'])
                line_color = cmap((color_variable-min(levels))/(max(levels)-min(levels)))[i]

            #Otherwise go with user input as is
            else:
                segmented_colors = False
                line_color = prop['linecolor']

            #For tropical/subtropical types, color-code if requested
            if i > 0:
                if i_type in ['SD','TD','SS','TS','HU']:

                    #Plot underlying black and overlying colored line
                    self.ax.plot([lons[i-1],lons[i]],[storm_data['lat'][i-1],storm_data['lat'][i]],'-',
                                  linewidth=prop['linewidth']*1.33,color='k',zorder=5,
                                  transform=ccrs.PlateCarree())
                    self.ax.plot([lons[i-1],lons[i]],[storm_data['lat'][i-1],storm_data['lat'][i]],'-',
                                  linewidth=prop['linewidth'],color=line_color,zorder=6,
                                  transform=ccrs.PlateCarree())

                #For non-tropical types, plot dotted lines
                else:

                    #Restrict line width to 1.5 max
                    line_width = prop['linewidth'] + 0.0
                    if line_width > 1.5: line_width = 1.5

                    #Plot dotted line
                    self.ax.plot([lons[i-1],lons[i]],[storm_data['lat'][i-1],storm_data['lat'][i]],':',
                                  linewidth=line_width,color=line_color,zorder=6,
                                  transform=ccrs.PlateCarree(),
                                  path_effects=[path_effects.Stroke(linewidth=line_width*1.33, foreground='k'),
                                                path_effects.Normal()])

            #Plot dots if requested
            if prop['dots'] == True:

                #Skip if plot_all_dots == False and not in 0,6,12,18z
                if plot_all_dots == False:
                    if i_date.strftime('%H%M') not in ['0000','0600','1200','1800']: continue

                #Determine fill color, with SSHWS scale used as default
                if prop['fillcolor'] == 'category':
                    segmented_colors = True
                    fill_color = get_colors_sshws(np.nan_to_num(i_vmax))

                #Use user-defined colormap if another storm variable
                elif isinstance(prop['fillcolor'],str) == True and prop['fillcolor'] in ['vmax','mslp']:
                    segmented_colors = True
                    color_variable = storm_data[prop['fillcolor']]
                    if prop['levels'] is None: #Auto-determine color levels if needed
                        prop['levels'] = [np.nanmin(color_variable),np.nanmax(color_variable)]
                    cmap,levels = get_cmap_levels(prop['fillcolor'],prop['cmap'],prop['levels'])
                    fill_color = cmap((color_variable-min(levels))/(max(levels)-min(levels)))[i]

                #Otherwise go with user input as is
                else:
                    segmented_colors = False
                    fill_color = prop['fillcolor']

                #Determine dot type
                marker_type = '^'
                if i_type in ['SD','SS']:
                    marker_type = 's'
                elif i_type in ['TD','TS','HU']:
                    marker_type = 'o'

                #Plot marker
                self.ax.plot(i_lon,i_lat,marker_type,mfc=fill_color,mec='k',mew=0.5,
                             zorder=7,ms=prop['ms'],transform=ccrs.PlateCarree())

            #Label track dots
            if track_labels in ['valid_utc']:
                if track_labels == 'valid_utc':
                    strformat = '%H UTC \n%-m/%-d'
                    labels = {t.strftime(strformat):(x,y) for t,x,y in zip(sdate,lons,lats) if t.hour==0}
                    track = {t.strftime(strformat):(x,y) for t,x,y in zip(sdate,lons,lats)}
                self.plot_track_labels(self.ax, labels, track, k=.9)

        #--------------------------------------------------------------------------------------

        #Storm-centered plot domain
        if domain == "dynamic" or domain == "dynamic_tropical":

            bound_w,bound_e,bound_s,bound_n = self.dynamic_map_extent(min_lon,max_lon,min_lat,max_lat)
            self.ax.set_extent([bound_w,bound_e,bound_s,bound_n], crs=ccrs.PlateCarree())

        #Pre-generated or custom domain
        else:
            bound_w,bound_e,bound_s,bound_n = self.set_projection(domain)

        #Plot parallels and meridians
        #This is currently not supported for all cartopy projections.
        try:
            self.plot_lat_lon_lines([bound_w,bound_e,bound_s,bound_n],zorder=9)
        except:
            pass

        #--------------------------------------------------------------------------------------

        #Add left title
        type_array = np.array(storm_data['type'])
        if invest_bool == False:
            idx = np.where((type_array == 'SD') | (type_array == 'SS') | (type_array == 'TD') | (type_array == 'TS') | (type_array == 'HU'))
            tropical_vmax = np.array(storm_data['vmax'])[idx]

            #Coerce to include non-TC points if storm hasn't been designated yet
            add_ptc_flag = False
            if len(tropical_vmax) == 0:
                add_ptc_flag = True
                idx = np.where((type_array == 'LO') | (type_array == 'DB'))
            tropical_vmax = np.array(storm_data['vmax'])[idx]

            subtrop = classify_subtropical(np.array(storm_data['type']))
            peak_idx = storm_data['vmax'].index(np.nanmax(tropical_vmax))
            peak_basin = storm_data['wmo_basin'][peak_idx]
            storm_type = get_storm_classification(np.nanmax(tropical_vmax),subtrop,peak_basin)
            if add_ptc_flag == True: storm_type = "Potential Tropical Cyclone"
            left_title_string = f"{storm_type} {storm_data['name']}"
        else:
            #Use all indices for invests
            idx = np.array([True for i in type_array])
            add_ptc_flag = False
            tropical_vmax = np.array(storm_data['vmax'])

            #Determine letter in front of invest
            add_letter = 'L'
            if storm_data['id'][0] == 'C':
                add_letter = 'C'
            elif storm_data['id'][0] == 'E':
                add_letter = 'E'

            #Add title
            left_title_string = f"INVEST {storm_data['id'][2:4]}{add_letter}"

        #Add left title
        if plot_grid == True:
            left_title_string += "\nInterpolated WPC Storm Rainfall (in)"
        else:
            if minimum_threshold > 1:
                left_title_string += f"\nWPC Storm Rainfall (>{np.round(minimum_threshold,2)} inch)"
            else:
                left_title_string += f"\nWPC Storm Rainfall (inch)"
        self.ax.set_title(left_title_string,loc='left',fontsize=16,fontweight='bold')

        #Add right title
        ace = storm_data['ace']
        if add_ptc_flag == True: ace = 0.0
        type_array = np.array(storm_data['type'])

        #Get storm extrema for display
        mslp_key = 'mslp' if 'wmo_mslp' not in storm_data.keys() else 'wmo_mslp'
        if all_nan(np.array(storm_data[mslp_key])[idx]) == True:
            min_pres = "N/A"
        else:
            min_pres = int(np.nan_to_num(np.nanmin(np.array(storm_data[mslp_key])[idx])))
        if all_nan(np.array(storm_data['vmax'])[idx]) == True:
            max_wind = "N/A"
        else:
            max_wind = int(np.nan_to_num(np.nanmax(np.array(storm_data['vmax'])[idx])))
        start_date = dt.strftime(np.array(storm_data['date'])[idx][0],'%d %b %Y')
        end_date = dt.strftime(np.array(storm_data['date'])[idx][-1],'%d %b %Y')
        endash = u"\u2013"
        dot = u"\u2022"
        self.ax.set_title(f"{start_date} {endash} {end_date}\n{max_wind} kt {dot} {min_pres} hPa {dot} {ace:.1f} ACE",loc='right',fontsize=13)

        #--------------------------------------------------------------------------------------

        #Add plot credit
        warning_text=""
        if storm_data['source'] == 'ibtracs' and storm_data['source_info'] == 'World Meteorological Organization (official)':
            warning_text = f"This plot uses 10-minute averaged WMO official wind data converted\nto 1-minute average (factor of 0.88). Use this wind data with caution.\n\n"

            self.ax.text(0.99,0.01,warning_text,fontsize=9,color='k',alpha=0.7,
            transform=self.ax.transAxes,ha='right',va='bottom',zorder=10)

        credit_text = "Rainfall data (inch) from\nWeather Prediction Center (WPC)\n\n"
        self.ax.text(0.99,0.01,credit_text,fontsize=9,color='k',alpha=0.7,
                     transform=self.ax.transAxes,ha='right',va='bottom',zorder=10)

        credit_text = self.plot_credit()
        self.add_credit(credit_text)

        #--------------------------------------------------------------------------------------

        #Add legend
        if prop['fillcolor'] == 'category' and prop['dots'] == True:
            ex = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Non-Tropical', marker='^', color='w')
            sb = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Subtropical', marker='s', color='w')
            td = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Tropical Depression', marker='o', color=get_colors_sshws(33))
            ts = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Tropical Storm', marker='o', color=get_colors_sshws(34))
            c1 = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Category 1', marker='o', color=get_colors_sshws(64))
            c2 = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Category 2', marker='o', color=get_colors_sshws(83))
            c3 = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Category 3', marker='o', color=get_colors_sshws(96))
            c4 = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Category 4', marker='o', color=get_colors_sshws(113))
            c5 = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Category 5', marker='o', color=get_colors_sshws(137))
            self.ax.legend(handles=[ex,sb,td,ts,c1,c2,c3,c4,c5], prop={'size':11.5})

        elif prop['linecolor'] == 'category' and prop['dots'] == False:
            ex = mlines.Line2D([], [], linestyle='dotted', label='Non-Tropical', color='k')
            td = mlines.Line2D([], [], linestyle='solid', label='Sub/Tropical Depression', color=get_colors_sshws(33))
            ts = mlines.Line2D([], [], linestyle='solid', label='Sub/Tropical Storm', color=get_colors_sshws(34))
            c1 = mlines.Line2D([], [], linestyle='solid', label='Category 1', color=get_colors_sshws(64))
            c2 = mlines.Line2D([], [], linestyle='solid', label='Category 2', color=get_colors_sshws(83))
            c3 = mlines.Line2D([], [], linestyle='solid', label='Category 3', color=get_colors_sshws(96))
            c4 = mlines.Line2D([], [], linestyle='solid', label='Category 4', color=get_colors_sshws(113))
            c5 = mlines.Line2D([], [], linestyle='solid', label='Category 5', color=get_colors_sshws(137))
            self.ax.legend(handles=[ex,td,ts,c1,c2,c3,c4,c5], prop={'size':11.5}).set_zorder(10)

        elif prop['dots'] and segmented_colors == False:
            ex = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Non-Tropical', marker='^', color=prop['fillcolor'])
            sb = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Subtropical', marker='s', color=prop['fillcolor'])
            td = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Tropical', marker='o', color=prop['fillcolor'])
            handles=[ex,sb,td]
            self.ax.legend(handles=handles,fontsize=11.5).set_zorder(10)

        elif prop['dots'] == False and segmented_colors == False:
            ex = mlines.Line2D([], [], linestyle='dotted',label='Non-Tropical', color=prop['linecolor'])
            td = mlines.Line2D([], [], linestyle='solid',label='Tropical', color=prop['linecolor'])
            handles=[ex,td]
            self.ax.legend(handles=handles,fontsize=11.5).set_zorder(10)

        elif prop['dots'] == True:
            ex = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Non-Tropical', marker='^', color='w')
            sb = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Subtropical', marker='s', color='w')
            td = mlines.Line2D([], [], linestyle='None', ms=prop['ms'], mec='k',mew=0.5, label='Tropical', marker='o', color='w')
            handles=[ex,sb,td]
            for _ in range(7):
                handles.append(mlines.Line2D([], [], linestyle='-',label='',lw=0))
            l=self.ax.legend(handles=handles,fontsize=11.5).set_zorder(10)
            plt.draw()

            #Get the bbox
            bb = l.legendPatch.get_bbox().inverted().transform(self.fig.transFigure)

            #Define colorbar axis
            cax = self.fig.add_axes([bb.x0+0.47*bb.width, bb.y0+.057*bb.height, 0.015, .65*bb.height])
            norm = mlib.colors.Normalize(vmin=min(levels), vmax=max(levels))
            cbmap = mlib.cm.ScalarMappable(norm=norm, cmap=cmap)
            cbar = self.fig.colorbar(cbmap,cax=cax,orientation='vertical',\
                                     ticks=levels)

            cax.tick_params(labelsize=11.5)
            cax.yaxis.set_ticks_position('left')
            cbar.set_label(prop['fillcolor'],fontsize=11.5,rotation=90)

            rect_offset = 0.0
            if prop['cmap'] == 'category' and prop['fillcolor'] == 'vmax':
                cax.yaxis.set_ticks(np.linspace(min(levels),max(levels),len(levels)))
                cax.yaxis.set_ticklabels(levels)
                cax2 = cax.twinx()
                cax2.yaxis.set_ticks_position('right')
                cax2.yaxis.set_ticks((np.linspace(0,1,len(levels))[:-1]+np.linspace(0,1,len(levels))[1:])*.5)
                cax2.set_yticklabels(['TD','TS','Cat-1','Cat-2','Cat-3','Cat-4','Cat-5'],fontsize=11.5)
                cax2.tick_params('both', length=0, width=0, which='major')
                cax.yaxis.set_ticks_position('left')
                rect_offset = 0.7
            if prop['fillcolor'] == 'date':
                cax.set_yticklabels([f'{mdates.num2date(i):%b %-d}' for i in clevs],fontsize=11.5)

        else:
            ex = mlines.Line2D([], [], linestyle='dotted',label='Non-Tropical', color='k')
            td = mlines.Line2D([], [], linestyle='solid',label='Tropical', color='k')
            handles=[ex,td]
            for _ in range(7):
                handles.append(mlines.Line2D([], [], linestyle='-',label='',lw=0))
            l=self.ax.legend(handles=handles,fontsize=11.5).set_zorder(10)
            plt.draw()

            #Get the bbox
            bb = l.legendPatch.get_bbox().inverted().transform(self.fig.transFigure)

            #Define colorbar axis
            cax = self.fig.add_axes([bb.x0+0.47*bb.width, bb.y0+.057*bb.height, 0.015, .65*bb.height])
            norm = mlib.colors.Normalize(vmin=min(levels), vmax=max(levels))
            cbmap = mlib.cm.ScalarMappable(norm=norm, cmap=cmap)
            cbar = self.fig.colorbar(cbmap,cax=cax,orientation='vertical',\
                                     ticks=levels)

            cax.tick_params(labelsize=11.5)
            cax.yaxis.set_ticks_position('left')
            cbarlab = make_var_label(prop['linecolor'],storm_data)
            cbar.set_label(cbarlab,fontsize=11.5,rotation=90)

            rect_offset = 0.0
            if prop['cmap'] == 'category' and prop['linecolor'] == 'vmax':
                cax.yaxis.set_ticks(np.linspace(min(levels),max(levels),len(levels)))
                cax.yaxis.set_ticklabels(levels)
                cax2 = cax.twinx()
                cax2.yaxis.set_ticks_position('right')
                cax2.yaxis.set_ticks((np.linspace(0,1,len(levels))[:-1]+np.linspace(0,1,len(levels))[1:])*.5)
                cax2.set_yticklabels(['TD','TS','Cat-1','Cat-2','Cat-3','Cat-4','Cat-5'],fontsize=11.5)
                cax2.tick_params('both', length=0, width=0, which='major')
                cax.yaxis.set_ticks_position('left')
                rect_offset = 0.7
            if prop['linecolor'] == 'date':
                cax.set_yticklabels([f'{mdates.num2date(i):%b %-d}' for i in clevs],fontsize=11.5)

        #-----------------------------------------------------------------------------------------

        #Save image if specified
        if save_path is not None and isinstance(save_path,str) == True:
            plt.savefig(save_path,bbox_inches='tight')

        #Return axis if specified, otherwise display figure
        return self.ax

ext: py

max_stars_repo_name: ATMOcanes/tropycal

max_issues_repo_name: ATMOcanes/tropycal

max_forks_repo_name: ATMOcanes/tropycal

reason: import numpy,import scipy


text:
#this script is used to add rBC mass to the database

import sys
import os
import datetime
import pickle
import numpy as np
import matplotlib.pyplot as plt
from pprint import pprint
from scipy.optimize import curve_fit
from scipy import stats
from SP2_particle_record_UTC import ParticleRecord
from struct import *
import hk_new
import hk_new_no_ts_LEO
from scipy import linspace, polyval, polyfit, sqrt, stats
import math
import mysql.connector
from datetime import datetime
import calendar

#setup
data_dir = 'F:/Alert/2014/SP2B_files/'
start_analysis_at = datetime(2014,12,24)
end_analysis_at = 	datetime(2015,1,1)
SP2_number = 58
min_incand_BBHG = 50

record_size_bytes = 1658 #size of a single particle record in bytes(UBC_SP2 = 1498, EC_SP2 in 2009 and 2010 = 2458, Alert SP2 #4 and #58 = 832? )

#database connection
cnx = mysql.connector.connect(user='root', password='Suresh15', host='localhost', database='black_carbon')
cursor = cnx.cursor()

#**********parameters dictionary**********
parameters = {
'acq_rate': 2500000, #5000000,
}

def checkHKId(particle_event_time,instr_number):
	event_minute = int(particle_event_time-particle_event_time%60)  #hk data is in 1min intervals, so need event minute
	cursor.execute(('SELECT id FROM alert_hk_data WHERE SP2_ID =%s AND UNIX_UTC_ts = %s'),(instr_number,event_minute))
	hk_data = cursor.fetchall()
	if hk_data != []:
		hk_id= hk_data[0][0]
	else:
		hk_id = None

	return hk_id

def getParticleData(parameters_dict,instr_number,bbhg_incand_min,prev_particle_ts):

	f = open(parameters_dict['file'], 'rb')

	start_time = prev_particle_ts
	record_index = 0
	current_minute = 0

	##Retrieve HK id for the event to join tables on later
	housekeeping_id = checkHKId(start_time,instr_number)

	temp={
	'rBC_mass_fg_HG_low':[],
	'rBC_mass_fg_HG_high':[],
	'rBC_mass_fg_LG_low':[],
	'rBC_mass_fg_LG_high':[],
	}

	while record_index < parameters['number_of_records']:

		#read the binary for a particle
		record = f.read(parameters['record_size'])

		particle_record = ParticleRecord(record, parameters_dict['acq_rate'])

		#run the wideband HG incandPeakInfo method to retrieve various HG BB incandescence peak attributes
		particle_record.incandPeakInfo()
		bbhg_incand_pk_amp = float(particle_record.incandMax)

		#if this is an incandescent particle that we can detect with the HG channel then continue
		if bbhg_incand_pk_amp >= bbhg_incand_min:
			event_time = particle_record.timestamp  #UTC
			event_time_dt = datetime.utcfromtimestamp(event_time)

			#get incandescent signals
			particle_record.incandPeakInfoLG()
			bblg_incand_pk_amp = float(particle_record.incandMax_LG)

			particle_record.narrowIncandPeakInfo()
			nbhg_incand_pk_amp = float(particle_record.narrowIncandMax)

			particle_record.narrowIncandPeakInfoLG()
			nblg_incand_pk_amp = float(particle_record.narrowIncandMax_LG)

			#calculate masses
			bbhg_mass = 0.41527 + 2.13238E-4*bbhg_incand_pk_amp + 7.17406E-10*bbhg_incand_pk_amp*bbhg_incand_pk_amp
			bbhg_mass_uncertainty = 0.0869 + 1.09646E-5*bbhg_incand_pk_amp + 2.2873E-10*bbhg_incand_pk_amp*bbhg_incand_pk_amp

			bblg_mass = 0.70095 + 0.00182*bblg_incand_pk_amp + 1.22794E-7*bblg_incand_pk_amp*bblg_incand_pk_amp
			bblg_mass_uncertainty = 0.30863 + 1.38156E-4*bblg_incand_pk_amp + 1.17569E-8*bblg_incand_pk_amp*bblg_incand_pk_amp

			######
			if event_time_dt.minute == current_minute:
				if 0.33 <= bbhg_mass < 1.8:
					temp['rBC_mass_fg_HG_low'].append(bbhg_mass)
				if 1.8 <= bbhg_mass < 12.8:
					temp['rBC_mass_fg_HG_high'].append(bbhg_mass)
				if 1.8 <= bblg_mass < 12.8:
					temp['rBC_mass_fg_LG_low'].append(bblg_mass)
				if 12.8 <= bblg_mass < 36.3:
					temp['rBC_mass_fg_LG_high'].append(bblg_mass)
			else:
				end_time = event_time

				if 0.33 <= bbhg_mass < 1.8:
					temp['rBC_mass_fg_HG_low'].append(bbhg_mass)
				if 1.8 <= bbhg_mass < 12.8:
					temp['rBC_mass_fg_HG_high'].append(bbhg_mass)
				if 1.8 <= bblg_mass < 12.8:
					temp['rBC_mass_fg_LG_low'].append(bblg_mass)
				if 12.8 <= bblg_mass < 36.3:
					temp['rBC_mass_fg_LG_high'].append(bblg_mass)

				avg_rBC_mass_fg_HG_low =  float(np.nan_to_num(np.mean(temp['rBC_mass_fg_HG_low'])))
				rBC_number_HG_low =  len(temp['rBC_mass_fg_HG_low'])

				avg_rBC_mass_fg_HG_high = float(np.nan_to_num(np.mean(temp['rBC_mass_fg_HG_high'])))
				rBC_number_HG_high = len(temp['rBC_mass_fg_HG_high'])

				avg_rBC_mass_fg_LG_low =  float(np.nan_to_num(np.mean(temp['rBC_mass_fg_LG_low'])))
				rBC_number_LG_low = len(temp['rBC_mass_fg_LG_low'])

				avg_rBC_mass_fg_LG_high = float(np.nan_to_num(np.mean(temp['rBC_mass_fg_LG_high'])))
				rBC_number_LG_high = len(temp['rBC_mass_fg_LG_high'])

				#put data into database
				add_interval = ('''INSERT INTO alert_1min_mass_number_data (
						instrument_ID,
						UNIX_UTC_ts_int_start,
						UNIX_UTC_ts_int_end,
						rBC_mass_fg_HG_low,
						rBC_number_HG_low,
						rBC_mass_fg_HG_high,
						rBC_number_HG_high,
						rBC_mass_fg_LG_low,
						rBC_number_LG_low,
						rBC_mass_fg_LG_high,
						rBC_number_LG_high,
						HK_id
						)
						VALUES (
						%(instrument_ID)s,
						%(UNIX_UTC_ts_int_start)s,
						%(UNIX_UTC_ts_int_end)s,
						%(rBC_mass_fg_HG_low)s,
						%(rBC_number_HG_low)s,
						%(rBC_mass_fg_HG_high)s,
						%(rBC_number_HG_high)s,
						%(rBC_mass_fg_LG_low)s,
						%(rBC_number_LG_low)s,
						%(rBC_mass_fg_LG_high)s,
						%(rBC_number_LG_high)s,
						%(HK_id)s
						)''')

				interval_data = {
				'instrument_ID':instr_number,
				'UNIX_UTC_ts_int_start':start_time,
				'UNIX_UTC_ts_int_end':end_time,
				'rBC_mass_fg_HG_low':avg_rBC_mass_fg_HG_low,
				'rBC_number_HG_low':rBC_number_HG_low,
				'rBC_mass_fg_HG_high':avg_rBC_mass_fg_HG_high,
				'rBC_number_HG_high':rBC_number_HG_high,
				'rBC_mass_fg_LG_low': avg_rBC_mass_fg_LG_low,
				'rBC_number_LG_low': rBC_number_LG_low,
				'rBC_mass_fg_LG_high':avg_rBC_mass_fg_LG_high,
				'rBC_number_LG_high':rBC_number_LG_high,
				'HK_id':housekeeping_id,
				}

				cursor.execute(add_interval, interval_data)
				cnx.commit()

				temp={
				'rBC_mass_fg_HG_low':[],
				'rBC_mass_fg_HG_high':[],
				'rBC_mass_fg_LG_low':[],
				'rBC_mass_fg_LG_high':[],
				}

				start_time = event_time
				##Retrieve HK id for the event to join tables on later
				housekeeping_id = checkHKId(start_time,instr_number)

			current_minute = event_time_dt.minute

			#####

		record_index+=1

	f.close()
	return prev_particle_ts

os.chdir(data_dir)
for directory in os.listdir(data_dir):
	if os.path.isdir(directory) == True and directory.startswith('20'):
		parameters['folder']= directory
		folder_date = datetime.strptime(directory, '%Y%m%d')
		if folder_date >= start_analysis_at and folder_date < end_analysis_at:
			parameters['directory']=os.path.abspath(directory)
			os.chdir(parameters['directory'])

			#start the fitting
			prev_event_ts = 0
			for file in os.listdir('.'):
				if file.endswith('.sp2b') and (file.endswith('gnd.sp2b')==False):
					print file
					parameters['file'] = file
					path = parameters['directory'] + '/' + str(file)
					file_bytes = os.path.getsize(path) #size of entire file in bytes
					parameters['record_size'] = record_size_bytes
					parameters['number_of_records']= (file_bytes/parameters['record_size'])-1

					prev_event_ts = getParticleData(parameters,SP2_number,min_incand_BBHG,prev_event_ts)
			os.chdir(data_dir)
cnx.close()

reason: import numpy,from scipy


text:
"""Class for Rapidly Exponentially Stabilizing Control Lyapunov Functions (RES-CLFs)."""

from numpy import dot, identity
from numpy.linalg import eigvals
from scipy.linalg import block_diag

from .quadratic_control_lyapunov_function import QuadraticControlLyapunovFunction

class RESQuadraticControlLyapunovFunction(QuadraticControlLyapunovFunction):
    """Class for Rapidly Exponentially Stabilizing Control Lyapunov Functions (RES-CLFs)."""

    def __init__(self, robotic_system_output, P, Q, epsilon):
        M = block_diag(identity(robotic_system_output.k) / epsilon, identity(robotic_system_output.k))
        P = dot(M, dot(P, M)) * epsilon
        Q = dot(M, dot(Q, M))
        alpha = min(eigvals(Q)) / max(eigvals(P))
        QuadraticControlLyapunovFunction.__init__(self, robotic_system_output, P, alpha)

    def build_ctle(robotic_system_output, K, Q, epsilon=1):
        lyapunov_function = QuadraticControlLyapunovFunction.build_ctle(robotic_system_output, K, Q)
        return RESQuadraticControlLyapunovFunction(robotic_system_output, lyapunov_function.P, Q, epsilon)

    def build_care(robotic_system_output, Q, epsilon=1):
        lyapunov_function = QuadraticControlLyapunovFunction.build_care(robotic_system_output, Q)
        return RESQuadraticControlLyapunovFunction(robotic_system_output, lyapunov_function.P, Q, epsilon)

max_stars_repo_name: vdorobantu/lyapy

max_issues_repo_name: HaiminClack/lyapy

max_forks_repo_name: HaiminClack/lyapy

reason: from numpy,from scipy


text:
import gym
import gym_duckietown
from gym_duckietown.envs import DuckietownEnv
from gym_duckietown.simulator import Simulator
from gym_duckietown.wrappers import UndistortWrapper
import argparse
import numpy as np
import time
import pickle
# parser = argparse.ArgumentParser()
# parser.add_argument('--env-name', default=None)
# parser.add_argument('--map-name', default='udem1')
# parser.add_argument('--distortion', default=False, action='store_true')
# parser.add_argument('--draw-curve', action='store_true', help='draw the lane following curve')
# parser.add_argument('--draw-bbox', action='store_true', help='draw collision detection bounding boxes')
# parser.add_argument('--domain-rand', action='store_true', help='enable domain randomization')
# parser.add_argument('--frame-skip', default=1, type=int, help='number of frames to skip')
# args = parser.parse_args()
#
# if args.env_name is None:
#     env = DuckietownEnv(
#         map_name = args.map_name,
#         draw_curve = args.draw_curve,
#         draw_bbox = args.draw_bbox,
#         domain_rand = args.domain_rand,
#         frame_skip = args.frame_skip,
#         distortion = args.distortion,
#     )
# else:
#     env = gym.make(args.env_name)
class ActionWrapper(gym.ActionWrapper):
    def __init__(self, env):
        super(ActionWrapper, self).__init__(env)

    def action(self, action):
        action_ = [action[0] * 0.8, action[1]]
        return action_

def sigmoid(x):

    return 1/(1+np.exp(-x))
env = gym.make("Duckietown-small_loop-v0")
# env = launch_env()
env = ActionWrapper(env)
observation, tile_idx = env.reset()
# print(observation)

tile = []
tile.append(tile_idx)
for i in range(10000):
  action = env.action_space.sample() # your agent here (this takes random actions)
  action[0] = sigmoid(action[0])
  # env.render()
  observation, reward, done, info = env.step(action)
  if done:
      _, tile_idx = env.reset()
      tile.append(tile_idx)
import matplotlib.pyplot as plt
plt.hist(tile)
plt.show()

reason: import numpy


text:
import numpy as np
import pandas as pd
import ml_metrics
import base64
import matplotlib.pyplot as plt
import seaborn as sns
from mlxtend.frequent_patterns import apriori
from scipy.sparse import csc_matrix
from scipy.sparse.linalg import svds
from IPython.display import HTML

"""
redcarpet: Module for recommender systems using sets
"""

"""
HELPER METHODS
"""

def nonzero_index_set(arr):
    """
    Returns a set of indices corresponding to non-zero
    entries in a numpy array (or other list-like).
    """
    res = set()
    for i, val in enumerate(arr):
        if val > 0:
            res.add(i)
    return res

def mat_to_sets(mat):
    """
    Converts a numpy matrix into a list of sets of column
    indices corresponding to non-zero row entries.
    """
    return [nonzero_index_set(row) for row in mat]

def get_recs(user_recs, k=None):
    """
    Extracts recommended item indices, leaving out their scores.
    params:
        user_recs: list of lists of tuples of recommendations where
            each tuple has (item index, relevance score) with the
            list of tuples sorted in order of decreasing relevance
        k: maximumum number of recommendations to include for each
            user, if None, include all recommendations
    returns:
        list of lists of recommendations where each
            list has the column indices of recommended items
            sorted in order they appeared in user_recs
    """
    recs = [[item for item, score in recs][0:k] for recs in user_recs]
    return recs

def write_kaggle_recs(recs_list, filename=None, headers=["Id", "Predicted"]):
    """
    Writes recommendations to file in Kaggle submission format.
    params:
        recs_list: list of lists of recommendations where each
            list has the column indices of recommended items
            sorted in order of decreasing relevance
        filename: path to file for writing output
        headers: list of strings of output columns, defaults to
            submission columns: ["Id", "Predicted"]
    returns:
        int: number of non-header lines, where each line represents
            a user and the recommendations given to them
    """
    if filename is None:
        raise ValueError("Must provide a filename.")
    lines = [",".join(headers)]
    for i, recs in enumerate(recs_list):
        lines.append("{},{}".format(i, " ".join([str(v) for v in recs])))
    text = "\n".join(lines)
    with open(filename, "w") as file:
        file.write(text)
    return len(lines) - 1

def download_kaggle_recs(recs_list, filename=None, headers=["Id", "Predicted"]):
    """
    Writes recommendations to file in Kaggle submission format.
    params:
        recs_list: list of lists of recommendations where each
            list has the column indices of recommended items
            sorted in order of decreasing relevance
        filename: path to file for writing output
        headers: list of strings of output columns, defaults to
            submission columns: ["Id", "Predicted"]
    returns:
        html: HTML download link to display in a notebook, click
            to download the submission file
    """
    # Based on: https://www.kaggle.com/rtatman/download-a-csv-file-from-a-kernel
    if filename is None:
        raise ValueError("Must provide a filename.")
    rec_df = pd.DataFrame(
        [(i, " ".join([str(r) for r in recs])) for i, recs in enumerate(recs_list)],
        columns=headers,
    )
    csv = rec_df.to_csv(index=False)
    b64 = base64.b64encode(csv.encode())
    payload = b64.decode()
    html = """Download ({lines} lines): {filename}"""
    html = html.format(payload=payload, filename=filename, lines=len(rec_df))
    return HTML(html)

def check_list_of_sets(s_data, var_name):
    if not isinstance(s_data, list):
        raise ValueError(
            "{} must be a list of sets. Got: {}".format(var_name, type(s_data))
        )
    if len(s_data) > 0:
        entry = s_data[0]
        if not isinstance(entry, set):
            raise ValueError(
                "{} must be a list of sets. Got list of: {}".format(
                    var_name, type(entry)
                )
            )

"""
EVALUATION METRICS
"""

def mapk_score(s_hidden, recs_pred, k=10):
    """
    Computes the mean average precision at k (MAP@K) of recommendations.
    MAP@K = mean AP@K score over all users
    AP@K = (1 / min(m, k)) * sum from 1 to k of (precision at i * relevance of ith item)
    Where m is the number of items in a user's hidden set
    Where k is the number of items recommended to each user
    params:
        s_hidden: list of sets of hidden items for each user
        recs_pred: list of lists of recommended items, with each list
        k: number of recommendations to use in top set
    returns:
        float, range [0, 1]
    """
    check_list_of_sets(s_hidden, "s_hidden")
    return ml_metrics.mapk(s_hidden, recs_pred, k)

def uhr_score(s_hidden, recs_pred, k=10):
    """
    Computes the user hit rate (UHR) score of recommendations.
    UHR = the fraction of users whose top list included at
    least one item also in their hidden set.
    params:
        s_hidden: list of sets of hidden items for each user
        recs_pred: list of lists of recommended items, with each list
        k: number of recommendations to use in top set
    returns:
        float, range [0, 1]
    """
    check_list_of_sets(s_hidden, "s_hidden")
    if len(s_hidden) != len(recs_pred):
        note = "Length of true list {} does not match length of recommended list {}."
        raise ValueError(note.format(len(s_hidden), len(recs_pred)))
    scores = []
    for r_true, r_pred_orig in zip(s_hidden, recs_pred):
        r_pred = list(r_pred_orig)[0:k]
        intersect = set(r_true).intersection(set(r_pred))
        scores.append(1 if len(intersect) > 0 else 0)
    return np.mean(scores)

def get_apk_scores(s_hidden, recs_pred, k=10):
    """
    Returns the average precision at k (AP@K) for each user.
    AP@K = (1 / min(m, k)) * sum from 1 to k of (precision at i * relevance of ith item)
    Where m is the number of items in a user's hidden set
    Where k is the number of items recommended to each user
    params:
        s_hidden: list of sets of hidden items for each user
        recs_pred: list of lists of recommended items, with each list
        k: number of recommendations to use in top set
    returns:
        list of floats, each float in the range [0, 1]
    """
    check_list_of_sets(s_hidden, "s_hidden")
    apks = []
    for r_true, r_pred in zip(s_hidden, recs_pred):
        apk = mapk_score([r_true], [r_pred], k=k)
        apks.append(apk)
    return apks

def get_hit_counts(s_hidden, recs_pred, k=10):
    """
    Returns the number of successful recommendations for each user.
    params:
        s_hidden: list of sets of hidden items for each user
        recs_pred: list of lists of recommended items, with each list
        k: number of recommendations to use in top set
    returns:
        list of integers, each integer in the range [0, k]
    """
    check_list_of_sets(s_hidden, "s_hidden")
    hits = []
    for r_true, r_pred in zip(s_hidden, recs_pred):
        ix = r_true.intersection(set(r_pred[0:k]))
        hits.append(len(ix))
    return hits

def get_all_scores(rec_scores, k=10):
    """
    Get scores of all items in the list of lists of recommendations.
    """
    all_scores = []
    for recs in rec_scores:
        for (item, score) in recs[0:k]:
            all_scores.append(score)
    return all_scores

"""
ANALYSIS TOOLS
"""

def show_apk_dist(s_hidden, models, k=10, bin_size=0.1):
    """
    Plot a histogram of average precision scores for all users.
    """
    bins = np.arange(0, 1 + bin_size, bin_size)
    pal = sns.color_palette("hls", len(models))
    for ((rec_scores, name), color) in zip(models, pal):
        apks = get_apk_scores(s_hidden, get_recs(rec_scores), k=k)
        sns.distplot(apks, kde=False, label=name, bins=bins, color=color)
    plt.xticks(bins)
    plt.xlabel("Average Precision in Top {}".format(k))
    plt.ylabel("Number of Users")
    plt.title("AP@K Score Distribution")
    plt.gcf().set_size_inches((8, 5))
    plt.grid()
    plt.legend(
        loc="upper left", bbox_to_anchor=(1.0, 1.0), title="Models", frameon=False
    )
    plt.show()

def show_hit_dist(s_hidden, models, k=10):
    """
    Plot a histogram of hit counts for all users.
    """
    bins = range(k + 1)
    pal = sns.color_palette("hls", len(models))
    for ((rec_scores, name), color) in zip(models, pal):
        hits = get_hit_counts(s_hidden, get_recs(rec_scores), k=k)
        sns.distplot(hits, kde=False, label=name, bins=bins, color=color)
    plt.xticks(bins)
    plt.xlabel("Number of Successful Recommendations in Top {}".format(k))
    plt.ylabel("Number of Users")
    plt.title("Hit Count Distribution")
    plt.gcf().set_size_inches((8, 5))
    plt.grid()
    plt.legend(
        loc="upper left", bbox_to_anchor=(1.0, 1.0), title="Models", frameon=False
    )
    plt.show()

def show_score_dist(models, k=10, bins=None):
    """
    Plot a histogram of item recommendation scores for all users.
    """
    pal = sns.color_palette("hls", len(models))
    for ((rec_scores, name), color) in zip(models, pal):
        scores = get_all_scores(rec_scores, k=k)
        if bins is not None:
            sns.distplot(scores, kde=False, label=name, color=color, bins=bins)
        else:
            sns.distplot(scores, kde=False, label=name, color=color)
    if bins is not None:
        plt.xticks(bins)
    plt.xlabel("Score for Recommended Item in Top {}".format(k))
    plt.ylabel("Number of Items")
    plt.title("Item Score Distribution")
    plt.gcf().set_size_inches((8, 5))
    plt.grid()
    plt.legend(
        loc="upper left", bbox_to_anchor=(1.0, 1.0), title="Models", frameon=False
    )
    plt.show()

def show_user_detail(s_input, s_hidden, rec_scores, uid, name_fn=None, k=10):
    """
    Show the detailed results of recommendations to a user.
    """
    s_pred = get_recs(rec_scores)
    print("User: {}".format(uid))
    print("Given:       {}".format(sorted(s_input[uid])))
    print("Recommended: {}".format(sorted(s_pred[uid])))
    print("Actual:      {}".format(sorted(s_hidden[uid])))
    set_intersect = set(s_pred[uid]).intersection(set(s_hidden[uid]))
    n_intersect = len(set_intersect)
    apk = mapk_score([s_hidden[uid]], [s_pred[uid]], k)
    print()
    print("Recommendation Hits = {}".format(n_intersect))
    print("Average Precision   = {0:.3f}".format(apk))
    print()
    print("All Recommendation Scores:")
    for i, (item_id, score) in enumerate(rec_scores[uid]):
        hit = "Y" if item_id in s_hidden[uid] else " "
        item_name = "Item {}".format(item_id)
        if name_fn is not None:
            item_name = name_fn(item_id)
        print(
            "{0}. [{3}] ({2:.3f}) {1}".format(
                str(i + 1).zfill(2), item_name, score, hit
            )
        )

def show_user_recs(s_hidden, rec_scores, k=10):
    """
    Show a table of recommendation results by user.
    """
    apks = get_apk_scores(s_hidden, get_recs(rec_scores), k=k)
    hits = get_hit_counts(s_hidden, get_recs(rec_scores), k=k)
    cols = ["User", "APK", "Hits"]
    data = {"User": range(len(rec_scores)), "APK": apks, "Hits": hits}
    return pd.DataFrame(data)[cols]

def show_item_recs(s_hidden, rec_scores, k=10):
    """
    Show a table of recommendation results by item.
    """
    item_res = {}
    for (user, likes) in zip(rec_scores, s_hidden):
        for (i, score) in user:
            if i not in item_res:
                item_res[i] = {"Item": i, "Results": [], "Scores": []}
            item_res[i]["Results"].append(1 if i in likes else 0)
            item_res[i]["Scores"].append(score)
    res = []
    for i in item_res:
        record = item_res[i]
        total = len(record["Results"])
        hits = sum(record["Results"])
        res.append(
            {
                "Item": i,
                "Recommended": total,
                "Hits": hits,
                "Hit Rate": hits / total,
                "Avg Score": np.mean(record["Scores"]),
            }
        )
    cols = ["Item", "Recommended", "Hits", "Hit Rate", "Avg Score"]
    return pd.DataFrame(res)[cols]

"""
SIMILARITY MEASURES
"""

def jaccard_sim(u, v):
    """
    Computes the Jaccard similarity between sets u and v.
    sim = intersection(u, v) / union(u, v)
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    intersection = len(u.intersection(v))
    union = len(u.union(v))
    zero = 1e-10
    # Add small value to denominator to avoid divide by zero
    sim = intersection / (union + zero)
    return sim

def cosine_sim(u, v):
    """
    Computes the Cosine similarity between sets u and v.
    sim = intersection(u, v) / sqrt(|u| * |v|)
    Where |s| is the number of items in set s
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    intersection = len(u.intersection(v))
    mag_u = len(u)
    mag_v = len(v)
    zero = 1e-10
    # Add small value to denominator to avoid divide by zero
    sim = intersection / (np.sqrt(mag_u * mag_v) + zero)
    return sim

def forbes_sim(u, v):
    """
    Computes the Forbes similarity between sets u and v.
    sim = a/((a+b)*(a+c))
    Where a = # of items in intersection(u, v)
          b = # of items only in u
          c = # of items only in v
    Note: n is omitted since it is constant for all vectors.
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    a = len(u.intersection(v))
    b = len(u) - a
    c = len(v) - a
    zero = 1e-10
    sim = a / (((a + b) * (a + c)) + zero)
    return sim

def mcconnoughy_sim(u, v):
    """
    Computes the McConnoughy similarity between sets u and v.
    sim = (a*a - b*c) / sqrt((a+b)*(a+c))
    Where a = # of items in intersection(u, v)
          b = # of items only in u
          c = # of items only in v
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    a = len(u.intersection(v))
    b = len(u) - a
    c = len(v) - a
    zero = 1e-10
    sim = ((a * a) - (b * c)) / (np.sqrt((a + b) * (a + c)) + zero)
    return sim

def simpson_sim(u, v):
    """
    Computes the Simpson similarity coefficient between sets u and v.
    sim = intersection(u, v) / min(|u|, |v|)
    Where |s| is the number of items in set s
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    ix = len(u.intersection(v))
    zero = 1e-10
    sim = ix / (min(len(u), len(v)) + zero)
    return sim

def first_kulczynski_sim(u, v):
    """
    Computes the first Kulczynski similarity between sets u and v.
    sim = a / (b + c)
    Where a = # of items in intersection(u, v)
          b = # of items only in u
          c = # of items only in v
    Note: If (b + c) is zero, this measure is undefined. In this
        implementation, a small value (1e-4) is added to the
        denominator to avoid division by zero. Consequently, the
        similarity between two sets with `a` matches will be equal
        to a / 1e-4, which is equivalent to a * 1000
    params:
        u, v: sets to compare
    returns:
        float from zero to infinity, where higher scores represents
            greater similarity and zero represents no similarity
    """
    a = len(u.intersection(v))
    b = len(u) - a
    c = len(v) - a
    zero = 1e-4
    sim = a / (b + c + zero)
    return sim

def second_kulczynski_sim(u, v):
    """
    Computes the second Kulczynski similarity between sets u and v.
    sim = (1/2) * ((a / (a + b)) + (a / (a + c)) )
    Where a = # of items in intersection(u, v)
          b = # of items only in u
          c = # of items only in v
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    a = len(u.intersection(v))
    b = len(u) - a
    c = len(v) - a
    zero = 1e-10
    sim = ((a / (a + b + zero)) + (a / (a + c + zero))) / 2
    return sim

def sorenson_dice_sim(u, v):
    """
    Computes the Sørensen-Dice similarity coefficient between sets u and v.
    sim = (2 * intersection(u, v)) / (|u| + |v|)
    Where |s| is the number of items in set s
    params:
        u, v: sets to compare
    returns:
        float between 0 and 1, where 1 represents perfect
            similarity and 0 represents no similarity
    """
    ix = len(u.intersection(v))
    zero = 1e-10
    sim = (2 * ix) / (len(u) + len(v) + zero)
    return sim

"""
ASSOCIATION RULE METRICS
Based on: https://github.com/resumesai/resumesai.github.io/blob/master/analysis/Rule%20Mining.ipynb
"""

def sets_to_contingency(a, b, N):
    """
    Creates a contingency table from two sets.
    params:
        a, b: sets to compare
        N: total number of possible items
    returns:
        (f11, f10, f01, f00) tuple of contingency table entries:
        f11 = # of items both in a and b
        f10 = # of items only in a
        f01 = # of items only in b
        f00 = # of items not in either a or b
    """
    f11 = len(a.intersection(b))
    f10 = len(a) - f11
    f01 = len(b) - f11
    f00 = N - (f11 + f10 + f01)
    return (f11, f10, f01, f00)

def rule_support(f11, f10, f01, f00):
    """
    Computes the support for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1] where 1 indicates maximum support and
            0 indicates no support
    """
    N = f11 + f10 + f01 + f00
    zero = 1e-10
    return f11 / (N + zero)

def rule_confidence(f11, f10, f01, f00):
    """
    Computes the confidence for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1] where 1 indicates maximum confidence and
            0 indicates no confidence
    """
    zero = 1e-10
    return f11 / (f11 + f10 + zero)

def rule_lift(f11, f10, f01, f00):
    """
    Computes the lift for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float ranging from zero to infinity where 1 implies independence, greater
        than 1 implies positive association, less than 1 implies negative association
    """
    N = f11 + f10 + f01 + f00
    zero = 1e-10
    supp_ab = f11 / N
    supp_a = f10 / N
    supp_b = f01 / N
    return supp_ab / ((supp_a * supp_b) + zero)

def rule_conviction(f11, f10, f01, f00):
    """
    Computes the conviction for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1]
    """
    N = f11 + f10 + f01 + f00
    zero = 1e-10
    supp_b = f01 / N
    conf = rule_confidence(f11, f10, f01, f00)
    return (1 - supp_b) / ((1 - conf) + zero)

def rule_power_factor(f11, f10, f01, f00):
    """
    Computes the rule power factor (RPF) for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1]
    """
    N = f11 + f10 + f01 + f00
    zero = 1e-10
    supp_ab = f11 / N
    supp_a = f10 / N
    return (supp_ab * supp_ab) / (supp_a + zero)

def rule_interest_factor(f11, f10, f01, f00):
    """
    Computes the interest factor for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1] where 1 indicates maximum interest and
            0 indicates no interest
    """
    N = f11 + f10 + f01 + f00
    zero = 1e-10
    f1p = f11 + f10
    fp1 = f11 + f01
    return (N * f11) / ((f1p * fp1) + zero)

def rule_phi_correlation(f11, f10, f01, f00):
    """
    Computes the phi correlation for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [-1, 1] where 1 indicates perfect positive correlation and
            -1 indicates perfect negative correlation.
    """
    f1p = f11 + f10
    f0p = f01 + f00
    fp1 = f11 + f01
    fp0 = f10 + f00
    num = (f11 * f00) - (f01 * f10)
    denom = np.sqrt(f1p * fp1 * f0p * fp0)
    if denom == 0:
        return 0.0
    return num / denom

def rule_is_score(f11, f10, f01, f00):
    """
    Computes the IS score for a rule `a -> b` based on the contingency table.
    params:
        f11 = count a and b appearing together
        f10 = count of a appearing without b
        f01 = count of b appearing without a
        f00 = count of neither a nor b appearing
    returns:
        float in range [0, 1] where 1 indicates maximum int3erest and
            0 indicates no int3erest
    """
    intfac = rule_interest_factor(f11, f10, f01, f00)
    supp = rule_support(f11, f10, f01, f00)
    return np.sqrt(intfac * supp)

def mine_association_rules(m_train, min_support=0.5):
    """
    Finds association rules using the Apriori algorithm. Produces rules of the form `a -> b`,
    which suggests that if a user likes item `a`, then they may also like item `b`.
    params:
        m_train: matrix of train data, rows = users, columns = items, 1 = like, 0 otherwise
        min_support: A float between 0 and 1 for minumum support of the itemsets returned.
          The support is computed as the fraction:
            transactions_where_item(s)_occur / total_transactions
    returns:
        rule_df: Pandas dataframe of association rules, with columns:
            "a": antecedent (LHS) item index
            "b": consequent (RHS) item index
            "ct": tuple of contingency table entries for the rule
            "support": support for the rule `a -> b` in m_train
    """
    freq_is = apriori(pd.DataFrame(m_train), max_len=2, min_support=min_support)
    freq_is["len"] = freq_is["itemsets"].apply(lambda s: len(s))
    freq_is = freq_is.query("len == 2")
    if len(freq_is) == 0:
        return pd.DataFrame([], columns=["a", "b", "ct", "support"])
    item_counts = m_train.sum(axis=0)
    rules = []
    for record in freq_is.to_dict(orient="records"):
        fset = record["itemsets"]
        a = min(fset)
        b = max(fset)
        n = len(m_train)
        supp = record["support"]
        all_a = item_counts[a]
        all_b = item_counts[b]
        both = supp * n
        f11 = int(both)
        f10 = int(all_a - both)
        f01 = int(all_b - both)
        f00 = int(n - (f11 + f10 + f01))
        rules.append({"a": a, "b": b, "ct": (f11, f10, f01, f00), "support": supp})
        rules.append({"a": b, "b": a, "ct": (f11, f01, f10, f00), "support": supp})
    rule_df = pd.DataFrame(rules)
    return rule_df

def rank_association_rules(mined_rules_df, score_fn, score_name="score"):
    """
    Ranks association rules according to a given score function.
    params:
        mined_rules_df: Pandas dataframe of association rules to rank, with columns:
            "a": antecedent (LHS) item index
            "b": consequent (RHS) item index
            "ct": tuple of contingency table entries for the rule
        score_fn(f11, f10, f01, f11): function for scoring rules, takes the entries
            of the contingency table as parameters
        score_name: label to name column with result of the score function
    returns:
        rule_df: copy of mined_rules_df, with the additional column:
            score_name: result of the score function
            Sorted in descending order by score_name
    """
    rule_df = pd.DataFrame(mined_rules_df.copy())
    rule_df[score_name] = rule_df["ct"].apply(lambda ct: score_fn(*ct))
    return rule_df.sort_values(by=score_name, ascending=False)

"""
RECOMMENDATION ALGORITHMS
"""

def collaborative_filter(s_train, s_input, j=3, sim_fn=None, threshold=0.01, k=10):
    """
    Collaborative filtering recommender system.
    params:
        s_train: list of sets of liked item indices for train data
        s_input: list of sets of liked item indices for input data
        j: number of similar users to base recommendations on
        sim_fn(u, v): function that returns a float value representing
            the similarity between sets u and v
        threshold: minimum similarity required to consider a similar user
        k: number of items to recommend for each user
    returns:
        recs_pred: list of lists of tuples of recommendations where
            each tuple has (item index, relevance score) with the list
            of tuples sorted in order of decreasing relevance
    """
    if sim_fn is None:
        raise ValueError("Must specify a similarity function.")
    check_list_of_sets(s_train, "s_train")
    check_list_of_sets(s_input, "s_input")
    recs_pred = []
    for src in s_input:
        users = []
        for vec in s_train:
            sim = sim_fn(src, vec)
            if sim >= threshold:
                users.append((sim, vec))
        j_users = min(len(users), j)
        if j_users > 0:
            top_users = sorted(users, key=lambda p: p[0], reverse=True)
            sim_user_sets = [vec for (sim, vec) in top_users[0:j_users]]
            opts = dict()
            for user_set in sim_user_sets:
                for item in user_set:
                    if item not in src:
                        if item not in opts:
                            opts[item] = 0
                        opts[item] += 1
            ranks = []
            for item in opts:
                ranks.append((item, opts[item] / len(sim_user_sets)))
            top_ranks = sorted(ranks, key=lambda p: p[1], reverse=True)
            k_recs = min(len(top_ranks), k)
            recs = top_ranks[0:k_recs]
            recs_pred.append(recs)
        else:
            recs_pred.append([])
    return recs_pred

def content_filter(items_train, s_input, sim_fn=None, threshold=0.01, k=10):
    """
    Content-based filtering recommender system.
    params:
        items_train: list of sets of non-zero attribute indices for items
        s_input: list of sets of liked item indices for input data
        sim_fn(u, v): function that returns a float value representing
            the similarity between sets u and v
        threshold: minimum similarity required to consider a similar item
        k: number of items to recommend for each user
    returns:
        recs_pred: list of lists of tuples of recommendations where
            each tuple has (item index, relevance score) with the list
            of tuples sorted in order of decreasing relevance
    """
    if sim_fn is None:
        raise ValueError("Must specify a similarity function.")
    check_list_of_sets(items_train, "items_train")
    check_list_of_sets(s_input, "s_input")
    recs_pred = []
    for src in s_input:
        sim_items = []
        for item_new_idx, item_new in enumerate(items_train):
            total_sim = 0
            for item_liked_idx in src:
                item_liked = items_train[item_liked_idx]
                sim = sim_fn(item_new, item_liked)
                total_sim += sim
            mean_sim = total_sim / len(src)
            if mean_sim > 0:
                sim_items.append((item_new_idx, mean_sim))
        top_ranks = sorted(sim_items, key=lambda p: p[1], reverse=True)
        k_recs = min(len(top_ranks), k)
        recs = top_ranks[0:k_recs]
        recs_pred.append(recs)
    return recs_pred

def svd_filter(m_train, m_input, n_factors=2, baselines=None, threshold=0.01, k=10):
    """
    Matrix factorization recommender system using Singular Value Decomposition (SVD).
    params:
        m_train: matrix of train data, rows = users, columns = items, 1 = like, 0 otherwise
        m_input: matrix of input data, rows = users, columns = items, 1 = like, 0 otherwise
        n_factors: number of latent factors to estimate (default: 2)
        baselines: numpy array of values to fill empty input entries, array length
            equal to the number of columns in `m_train` and `m_input` (default: None)
        threshold: minimum score to qualify as a recommended item
        k: number of items to recommend for each user
        sim_fn(u, v): function that returns a float value representing
            the similarity between sets u and v
    returns:
        recs_pred: list of lists of tuples of recommendations where
            each tuple has (item index, relevance score) with the list
            of tuples sorted in order of decreasing relevance
        (u, s, vt): tuple of matrix factors produced by SVD:
            u: latent user matrix, rows = users, columns = latent factors
            s: array of latent factor weights
            vt: transposed latent item matrix, rows = latent factors, columns = items
            To estimate the matrix, compute: `B = u.dot(np.diag(s)).dot(vt)`
    """
    m_input_filled = m_input
    if baselines is not None:
        m_input_filled = m_input + np.vstack([baselines for i in range(len(m_input))])
        m_input_filled = np.clip(m_input_filled, a_min=0, a_max=1)
    m_all = np.vstack([m_train, m_input_filled])
    A = csc_matrix(m_all, dtype=float)
    u, s, vt = svds(A, k=n_factors)
    B = u.dot(np.diag(s)).dot(vt)
    s_b_test = []
    row_base = len(m_train)
    for row_add in range(len(m_input)):
        inp = m_input[row_add]
        row_id = row_base + row_add
        row = B[row_id]
        rec_scores = []
        for col, (score, orig) in enumerate(zip(row, inp)):
            if orig < 1:
                if score >= threshold:
                    rec_scores.append((col, score))
        ranks = sorted(rec_scores, key=lambda p: p[1], reverse=True)
        s_b_test.append(ranks[0:k])
    return s_b_test, (u, s, vt)

def association_filter(rule_df, m_train, s_input, score_fn=None, min_score=0.01, k=10):
    """
    Association rule recommender system using frequent itemsets.
    params:
        rule_df: Pandas dataframe of association rules to use, with columns:
            "a": antecedent (LHS) item index
            "b": consequent (RHS) item index
            "ct": tuple of contingency table entries for the rule
        m_train: matrix of train data, rows = users, columns = items, 1 = like, 0 otherwise
        s_input: list of sets of liked item indices for input data
        score_fn(f11, f10, f01, f11): function for scoring rules, takes the entries
            of the contingency table as parameters
        min_score: minimum result from score function required to use an association rule
        k: number of items to recommend for each user
    returns:
        top_rules_df: copy of rule_df, with the additional column:
            "score": result of the score function
            Sorted in descending order by "score"
    """
    if score_fn is None:
        raise ValueError("Must specify a scoring function for association rules.")
    score_name = "score"
    ranked_rules = rank_association_rules(
        rule_df, score_fn=score_fn, score_name=score_name
    )
    top_rules_df = ranked_rules.query("{} >= {}".format(score_name, min_score))
    rule_records = top_rules_df.to_dict(orient="records")
    all_recs = []
    for likes in s_input:
        rec_map = {}
        for rule in rule_records:
            if rule["a"] in likes and rule["b"] not in likes:
                if rule["b"] not in rec_map:
                    rec_map[rule["b"]] = 0
                rec_map[rule["b"]] += rule[score_name]
        ranks = sorted(rec_map.items(), key=lambda p: p[1], reverse=True)
        all_recs.append(ranks[0:k])
    return all_recs, top_rules_df

def weighted_hybrid(components, use_ranks=False, k=10):
    """
    Hybrid recommender system using weights.
    params:
        components: list of tuples where each tuple has (recs list, weight)
            where recs_list is a list of tuples where each tuple has
            (item index, relevance score) and weight is the factor used to
            scale the relevance scores for this component
        use_ranks: boolean (default: False), if True, apply weights to the
            inverse rank of each item instead of its score, for example:
            the first recommended item will earn 1 * weight, the second will
            earn 1/2 * weight, the third will earn 1/3 * weight and so on
        k: number of items to recommend for each user
    returns:
        recs_pred: list of lists of tuples of recommendations where
            each tuple has (item index, relevance score) with the list
            of tuples sorted in order of decreasing relevance
    """
    if len(components) < 1:
        raise ValueError("Must provide at least one component.")
    n_users = len(components[0][0])
    user_recs = []
    user_item_maps = [{} for i in range(n_users)]
    for (recs_list, weight) in components:
        if len(recs_list) != n_users:
            raise ValueError(
                "Length of components do not match. Should be equal to number of users."
            )
        for user_id, recs in enumerate(recs_list):
            item_map = user_item_maps[user_id]
            for i, (item, score) in enumerate(recs):
                if item not in item_map:
                    item_map[item] = 0
                if use_ranks:
                    rank = i + 1
                    inv_rank = 1 / rank
                    item_map[item] += weight * inv_rank
                else:
                    item_map[item] += weight * score
    for item_map in user_item_maps:
        item_scores = item_map.items()
        ranks = sorted(item_scores, key=lambda p: p[1], reverse=True)
        top_recs = ranks[0:k]
        user_recs.append(top_recs)
    return user_recs

max_stars_repo_name: acmillinoistech/redcarpet

max_issues_repo_name: acmillinoistech/redcarpet

max_forks_repo_name: acmillinoistech/redcarpet


text:
import time
from typing import TypeVar, Union, Set

import numpy as np

from rule_models.ids.objective_function.ids_objective_function_abstract import AbstractIDSObjectiveFunction

from pyids.data_structures import IDSRuleSet
from pyids.data_structures.ids_objective_function import ObjectiveFunctionParameters

from submodmax.abstract_optimizer import AbstractSubmodularFunction

E = TypeVar('E')

class IDSObjectiveFunction(AbstractSubmodularFunction, AbstractIDSObjectiveFunction):
    def __init__(self, objective_func_params=ObjectiveFunctionParameters(), cacher=None, scale_factor=1,
                 normalize=True):
        AbstractIDSObjectiveFunction.__init__(self, objective_func_params, cacher=cacher, scale_factor=scale_factor,
                                              normalize=normalize)

    def f1_minimize_total_nb_of_literals(self, solution_set):
        f1_unnormalized = self.f1_upper_bound_nb_of_literals - solution_set.sum_rule_length()
        if self.normalize:
            f1 = f1_unnormalized / self.f1_upper_bound_nb_of_literals
        else:
            f1 = f1_unnormalized
        self._f1_boundary_check(f1)
        return f1

    def f2_minimize_overlap_predicting_the_same_class(self, solution_set):
        overlap_intraclass_sum = 0

        for i, r1 in enumerate(solution_set.ruleset):
            for j, r2 in enumerate(solution_set.ruleset):
                if i >= j:
                    continue

                if r1.car.consequent.value == r2.car.consequent.value:
                    overlap_tmp = self.cacher.overlap(r1, r2)

                    overlap_intraclass_sum += overlap_tmp

        f2_unnormalized = self.f2_f3_upper_bound - overlap_intraclass_sum

        if self.normalize:
            f2 = f2_unnormalized / self.f2_f3_upper_bound
        else:
            f2 = f2_unnormalized

        self._boundary_check(f2, 'f2')
        return f2

    def f3_minimize_overlap_predicting_different_class(self, solution_set):
        overlap_interclass_sum = 0

        for i, r1 in enumerate(solution_set.ruleset):
            for j, r2 in enumerate(solution_set.ruleset):
                if i >= j:
                    continue

                if r1.car.consequent.value != r2.car.consequent.value:
                    overlap_tmp = self.cacher.overlap(r1, r2)

                    overlap_interclass_sum += overlap_tmp

        f3_unnormalized = self.f2_f3_upper_bound - overlap_interclass_sum

        if self.normalize:
            f3 = f3_unnormalized / self.f2_f3_upper_bound
        else:
            f3 = f3_unnormalized
        self._boundary_check(f3, 'f3')
        return f3

    def f4_at_least_one_rule_per_target_value(self, solution_set):
        classes_covered = set()

        for rule in solution_set.ruleset:
            classes_covered.add(rule.car.consequent.value)

        f4_unnormalized = len(classes_covered)

        if self.normalize:
            f4 = f4_unnormalized / self.nb_of_target_values
        else:
            f4 = f4_unnormalized
        self._boundary_check(f4, 'f4')
        return f4

    def f5_minimize_incorrect_cover(self, solution_set):
        sum_incorrect_cover = 0

        for rule in solution_set.ruleset:

            incorrect_cover_size = np.sum(rule._incorrect_cover(self.quant_dataframe))
            # incorrect_cover_size = np.sum(rule.incorrect_cover(self.quant_dataframe))
            # print(f"IDS incorrect cover size: {incorrect_cover_size} for rule {rule}")
            sum_incorrect_cover += incorrect_cover_size

        # print(f"IDS f5 upper bound: {self.f5_upper_bound}")
        # print(f"IDS f5 sum incorrect cover: {sum_incorrect_cover}")
        f5_unnormalized = self.f5_upper_bound - sum_incorrect_cover
        if self.normalize:
            f5 = f5_unnormalized / self.f5_upper_bound
        else:
            f5 = f5_unnormalized
        self._boundary_check(f5, 'f5')
        return f5

    def f6_cover_each_example(self, solution_set):
        correctly_covered = np.zeros(self.nb_of_training_examples).astype(bool)

        for rule in solution_set.ruleset:
            correctly_covered = correctly_covered | rule.correct_cover(self.quant_dataframe)

        f6_unnormalized = np.sum(correctly_covered)
        if self.normalize:
            f6 = f6_unnormalized / self.nb_of_training_examples
        else:
            f6 = f6_unnormalized
        self._boundary_check(f6, 'f6')
        return f6

    def evaluate(self, solution_set: Union[IDSRuleSet, Set[E]]) -> float:
        if type(solution_set) == set:
            solution_set = IDSRuleSet(solution_set)

        if type(solution_set) != IDSRuleSet:
            raise Exception("Type of solution_set must be IDSRuleSet, but is ", type(solution_set))

        self.call_counter += 1
        self.call_set_sizes.append(len(solution_set))
        start_time = time.time()

        l = self.objective_func_params.params["lambda_array"]

        f0 = self.f0_minimize_rule_set_size(self.ground_set_size, len(solution_set)) if l[0] != 0 else 0
        f1 = self.f1_minimize_total_nb_of_literals(solution_set) if l[1] != 0 else 0
        f2 = self.f2_minimize_overlap_predicting_the_same_class(solution_set) if l[2] != 0 else 0
        f3 = self.f3_minimize_overlap_predicting_different_class(solution_set) if l[3] != 0 else 0
        f4 = self.f4_at_least_one_rule_per_target_value(solution_set) if l[4] != 0 else 0
        f5 = self.f5_minimize_incorrect_cover(solution_set) if l[5] != 0 else 0
        f6 = self.f6_cover_each_example(solution_set) if l[6] != 0 else 0

        self.f0_val = f0
        self.f1_val = f1
        self.f2_val = f2
        self.f3_val = f3
        self.f4_val = f4
        self.f5_val = f5
        self.f6_val = f6

        # print(f"IDS f1:{f1}")

        # print()
        # print(tabulate([['value', f0, f1, f2, f3, f4, f5, f6],
        #                 ['l*val', f0 * l[0], f1 * l[1], f2 * l[2], f3 * l[3], f4 * l[4], f5 * l[5], f6 * l[6]]
        #                 ],
        #                headers=['type', 'f0', 'f1', 'f2', 'f3', 'f4', 'f5', 'f6']))
        # print()

        fs = np.array([
            f0, f1, f2, f3, f4, f5, f6
        ]) / self.scale_factor

        result = np.dot(l, fs)

        if self.stat_collector is not None:
            self.stat_collector.add_values(f0, f1, f2, f3, f4, f5, f6, result)

        end_time = time.time()
        elapsed_time = end_time - start_time
        self.call_run_times.append(elapsed_time)

        return result

    def f0(self, solution_set):
        current_nb_of_rules: int = len(solution_set)
        ground_set_size = len(self.objective_func_params.params['all_rules'])
        return self.f0_minimize_rule_set_size(ground_set_size=ground_set_size,
                                              current_nb_of_rules=current_nb_of_rules)

    def f1(self, solution_set):
        return self.f1_minimize_total_nb_of_literals(solution_set)

    def f2(self, solution_set):
        return self.f2_minimize_overlap_predicting_the_same_class(solution_set)

    def f3(self, solution_set):
        return self.f3_minimize_overlap_predicting_different_class(solution_set)

    def f4(self, solution_set):
        return self.f4_at_least_one_rule_per_target_value(solution_set)

    def f5(self, solution_set):
        return self.f5_minimize_incorrect_cover(solution_set)

    def f6(self, solution_set):
        return self.f6_cover_each_example(solution_set)


text:
from flask import Flask, render_template, request, make_response, jsonify
import os
import numpy as np
import torch
import  soundfile as sf
from transformers import Wav2Vec2ForCTC, Wav2Vec2Processor, Wav2Vec2CTCTokenizer
from werkzeug.utils import secure_filename
import librosa
import subprocess

from wav2vec2_kenlm.decoder import *
import wav2vec2_kenlm.utils
import numpy as np

device = 'cuda' if torch.cuda.is_available() else 'cpu'
processor = Wav2Vec2Processor.from_pretrained("./wav2vec2-large-xlsr-min/processor")#Load Processor(Input)
model = Wav2Vec2ForCTC.from_pretrained("./wav2vec2-large-xlsr-min/model/checkpoint-37100")#Load Model(Input)
model.to(device)

lm_path = "./3-gram-tat.arpa"
vocab_dict = processor.tokenizer.get_vocab()
sort_vocab = sorted((value, key) for (key,value) in vocab_dict.items())
vocab = []
for _, token in sort_vocab:
    vocab.append(token)
vocab[vocab.index(processor.tokenizer.word_delimiter_token)] = ' '

alpha=0.9
beta=0.9
beam_width=1024
beam_decoder = BeamCTCDecoder(vocab[:-2], lm_path=lm_path,
                alpha=alpha, beta=beta,
                cutoff_top_n=40, cutoff_prob=1.0,
                beam_width=beam_width, num_processes=16,
                blank_index=vocab.index(processor.tokenizer.pad_token))

# bulid and load config webserver
UPLOAD_FOLDER = './static/tmpfile'
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = 64 * 1024 * 1024  # 64MB

# Predicte
def pred(audio):
    # 採樣率問題所以我改用librosa, 這樣輸入就不用管採樣率
	speech, sr = librosa.load(audio, sr=16000, mono=True)

	input_values = processor(speech, sampling_rate=sr, return_tensors="pt").input_values.to(device)

	with torch.no_grad():
	  logits = model(input_values).logits

	pred_ids = torch.argmax(logits, dim=-1)
	pred_str = processor.batch_decode(pred_ids)[0]

	return pred_str

def decode_kenlm(audio):
    speech, sr = librosa.load(audio, sr = 16000, mono = True)
    input_values = processor(speech, sampling_rate=sr, return_tensors="pt").input_values.to(device)
    with torch.no_grad():
        logits = model(input_values).logits
    pred_ids = torch.argmax(logits, dim=-1)
    pred_str = processor.batch_decode(pred_ids)[0]
    beam, beam_decoded_offsets = beam_decoder.decode(logits)
    return pred_str, beam[0][0]

@app.route("/")
def index():
    return render_template("index.html", res_text="")

@app.route("/predict",  methods=["POST"])
def predict():
    file = request.files['file_loader']
    filename = secure_filename(file.filename)
    file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    if filename[-4:] != ".wav":
        ffmpeg = " ".join(
            ["ffmpeg", "-i", os.path.join(app.config['UPLOAD_FOLDER'], filename), os.path.join(app.config['UPLOAD_FOLDER'], filename[:-4] + ".wav"), "-y"]
        )
        print(ffmpeg)
        subprocess.run(ffmpeg, shell=True)
        filename = filename[:-4] + ".wav"

    text, text2 = decode_kenlm(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    return render_template("index.html", rt=1,res_text=text, res_text2=text2, audio_path=os.path.join(app.config['UPLOAD_FOLDER'], filename))

@app.route("/test")
def test():
    return render_template("test.html", res_text="")

def main():
    app.run(debug=True, host="::", port=5010)

if __name__ == "__main__":
    main()


text:
import unittest
from numba.tests.support import TestCase, run_in_subprocess

class TestNumbaImport(TestCase):
    """
    Test behaviour of importing Numba.
    """

    def test_laziness(self):
        """
        Importing top-level numba features should not import too many modules.
        """
        # A heuristic set of modules that shouldn't be imported immediately
        banlist = ['cffi',
                   'distutils',
                   'numba.cuda',
                   'numba.cpython.mathimpl',
                   'numba.cpython.randomimpl',
                   'numba.tests',
                   'numba.core.typing.collections',
                   'numba.core.typing.listdecl',
                   'numba.core.typing.npdatetime',
                   ]
        # Sanity check the modules still exist...
        for mod in banlist:
            if mod not in ('cffi',):
                __import__(mod)

        code = """if 1:
            from numba import jit, vectorize
            from numba.core import types
            import sys
            print(list(sys.modules))
            """

        out, _ = run_in_subprocess(code)
        modlist = set(eval(out.strip()))
        unexpected = set(banlist) & set(modlist)
        self.assertFalse(unexpected, "some modules unexpectedly imported")

    def test_no_impl_import(self):
        """
        Tests that importing jit does not trigger import of modules containing
        lowering implementations that would likely install things in the
        builtins registry and have side effects impacting other targets
        """
        # None of these modules should be imported through the process of
        # doing 'import numba' or 'from numba import njit'
        banlist = ['numba.cpython.slicing',
                   'numba.cpython.tupleobj',
                   'numba.cpython.enumimpl',
                   'numba.cpython.hashing',
                   'numba.cpython.heapq',
                   'numba.cpython.iterators',
                   'numba.cpython.numbers',
                   'numba.cpython.rangeobj',
                   'numba.cpython.cmathimpl',
                   'numba.cpython.mathimpl',
                   'numba.cpython.printimpl',
                   'numba.cpython.randomimpl',
                   'numba.core.optional',
                   'numba.misc.gdb_hook',
                   'numba.misc.literal',
                   'numba.misc.cffiimpl',
                   'numba.np.linalg',
                   'numba.np.polynomial',
                   'numba.np.arraymath',
                   'numba.np.npdatetime',
                   'numba.np.npyimpl',
                   'numba.typed.typeddict',
                   'numba.typed.typedlist',
                   'numba.experimental.jitclass.base',]

        code1 = """if 1:
            import sys
            import numba
            print(list(sys.modules))
            """

        code2 = """if 1:
            import sys
            from numba import njit
            @njit
            def foo():
                pass
            print(list(sys.modules))
            """

        for code in (code1, code2):
            out, _ = run_in_subprocess(code)
            modlist = set(eval(out.strip()))
            unexpected = set(banlist) & set(modlist)
            self.assertFalse(unexpected, "some modules unexpectedly imported")

    def test_no_accidental_warnings(self):
        # checks that importing Numba isn't accidentally triggering warnings due
        # to e.g. deprecated use of import locations from Python's stdlib
        code = "import numba"
        # See: https://github.com/numba/numba/issues/6831
        # bug in setuptools/packaging causing a deprecation warning
        flags = ["-Werror", "-Wignore::DeprecationWarning:packaging.version:"]
        run_in_subprocess(code, flags)

    def test_import_star(self):
        # checks that "from numba import *" works.
        code = "from numba import *"
        run_in_subprocess(code)

if __name__ == '__main__':
    unittest.main()

max_stars_repo_name: auderson/numba

max_issues_repo_name: auderson/numba

max_forks_repo_name: auderson/numba

reason: import numba,from numba


text:
"""
Scene graph evaluator wrapper for Detectron2.

Modified to be compatible with Detectron2 based on
https://github.com/facebookresearch/detectron2/blob/main/detectron2/evaluation/coco_evaluation.py
"""
__author__ = "Hengyue Liu"
__copyright__ = "Copyright (c) 2021 Futurewei Inc."
__credits__ = ["Detectron2"]
__license__ = "MIT License"
__version__ = "0.1"
__maintainer__ = "Hengyue Liu"
__email__ = "onehothenry@gmail.com"

import contextlib
import copy
import io
import itertools
import json
import logging
import numpy as np
import os
import pickle
from collections import OrderedDict
import pycocotools.mask as mask_util
import torch
from fvcore.common.file_io import PathManager
from pycocotools.coco import COCO
from tabulate import tabulate

import detectron2.utils.comm as comm
from detectron2.data import MetadataCatalog
from detectron2.structures import Boxes, BoxMode, pairwise_iou
from detectron2.utils.logger import create_small_table
from detectron2.evaluation.evaluator import DatasetEvaluator
from fcsgg.data.datasets.visual_genome import convert_to_vg_json
from .sgg_evaluation import SceneGraphAPI

class VGEvaluator(DatasetEvaluator):
    def __init__(self, dataset_name, cfg, distributed, output_dir=None):
        self._tasks = self._tasks_from_config(cfg)
        self._distributed = distributed
        self._output_dir = output_dir

        self._cpu_device = torch.device("cpu")
        # looger parent is fcsgg, which is not set therefore get it from detectron2
        self._logger = logging.getLogger("detectron2.evaluation")

        self._metadata = MetadataCatalog.get(dataset_name)
        if not hasattr(self._metadata, "sg_json_file"):
            self._logger.info(
                f"'{dataset_name}' has no json_file for scene graph."
                " Therefore trying to convert it to VG format ..."
            )

            cache_path = os.path.join(output_dir, f"{dataset_name}_sg_format.json")
            self._metadata.sg_json_file = cache_path
            convert_to_vg_json(dataset_name, cache_path, allow_cached=True)

        sg_json_file = PathManager.get_local_path(self._metadata.sg_json_file)
        with contextlib.redirect_stdout(io.StringIO()):
            self._sg_api = SceneGraphAPI(cfg, sg_json_file, logger=self._logger)

        self._kpt_oks_sigmas = cfg.TEST.KEYPOINT_OKS_SIGMAS
        # Test set json files do not contain annotations (evaluation must be
        # performed using the COCO evaluation server).
        self._do_evaluation = "annotations" in self._sg_api.dataset

    def reset(self):
        self._predictions = []

    def _tasks_from_config(self, cfg):
        """
        Returns:
            tuple[str]: tasks that can be evaluated under the given configuration.
        """
        tasks = ()
        if cfg.RELATION.RELATION_ON:
            if cfg.RELATION.USE_GT_BOX:
                tasks = tasks + ("sgcls",)
            elif cfg.RELATION.USE_GT_OBJECT_LABEL:
                tasks = tasks + ("predcls",)
            else:
                tasks = tasks + ("sgdet",)
        assert len(tasks) < 2, "For SGG evaluation, each task needs separate evaluation."
        return tasks

    def process(self, inputs, outputs):
        """
        Args:
            inputs: the inputs to a COCO model (e.g., GeneralizedRCNN).
                It is a list of dict. Each dict corresponds to an image and
                contains keys like "height", "width", "file_name", "image_id".
            outputs: the outputs of a COCO model. It is a list of dicts with key
                "instances" that contains :class:`Instances`.
        """
        for input, output in zip(inputs, outputs):
            prediction = {"image_id": input["image_id"]}

            assert "scene_graph" in output, "No scene_graph is found in outputs."
            scene_graph = output["scene_graph"].to(self._cpu_device)
            prediction.update(scenegraph_to_json(scene_graph))

            self._predictions.append(prediction)

    def evaluate(self):
        if self._distributed:
            comm.synchronize()
            predictions = comm.gather(self._predictions, dst=0)
            predictions = list(itertools.chain(*predictions))

            if not comm.is_main_process():
                return {}
        else:
            predictions = self._predictions

        if len(predictions) == 0:
            self._logger.warning("[VGEvaluator] Did not receive valid predictions.")
            return {}

        if self._output_dir:
            PathManager.mkdirs(self._output_dir)
            file_path = os.path.join(self._output_dir, "sg_predictions.pth")
            self._logger.info("Saving SGG predictions to {}".format(file_path))
            with PathManager.open(file_path, "wb") as f:
                torch.save(predictions, f)

        self._results = OrderedDict()
        self._eval_predictions(set(self._tasks), predictions)
        # Copy so the caller can do whatever with results
        return copy.deepcopy(self._results)

    def _eval_predictions(self, tasks, predictions):
        """
        Evaluate predictions on the given tasks.
        Fill self._results with the metrics of the tasks.
        """
        self._logger.info("Preparing results for VG format ...")

        if self._output_dir:
            file_path = os.path.join(self._output_dir, "vg_sg_results.json")
            self._logger.info("Saving results to {}".format(file_path))
            with PathManager.open(file_path, "w") as f:
                f.write(json.dumps(predictions))
                f.flush()

        if not self._do_evaluation:
            self._logger.info("Annotations are not available for evaluation.")
            return

        self._logger.info("Evaluating predictions ...")
        for task in sorted(tasks):
            res = self._sg_api.evaluate(predictions, task)
            self._results[task] = res

def scenegraph_to_json(sg):
    """
    Dump an "SceneGraph" object to a dict that's used for evaluation.
    The output will be saved as json.

    Args:
        sg (SceneGraph):

    Returns:
        dict: contains predictions for one image.
    """
    boxes = sg.pred_boxes.tensor.numpy().tolist()
    # for vg evaluation, all boxes should be in XYXY_ABS
    scores = sg.scores.numpy().tolist()
    classes = sg.pred_classes.numpy().tolist()
    rel_scores = sg.rel_scores.numpy().tolist()
    rel_inds = sg.rel_inds.numpy().tolist()

    result = {
        "category_ids": classes,
        "bboxes": boxes,
        "scores": scores,
        "rel_scores": rel_scores,
        "rel_inds": rel_inds
    }
    return result

max_stars_repo_name: liuhengyue/fcsgg

max_issues_repo_name: liuhengyue/fcsgg

max_forks_repo_name: liuhengyue/fcsgg