text:
PATH_ROOT='C:/Users/Louis Owen/Desktop/ICoDSA 2020/SENN/'

print('==================== Importing Packages ====================')

import warnings
warnings.filterwarnings("ignore", category=FutureWarning)

import pandas as pd
import re
import json
import math
import string
import numpy as np
from bs4 import BeautifulSoup
import gensim

import contractions
import inflect
import string

import stanfordnlp
stanfordnlp.download('en')
import nltk
nltk.download('wordnet')
nltk.download('punkt')
nltk.download('averaged_perceptron_tagger')
nltk.download('maxent_ne_chunker')
nltk.download('words')

from nltk import pos_tag
from nltk.corpus import wordnet as wn
from nltk.tokenize import wordpunct_tokenize,TweetTokenizer
from nltk import word_tokenize, pos_tag, ne_chunk
from nltk.stem.porter import *
from nltk.stem import WordNetLemmatizer
from scipy import sparse as sp
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
from sklearn.impute import KNNImputer
from textblob import TextBlob
from sklearn.preprocessing import StandardScaler,MinMaxScaler

#------------------------------------------------------------------------------------------
print("==================== Importing Supporting Files ====================")

#RF N-Grams Dictionary
rf_1_gram_df = pd.read_csv(PATH_ROOT+'Supporting_Files/'+'rf_1_gram_df.csv', engine='python',index_col='Unnamed: 0')
rf_2_gram_df = pd.read_csv(PATH_ROOT+'Supporting_Files/'+'rf_2_gram_df.csv', engine='python',index_col='Unnamed: 0')
rf_3_gram_df = pd.read_csv(PATH_ROOT+'Supporting_Files/'+'rf_3_gram_df.csv', engine='python',index_col='Unnamed: 0')
rf_4_gram_df = pd.read_csv(PATH_ROOT+'Supporting_Files/'+'rf_4_gram_df.csv', engine='python',index_col='Unnamed: 0')

#PMI Dictionary
PMI_df = pd.read_csv(PATH_ROOT+'Supporting_Files/'+'PMI_df.csv', engine='python',index_col='Unnamed: 0')

#Abbreviation & Slang Dict
abb_df = pd.read_table(PATH_ROOT+'Supporting_Files/'+'emnlp_dict.txt', sep='\s+', names=('Abbreviation', 'Normal'))
abb=pd.Series(abb_df['Normal'])
abb.index=abb_df['Abbreviation']
abb_dict=dict(abb)

#AFINN Sentiment Lexicon
AFINN_df = pd.read_table(PATH_ROOT+'Supporting_Files/'+'AFINN/'+'AFINN-111.txt', names=('Word', 'Sentiment'))
AFINN=pd.Series(AFINN_df['Sentiment'])
AFINN=((AFINN-AFINN.min())/(AFINN.max()-AFINN.min()))*(1-(-1))+(-1) #Rescaling in [-1,1]
AFINN.index=AFINN_df['Word']
AFINN_dict=dict(AFINN)

#Bing-Liu Sentiment Lexicon
pos = pd.read_table(PATH_ROOT+'Supporting_Files/'+'Bing-Liu-opinion-lexicon-English/'+'positive-words.txt',names='P')
neg=pd.read_table(PATH_ROOT+'Supporting_Files/'+'Bing-Liu-opinion-lexicon-English/'+'negative-words.txt',names='N',encoding='latin-1')
BingLiu_dict={'pos':pos['P'].tolist(),'neg':neg['N'].tolist()}

#General Enquirer Sentiment Lexicon
General_Inquirer_df=pd.read_csv(PATH_ROOT+'Supporting_Files/'+'General Inquirer Lexicon/'+'inquirerbasic.csv',index_col='Entry')
General_Inquirer_df=General_Inquirer_df[['Positiv','Negativ']]
General_Inquirer_dict={'pos':General_Inquirer_df[pd.isnull(General_Inquirer_df['Positiv'])==False]['Positiv'].index.tolist(),
                      'neg':General_Inquirer_df[pd.isnull(General_Inquirer_df['Negativ'])==False]['Positiv'].index.tolist()}

#NRC Hashtag Sentiment Lexicon
hs=pd.read_table(PATH_ROOT+'Supporting_Files/'+'NRC-Sentiment-Emotion-Lexicons/AutomaticallyGeneratedLexicons/NRC-Hashtag-Sentiment-Lexicon-v1.0/'+'HS-unigrams.txt',names=('Hashtag','PMI(w, pos) -PMI(w, neg)','n_pos','n_neg'),encoding='latin-1')
hs=hs[pd.isnull(hs.Hashtag)==False]
hs['PMI(w, pos) -PMI(w, neg)']=((hs['PMI(w, pos) -PMI(w, neg)']-hs['PMI(w, pos) -PMI(w, neg)'].min())/(hs['PMI(w, pos) -PMI(w, neg)'].max()-hs['PMI(w, pos) -PMI(w, neg)'].min()))*(1-(-1))+(-1) #Rescaling in [-1,1]
nrc=hs['PMI(w, pos) -PMI(w, neg)']
nrc.index=hs['Hashtag']
NRC_hashtag_dict=dict(nrc)

#Sentiwordnet Sentiment Lexicon
sentiwordnet=pd.read_table(PATH_ROOT+'Supporting_Files/'+'SentiWordNet/'+'SentiWordNet_3.0.0.txt',names=('POS','ID','PosScore','NegScore','SynsetTerms','Gloss'),encoding='latin-1')
sentiwordnet=sentiwordnet[pd.isnull(sentiwordnet.POS)==False]
sentiwordnet['score']=sentiwordnet['PosScore']-sentiwordnet['NegScore']

#------------------------------------------------------------------------------------------
print("==================== Importing Data ====================")

df_stocktwits_full_BA = pd.read_csv(PATH_ROOT+'Dataset/'+'df_stocktwits_full_BA.csv', engine='python')
df_stocktwits_full_BA=df_stocktwits_full_BA[~pd.isnull(df_stocktwits_full_BA.id)]

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

def tokenize(sentence):
	'''
	tokenize input sentence into token
	'''
	token_list=nltk.regexp_tokenize(sentence, pattern=r"\s|[\.,;]\D", gaps=True)
	return(token_list)

def clean_data(concat_df):

	print("==================== Cleaning Data ====================")

	#Inspired by:
	# https://www.dotnetperls.com/punctuation-python
	# https://github.com/tthustla/twitter_sentiment_analysis_part1/blob/master/Capstone_part2.ipynb
	# https://github.com/Deffro/text-preprocessing-techniques/blob/master/techniques.py

	#Word ordinal encoding
	p = inflect.engine()
	word_to_number_mapping = {}
	for i in range(1, 2000):
		word_form = p.number_to_words(i)  # 1 -> 'one'
		ordinal_word = p.ordinal(word_form)  # 'one' -> 'first'
		ordinal_number = p.ordinal(i)  # 1 -> '1st'
		word_to_number_mapping[ordinal_word] = ordinal_number  # 'first': '1st'

	def elongated_word(word):
		"""
	  	Replaces an elongated word with its basic form, unless the word exists in the lexicon
	  	"""
		repeat_regexp = re.compile(r'(\w*)(\w)\2(\w*)')
		repl = r'\1\2\3'
		if (len(word)>2 and word[0] != '$'):#if not Stock Market symbol
			if wn.synsets(word):
				return word
			repl_word = repeat_regexp.sub(repl, word)
			if repl_word != word:
				return elongated_word(repl_word)
			else:
				return repl_word
		else:
			return word

	def isfloat(value):
		'''
		Check if value is float or not
		'''
		try:
			float(value)
			return True
		except ValueError:
			return False

	def deEmojify(inputString):
		'''
		Remove Emoji
		'''
		return inputString.encode('ascii', 'ignore').decode('ascii')

	def sentences_cleaner(sentence):
		'''
		clean input sentence
		'''
		try:
			mention_pat= r'@[A-Za-z0-9_]+'
			mention_2_pat=r'@[A-Za-z0-9_]+:\s'
			retweet_pat=r'^RT +'
			dollars_pat=r'\$ +'
			http_pat = r'https?://[^ ]+'
			www_pat = r'www.[^ ]+'
			apos_pat=r'"+|"$|"+"$'
			ticker_pat=r'\$[A-Za]+ '

			#Transform any url into '_url'
			sentence = re.sub(http_pat, '_url', sentence)
			sentence = re.sub(www_pat, '_url', sentence)

			#Delete Ticker
			sentence = re.sub(ticker_pat,"", sentence)

			#HTML decoding remove BOM
			soup = BeautifulSoup(sentence, 'lxml')
			souped = soup.get_text()
			try:
				bom_removed = souped.decode("utf-8-sig").replace(u"\ufffd", "?")
			except:
				bom_removed = souped

			#Delete Emoji
			stripped=deEmojify(bom_removed)

			#Delete mention
			stripped = re.sub(mention_2_pat,"", stripped)
			stripped = re.sub(mention_pat,"", stripped)

			#Delete retweet
			stripped=re.sub(retweet_pat,"",stripped)

			#Transfrom abbreviation & slang word  into normal words based on abb_dict corpus
			abbreviation_handled=' '.join(pd.Series(stripped.split()).apply(lambda x: abb_dict[x] if x in abb_dict.keys() else x).to_list())

			#Transform contracted words into normal words
			contraction_handled =contractions.fix(abbreviation_handled)

			#Join the stock symbol
			dollars_handled=re.sub(dollars_pat,'$',contraction_handled)

			#Transform elongated words into normal words
			elongated_handled=' '.join(pd.Series(dollars_handled.split()).apply(lambda x: elongated_word(x[:-1])+x[-1] if (x[-1] in string.punctuation and not isfloat(x)) else elongated_word(x) if not isfloat(x) else x))

			#Transform ordinal number
			ordinal_handled=' '.join(pd.Series(elongated_handled.split()).apply(lambda x: word_to_number_mapping[x.lower()] if x.lower() in word_to_number_mapping.keys() else x))

			#Remove unnecesary apostrophes
			apos_handled=re.sub(apos_pat,'',ordinal_handled)

			#Split Last Word Punctuation
			wordpunct=wordpunct_tokenize(apos_handled)
			if (len(wordpunct[-1])>1 and wordpunct[-1][-1] in string.punctuation and wordpunct[-2] not in string.punctuation) or (wordpunct[-1] in string.punctuation and wordpunct[-2] not in string.punctuation):
				words =tokenize(apos_handled)
				words[-1]=wordpunct[-2]
				words.append(wordpunct[-1])
			else:
				words =tokenize(apos_handled)

			return (" ".join(words)).strip()
		except:
			return sentence

	concat_df['clean_text']=concat_df['text'].apply(lambda x: sentences_cleaner(x))

	#Remove rows with len(clean_text) < 3
	concat_df['text_length']=concat_df['clean_text'].apply(lambda x: len(x))
	concat_df=concat_df[concat_df.text_length>3]
	concat_df=concat_df.reset_index(drop=True)
	concat_df=concat_df.drop(columns=['text_length'])

	return(concat_df)

#dictionary that contains pos tags and their explanations
# 'CC': 'coordinating conjunction','CD': 'cardinal digit','DT': 'determiner',
# 'EX': 'existential there (like: \"there is\" ... think of it like \"there exists\")',
# 'FW': 'foreign word','IN':  'preposition/subordinating conjunction','
# JJ': 'adjective \'big\'','JJR': 'adjective, comparative \'bigger\'',
# 'JJS': 'adjective, superlative \'biggest\'', 'LS': 'list marker 1)', 'MD': 'modal could, will',
# 'NN': 'noun, singular \'desk\'', 'NNS': 'noun plural \'desks\'',
#'NNP': 'proper noun, singular \'Harrison\'','NNPS': 'proper noun, plural \'Americans\'',
# 'PDT': 'predeterminer \'all the kids\'','POS': 'possessive ending parent\'s',
# 'PRP': 'personal pronoun I, he, she','PRP$': 'possessive pronoun my, his, hers',
# 'RB': 'adverb very, silently,', 'RBR': 'adverb, comparative better',
# 'RBS': 'adverb, superlative best','RP': 'particle give up', 'TO': 'to go \'to\' the store.',
# 'UH': 'interjection errrrrrrrm','VB': 'verb, base form take','VBD': 'verb, past tense took',
# 'VBG': 'verb, gerund/present participle taking','VBN': 'verb, past participle taken',
# 'VBP': 'verb, sing. present, non-3d take','VBZ': 'verb, 3rd person sing. present takes',
# 'WDT': 'wh-determiner which','WP': 'wh-pronoun who, what','WP$': 'possessive wh-pronoun whose',
# 'WRB': 'wh-abverb where, when','QF' : 'quantifier, bahut, thoda, kam (Hindi)',
# 'VM' : 'main verb','PSP' : 'postposition, common in indian langs','DEM' : 'demonstrative, common in indian langs'

#Extract Parts of Speech as BOW
def extract_pos(doc):
#pos_dict = {'CC':0, 'CD':0,'DT':0,'EX':0,'FW':0,'JJ':0,'JJR':0,'JJS':0,'LS':0,'MD':0,
#			'NN':0,'NNS':0,'NNP':0,'NNPS':0,'PDT':0,'POS':0,'PRP':0,'PRP$':0,'RB':0,
#			'RBR':0,'RBS':0,'RP':0,'TO':0,'UH':0,'VB':0,'VBD':0,'VBG':0,'VBN':0,'VBP':0,
#			'VBZ':0,'VM':0,'WDT':0,'WP':0,'WP$':0,'WRB':0,'QF':0,'PSP':0,'DEM':0}
	pos_dict = {'VB':0,'VBD':0,'VBG':0,'VBN':0,'VBP':0,'VBZ':0,'VM':0}
	try:
		for sent in doc.sentences:
			for wrd in sent.words:
				if wrd.pos in pos_dict.keys():
					pos_dict[wrd.pos]+=1
    	#return BOW of POS
		return pos_dict
	except:
		return pos_dict

def n_grams_handled(sentence):
	'''
	Filter before generate n-gram
	'''
	try:
		tk=TweetTokenizer()
		cashtag_pat=r'\$[^\s]+'
		hashtag_pat=r'#([^\s]+)'
		word_number_pat=r'\w*\d\w*'

		#Remove word which has length < 2
		stripped=' '.join([word for word in sentence.split() if len(word)>=2])

		#Remove hashtag
		hashtag_handled= re.sub(hashtag_pat,"", stripped)

		#Remove cashtag
		cashtag_handled= re.sub(cashtag_pat,"", hashtag_handled)

		#Remove word with number
		number_handled= re.sub(word_number_pat,"", cashtag_handled)

		#Remove unnecesary white spaces
		words = tk.tokenize(number_handled)
		words = [x for x in words if x not in string.punctuation]
		clean_sentence=(" ".join(words)).strip()
		return  clean_sentence
	except:
		return sentence

def rf_ngram(dict_source,df,gram):
	'''
	create rf-ngram
	'''
	def sentence_sparse(sentence,gram,rf_ngram,sparse_rf_ngram):
		#Initiate Linke List Sparse Matrix
		zero_sparse=sp.lil_matrix( (1,len(rf_ngram)), dtype=float)
		#Assign Value of rf_ngram to each word in sentence
		splitted_text=tokenize(n_grams_handled(sentence))
		#Unigram
		if gram==1:
			for word in splitted_text:
				if word in rf_ngram.index:
					zero_sparse[0,rf_ngram.index.get_loc(word)]+=sparse_rf_ngram[0,rf_ngram.index.get_loc(word)]
			#Convert LinkedList Sparse Matrix into CSR Sparse Matrix
			sparse=zero_sparse.tocsr()
		#Bigram
		elif gram==2:
			bigram=lambda x: splitted_text[x]+' '+splitted_text[x+1]
			it_2_gram=range(len(splitted_text)-1)
			for i in it_2_gram:
				if bigram(i) in rf_ngram.index:
					zero_sparse[0,rf_ngram.index.get_loc(bigram(i))]+=sparse_rf_ngram[0,rf_ngram.index.get_loc(bigram(i))]
			#Convert LinkedList Sparse Matrix into CSR Sparse Matrix
			sparse=zero_sparse.tocsr()
		#Trigram
		elif gram==3:
			trigram=lambda x: splitted_text[x]+' '+splitted_text[x+1]+' '+splitted_text[x+2]
			it_3_gram=range(len(splitted_text)-2)
			for i in it_3_gram:
				if trigram(i) in rf_ngram.index:
					zero_sparse[0,rf_ngram.index.get_loc(trigram(i))]+=sparse_rf_ngram[0,rf_ngram.index.get_loc(trigram(i))]
			#Convert LinkedList Sparse Matrix into CSR Sparse Matrix
			sparse=zero_sparse.tocsr()
		#4grams
		elif gram==4:
			fourgram=lambda x: splitted_text[x]+' '+splitted_text[x+1]+' '+splitted_text[x+2]+' '+splitted_text[x+3]
			it_4_gram=range(len(splitted_text)-3)
			for i in it_4_gram:
				if fourgram(i) in rf_ngram.index:
					zero_sparse[0,rf_ngram.index.get_loc(fourgram(i))]+=sparse_rf_ngram[0,rf_ngram.index.get_loc(fourgram(i))]
			#Convert LinkedList Sparse Matrix into CSR Sparse Matrix
			sparse=zero_sparse.tocsr()
		return(sparse)

	BOW_df= dict_source

	#Calculate rf_ngram for each word
	series_1=pd.Series([1 for x in range(len(BOW_df))])
	series_1.index=BOW_df.index
	series_2=pd.Series([2 for x in range(len(BOW_df))])
	series_2.index=BOW_df.index
	frac_1=np.log(series_2+(BOW_df['pos']/pd.concat([series_1,BOW_df['neg']],1).max(axis=1)))
	frac_2=np.log(series_2+(BOW_df['neg']/pd.concat([series_1,BOW_df['pos']],1).max(axis=1)))
	rf_ngram_series= pd.concat([frac_1,frac_2],1).max(axis=1)
	sparse_rf_ngram=sp.csr_matrix(rf_ngram_series)

	def rf_ngram_calculate(x):
		lst=[i for i in sentence_sparse(x,gram,rf_ngram_series,sparse_rf_ngram).toarray()[0].tolist() if i!=0]
		if type(x)!=str:
			return(np.nan)
		else:
			if len(lst)>0:
				return(np.mean(lst))
			else:
				return(np.nan)

	rf_ngram_avg_list=df['clean_text'].apply(lambda x: rf_ngram_calculate(x))

	return(rf_ngram_avg_list)

def PMI(dict_source,df):
	'''
	create PMI variable
	'''
	BOW_df= dict_source

	N=len(BOW_df) #Number of unique tokens in the corpus
	pos_N=len(BOW_df[BOW_df.pos!=0]) #Number of unique positive tokens in the corpus
	neg_N=len(BOW_df[BOW_df.neg!=0]) #Number of unique positive tokens in the corpus
	total=BOW_df.sum().sum() #Number of tokens in the corpus
	pos_total=BOW_df.sum()['pos'] #Number of tokens in the positive corpus
	neg_total=BOW_df.sum()['neg'] #Number of tokens in the negative corpus
	PMI_df=pd.DataFrame(columns=['freq_word','freq_word_pos','freq_word_neg'])
	PMI_df['freq_word']=pd.Series(BOW_df.index).apply(lambda x: (BOW_df.loc[x,'pos']+BOW_df.loc[x,'neutral']+BOW_df.loc[x,'neg'])/total)
	PMI_df['freq_word_pos']=pd.Series(BOW_df.index).apply(lambda x: BOW_df.loc[x,'pos']/pos_total) #Freq of word w in positive text
	PMI_df['freq_word_neg']=pd.Series(BOW_df.index).apply(lambda x: BOW_df.loc[x,'neg']/neg_total) #Freq of word w in negative text
	PMI_df.index=BOW_df.index

	#Calculate PMI for each word
	PMI_df['PMI_pos']=np.log2(1+((PMI_df['freq_word_pos']*N)/(PMI_df['freq_word']*pos_N)))
	PMI_df['PMI_neg']=np.log2(1+((PMI_df['freq_word_neg']*N)/(PMI_df['freq_word']*neg_N)))
	PMI_df['PMI']=PMI_df['PMI_pos']-PMI_df['PMI_neg']

	def PMI_calculate(x):
		lst=[PMI_df.loc[i,'PMI'] for i in tokenize(n_grams_handled(x)) if i in PMI_df.index]
		if type(x)!=str:
			return(np.nan)
		else:
			if len(lst)>0:
				return(np.mean(lst))
			else:
				return(np.nan)

	PMI_avg_list=df['clean_text'].apply(lambda x: PMI_calculate(x))

	return(PMI_avg_list)

def countAllCaps(text):
	""" Input: a text, Output: how many words are all caps """
	return len(re.findall("[A-Z]{2,}", text))

def countHashtag(text):
	""" Input: a text, Output: how many hastags in front of a word """
	return len(re.findall(r'#([^\s]+)', text))

def is_ordinal_numbers(sentences):
	occur=0
	for word in tokenize(sentences):
		if ((word[-2:] in ['st','nd','rd','th']) and (isfloat(word[:-2]))):
			occur=1
	return(occur)

def countMultiExclamationMarks(sentences):
	""" Replaces repetitions of exlamation marks """
	return len(re.findall(r"(\!)\1+", sentences))

def countMultiQuestionMarks(sentences):
	""" Count repetitions of question marks """
	return len(re.findall(r"(\?)\1+", sentences))

def sentence_synset(sentence):
	'''
	return the wordnet synset of each word in the sentence
	'''
	def penn_to_wn(tag):
		if tag.startswith('J'):
			return wn.ADJ
		elif tag.startswith('N'):
			return wn.NOUN
		elif tag.startswith('R'):
			return wn.ADV
		elif tag.startswith('V'):
			return wn.VERB
		return None

	tagged = pos_tag(tokenize(sentence))

	synsets_list = []
	lemmatzr = WordNetLemmatizer()

	for token in tagged:
		wn_tag = penn_to_wn(token[1])
		if not wn_tag:
	  		continue

		lemma = lemmatzr.lemmatize(token[0], pos=wn_tag)
		try:
			synsets_list.append(wn.synsets(lemma, pos=wn_tag)[0])
		except:
			None
	return synsets_list

def min_multiple_list(S):
	'''
	Minimum pooling
	'''
	it=range(len(S)-1)
	minim=S[0]
	for i in it:
		minim=np.minimum(minim,S[i])
	return(minim)

def max_multiple_list(S):
	'''
	Maximum pooling
	'''
	it=range(len(S)-1)
	maxim=S[0]
	for i in it:
		maxim=np.maximum(maxim,S[i])
	return(maxim)

def rescaling(df,columns,scale_type='Standard'):
	'''
	Function for Feature Scaling
	'''
	scale_type=scale_type.lower()
	scaled_X=df.drop(columns,1)
	X=df[columns]

	if scale_type=='minmax':
		scaler=MinMaxScaler(feature_range=(0,1))
	elif scale_type=='standard':
		scaler=StandardScaler()

	scaled_column=scaler.fit_transform(X)
	scaled_column=pd.DataFrame(scaled_column,columns=columns)
	for column in columns:
		scaled_X[column]=scaled_column[column].tolist()

	return(scaled_X)

def feature_engineering_split(df):

	print("==================== Feature Engineering by Splitting ====================")
	#List of POS-Tag
	#pos_key = ['CC', 'CD','DT','EX','FW','JJ','JJR','JJS','LS','MD', 'NN','NNS','NNP','NNPS','PDT'
	#	 	    ,'POS','PRP','PRP$','RB', 'RBR','RBS','RP','TO','UH','VB','VBD','VBG','VBN','VBP',
	#			'VBZ','VM','WDT','WP','WP$','WRB','QF','PSP','DEM']
	pos_key = ['VB','VBD','VBG','VBN','VBP','VBZ','VM']

	#Initiate pipeline for POS-Tagging
	nlp = stanfordnlp.Pipeline(processors = "tokenize,pos")

	#Inititate class for Lemmatization
	lemmatizer = WordNetLemmatizer()

	#Initiate class for Stemming
	stemmer = PorterStemmer()

	#Lemmatization+Stemming
	df['base_text']=df['clean_text'].apply(lambda x: ' '.join(pd.Series(tokenize(x)).apply(lambda wrd: stemmer.stem(lemmatizer.lemmatize(wrd)) if wrd[0]!='$' else wrd).to_list()) if type(x)==str else np.nan)
	print('Done Base Text')

	#Create POS-Tag features
	for tag in pos_key:
		df['POS_'+tag]=df['clean_text'].apply(lambda x: extract_pos(nlp(x))[tag] if type(x)==str else np.nan)
	print('Done POS Tag')

	#Binary Feature '+num'
	df["'+num"]=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\+\d+\s|\+\d+[!,.;:?/]|\+\d+$',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature '-num'
	df["'-num"]=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\-\d+\s|\-\d+[!,.;:?/]|\-\d+$',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature 'num%'
	df["num%"]=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r' \d.\d*%+|^\d.\d*%+|[!,.;:?/]\d.\d*%+| \d*%+|^\d*%+|[!,.;:?/]\d*%+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature '+num%'
	df["'+num%"]=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\+\d.\d*%+|\+\d*%+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature '-num%'
	df["'-num%'"]=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\-\d.\d*%+|\-\d*%+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Features '$num'
	df['$num']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\$\d*%+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature mixed number and word
	df['word_num']=df['clean_text'].apply(lambda x: 1 if (type(x)==str and len(re.findall(r'\w*\d\w*',x))>0) else 0 if (type(x)==str) else np.nan)

	#Binary Feature ordinal number
	df['ordinal_num']=df['clean_text'].apply(lambda x: is_ordinal_numbers(x) if type(x)==str else np.nan)

	#Binary Feature  'num-num'
	df['num-num']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\d*-\d+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature  'num-num%'
	df['num-num%']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\d*-\d%+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature  'num-num-num'
	df['num-num-num']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\d*-\d-\d+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature  'num/num'
	df['num/num']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\d*/\d+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature  'num/num/num'
	df['num/num/num']=df['clean_text'].apply(lambda x: 1 if ((type(x)==str) and len(re.findall(r'\d*/\d/\d+',x))>0) else 0  if type(x)==str else np.nan)

	#Binary Feature  only numbers(no symbol and characters)
	df['only_number']=df['clean_text'].apply(lambda x:1 if (type(x)==str and any(isfloat(wrd) for wrd in tokenize(x))) else 0 if type(x)==str else np.nan)
	print('Done Keyword+num')

	f_plus=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if len(re.findall(r'\+\d.\d*%+|\+\d*%+',wrd))>0 else 0)
	g_plus=lambda y: f_plus(y)[f_plus(y)==1].index.tolist()
	f_min=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if len(re.findall(r'\-\d.\d*%+|\-\d*%+',wrd))>0 else 0)
	g_min=lambda y: f_min(y)[f_min(y)==1].index.tolist()

	#Binary Feature  'call'(or 'calls' or 'called') before '+num%'
	df['call_+num%']=df['clean_text'].apply(lambda z: 1 if (type(z)==str and any((tokenize(z)[i-1]=='call' or tokenize(z)[i-1]=='calls' or tokenize(z)[i-1]=='called') for i in g_plus(z))) else 0 if type(z)==str else np.nan)

	#Binary Feature  'call'(or 'calls' or 'called') before '-num%'
	df['call_-num%']=df['clean_text'].apply(lambda z: 1 if (type(z)==str and any((tokenize(z)[i-1]=='call' or tokenize(z)[i-1]=='calls' or tokenize(z)[i-1]=='called') for i in g_min(z))) else 0 if type(z)==str else np.nan)

	#Binary Feature  'put'(or 'puts') before '+num%'
	df['put_+num%']=df['clean_text'].apply(lambda z: 1 if (type(z)==str and any((tokenize(z)[i-1]=='put' or tokenize(z)[i-1]=='puts') for i in g_plus(z))) else 0 if type(z)==str else np.nan)

	#Binary Feature  'put'(or 'puts') before '-num%'
	df['put_-num%']=df['clean_text'].apply(lambda z: 1 if (type(z)==str and any((tokenize(z)[i-1]=='put' or tokenize(z)[i-1]=='puts') for i in g_min(z))) else 0 if type(z)==str else np.nan)

	#Binary Feature 'Bull' or 'Bullish'
	df['bull']=df['clean_text'].apply(lambda x: 1 if (type(x)==str and ('bull' or 'bullish') in x.split()) else 0 if type(x)==str else np.nan)

	#Binary Feature 'Bear' or 'Bearish'
	df['bear']=df['clean_text'].apply(lambda x: 1 if (type(x)==str and ('bear' or 'bearish') in x.split()) else 0 if type(x)==str else np.nan)
	print('Done Specific Keyword')

	tk=TweetTokenizer()
	#Calculate the number of '!'
	df['number_of_!']=df['clean_text'].apply(lambda x: tk.tokenize(x).count('!') if type(x)==str else np.nan)

	#Calculate the number of '?'
	df['number_of_?']=df['clean_text'].apply(lambda x: tk.tokenize(x).count('?') if type(x)==str else np.nan)

	#Calculate the number of '$'
	df['number_of_$']=df['clean_text'].apply(lambda x: tk.tokenize(x).count('$') if type(x)==str else np.nan)

	#Calculate the number of continuous '!'
	df['continous_!']=df['clean_text'].apply(lambda x: countMultiExclamationMarks(x) if type(x)==str else np.nan)

	#Calculate the number of continuous '?'
	df['continous_?']=df['clean_text'].apply(lambda x: countMultiQuestionMarks(x) if type(x)==str else np.nan)
	print('Done Punctation Count')

	#Calculate the number of Caps word
	df['caps_word']=df['clean_text'].apply(lambda x: countAllCaps(' '.join([i for i in x.split() if i[0]!='$'])) if type(x)==str else np.nan)
	print('Done Caps words')

	#Calculate the number of Hashtags
	df['hashtags']=df['clean_text'].apply(lambda x: countHashtag(x) if type(x)==str else np.nan)
	print('Done Hashtags')

	#AFINN Sentiment Lexicon
	affin_sent_score=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: AFINN_dict[wrd.lower()] if wrd.lower() in AFINN_dict.keys() else 0)
	affin_sent_binary=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if (wrd.lower() in AFINN_dict.keys() and AFINN_dict[wrd.lower()]>0) else
	                                                     -1 if (wrd.lower() in AFINN_dict.keys() and AFINN_dict[wrd.lower()]<0) else 0)
	#Sum Score
	df['AFINN_sum_score']=df['clean_text'].apply(lambda x: affin_sent_score(x).sum() if type(x)==str else np.nan)
	#Max Score
	df['AFINN_max_score']=df['clean_text'].apply(lambda x: affin_sent_score(x).max() if type(x)==str else np.nan)
	#Min Score
	df['AFINN_min_score']=df['clean_text'].apply(lambda x: affin_sent_score(x).min() if type(x)==str else np.nan)
	#Ratio of Positive Words
	df['AFINN_pos_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in affin_sent_binary(x) if i==1)/len(tokenize(x)) if type(x)==str else np.nan)
	#Ratio of Negatiive Words
	df['AFINN_neg_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in affin_sent_binary(x) if i==-1)/len(tokenize(x)) if type(x)==str else np.nan)
	print('Done AFIIN Lexicon')

	#BingLiu Sentiment Lexicon
	bingliu_sent_binary=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if wrd.lower() in BingLiu_dict['pos'] else -1 if wrd.lower() in BingLiu_dict['neg'] else 0)
	#Ratio of Positive Words
	df['BingLiu_pos_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in bingliu_sent_binary(x) if i==1)/len(tokenize(x)) if type(x)==str else np.nan)
	#Ratio of Negative Words
	df['BingLiu_neg_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in bingliu_sent_binary(x) if i==-1)/len(tokenize(x)) if type(x)==str else np.nan)
	print('Done BingLiu Lexicon')

	#General Inquirer Sentiment Lexicon
	general_inquirer_binary=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if wrd.lower() in General_Inquirer_dict['pos'] else -1 if wrd.lower() in General_Inquirer_dict['neg'] else 0)
	#Ratio of Positive Words
	df['General_Inquirer_pos_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in general_inquirer_binary(x) if i==1)/len(tokenize(x)) if type(x)==str else np.nan)
	#Ratio of Negative Words
	df['General_Inquirer_neg_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in general_inquirer_binary(x) if i==-1)/len(tokenize(x)) if type(x)==str else np.nan)
	print('Done General Inquirer Lexicon')

	#NRC Hashtag Sentiment Lexicon
	nrc_hashtag_sent_score=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: NRC_hashtag_dict[wrd.lower()] if wrd.lower() in NRC_hashtag_dict.keys() else 0)
	nrc_hashtag_sent_binary=lambda x: pd.Series(tokenize(x)).apply(lambda wrd: 1 if (wrd.lower() in NRC_hashtag_dict.keys() and NRC_hashtag_dict[wrd.lower()]>0) else
	                                                     -1 if (wrd.lower() in NRC_hashtag_dict.keys() and NRC_hashtag_dict[wrd.lower()]<0) else 0)
	#Sum Score
	df['NRC_Hashtag_sum_score']=df['clean_text'].apply(lambda x: nrc_hashtag_sent_score(x).sum() if type(x)==str else np.nan)
	#Max Score
	df['NRC_Hashtag_max_score']=df['clean_text'].apply(lambda x: nrc_hashtag_sent_score(x).max() if type(x)==str else np.nan)
	#Min Score
	df['NRC_Hashtag_min_score']=df['clean_text'].apply(lambda x: nrc_hashtag_sent_score(x).min() if type(x)==str else np.nan)
	#Ratio of Positive Words
	df['NRC_Hashtag_pos_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in nrc_hashtag_sent_binary(x) if i==1)/len(tokenize(x)) if type(x)==str else np.nan)
	#Ratio of Negatiive Words
	df['NRC_Hashtag_neg_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in nrc_hashtag_sent_binary(x) if i==-1)/len(tokenize(x)) if type(x)==str else np.nan)
	print('Done NRC Hashtag Sentiment Lexicon')

	#SentiWordNet Sentiment Lexicon
	sentiwordnet_list=sentiwordnet.ID.tolist()
	sent_to_synset=lambda x: pd.Series(sentence_synset(x))
	synset_to_offset=lambda x: int(str(x.offset()).zfill(8))
	get_value=lambda x: sentiwordnet[sentiwordnet.ID==synset_to_offset(x)]['score'].values[0]
	score_offset_check=lambda x: get_value(x) if (synset_to_offset(x) in sentiwordnet_list) else 0
	binary_offset_check=lambda x: 1 if (synset_to_offset(x) in sentiwordnet_list and get_value(x)>0) else  -1 if  (synset_to_offset(x) in sentiwordnet_list and get_value(x)<0) else 0
	sentiwordnet_score=lambda sent: sent_to_synset(sent).apply(lambda z: score_offset_check(z))
	sentiwordnet_binary=lambda sent: sent_to_synset(sent).apply(lambda z: binary_offset_check(z))
	#Sum Score
	df['SentiWordNet_sum_score']=df['clean_text'].apply(lambda x: sentiwordnet_score(x).sum() if type(x)==str else np.nan)
	#Max Score
	df['SentiWordNet_max_score']=df['clean_text'].apply(lambda x: sentiwordnet_score(x).max() if type(x)==str else np.nan)
	#Min Score
	df['SentiWordNet_min_score']=df['clean_text'].apply(lambda x: sentiwordnet_score(x).min() if type(x)==str else np.nan)
	#Ratio of Positive Words
	df['SentiWordNet_pos_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in sentiwordnet_binary(x) if i==1)/len(sent_to_synset(x)) if (type(x)==str and len(sent_to_synset(x))>0) else 0 if type(x)==str else np.nan)
	#Ratio of Negatiive Words
	df['SentiWordNet_neg_ratio']=df['clean_text'].apply(lambda x: sum(1 for i in sentiwordnet_binary(x) if i==-1)/len(sent_to_synset(x)) if (type(x)==str and len(sent_to_synset(x))>0) else 0 if type(x)==str else  np.nan)
	print('Done SentiWordNet Lexicon')
	return(df)

def feature_engineering(df):

	print("==================== Feature Engineering ====================")
	#n-grams
	for grams in [1,2,3,4]:
		nan_checker=lambda x: x if type(x)==str else ''

		#Initiate class for BOW
		bow_vectorizer= CountVectorizer(ngram_range=(grams,grams))
		#Initiate class for TF-IDF
		tfidf_vectorizer = TfidfVectorizer(norm=None, ngram_range=(grams,grams))

		#Create docs
		docs=df['clean_text'].apply(lambda x: n_grams_handled(x))

		#Create TF-IDF matrix
		tfidf_matrix = tfidf_vectorizer.fit_transform(docs.apply(lambda x: nan_checker(x)).to_list())

		#Create TF-IDF n-grams
		df['Avg_TFIDF_'+str(grams)+'-grams']=[np.mean([x for x in tfidf_matrix[i].toarray()[0].tolist() if x!=0]) for i in range(len(df))]

		#Calculate sum of RF n-gram
		if grams==1:
			df['Avg_rf_'+str(grams)+'-grams']=rf_ngram(rf_1_gram_df,df,grams)
		elif grams==2:
			df['Avg_rf_'+str(grams)+'-grams']=rf_ngram(rf_2_gram_df,df,grams)
		elif grams==3:
			df['Avg_rf_'+str(grams)+'-grams']=rf_ngram(rf_3_gram_df,df,grams)
		elif grams==4:
			df['Avg_rf_'+str(grams)+'-grams']=rf_ngram(rf_4_gram_df,df,grams)

	print('Done n-grams')

	#Calculate PMI
	df['PMI_score']=PMI(PMI_df,df)

	#Impute missing value
	imputer = KNNImputer(n_neighbors=3)
	df_imputed=pd.DataFrame(imputer.fit_transform(df.drop(columns=['id','created_at','body','clean_text','base_text'])))
	df_imputed.columns=df.drop(columns=['id','created_at','body','clean_text','base_text']).columns
	df['Avg_rf_1-grams']=df_imputed['Avg_rf_1-grams']
	df['Avg_rf_2-grams']=df_imputed['Avg_rf_2-grams']
	df['Avg_rf_3-grams']=df_imputed['Avg_rf_3-grams']
	df['Avg_rf_4-grams']=df_imputed['Avg_rf_4-grams']
	df['PMI_score']=df_imputed['PMI_score']

	return(df)

def parallelize_dataframe(df, func, n_split):
	'''
	Function to parallelize a dataframe
	'''
	df_split = np.array_split(df, n_split)
	df_pool=func(df_split[0])
	for i in range(n_split-1):
		x=df_split[i+1]
		x=func(x.copy())
		df_pool = pd.concat([df_pool,x],ignore_index=True)
	return df_pool

def main():
	clean_df=clean_data(df_stocktwits_full_BA)

	draft_engineered_df=parallelize_dataframe(clean_df, feature_engineering_split,n_split=2)

	engineered_df=parallelize_dataframe(draft_engineered_df, feature_engineering,n_split=3)

	engineered_df.to_csv(PATH_ROOT+'Dataset/'+'df_stocktwits_prepared.csv')

if __name__ == '__main__':
	main()


text:
from volpy import Camera

import numpy as np
import numpy.testing as npt

import unittest

x = np.array([1, 0, 0])
y = np.array([0, 1, 0])
z = np.array([0, 0, 1])

class CameraTest(unittest.TestCase):

    def setUp(self):
        self.camera = Camera([0, 0, 0, 1], [0, 0, 1, 0])

    def test_near1(self):
        self.assertEqual(0.1, self.camera.near)

    def test_far1(self):
        self.assertEqual(2., self.camera.far)

    def test_eye1(self):
        self.assertIsInstance(self.camera.eye, np.ndarray)
        self.assertEqual(self.camera.eye.dtype, np.dtype(np.float32))
        npt.assert_almost_equal([0, 0, 0, 1], self.camera.eye)

    def test_eye2(self):
        self.camera.eye = (0, 1, 1, 1)
        self.assertIsInstance(self.camera.eye, np.ndarray)
        npt.assert_almost_equal([0, 1, 1, 1], self.camera.eye)

    def test_view1(self):
        self.assertIsInstance(self.camera.view, np.ndarray)
        self.assertEqual(self.camera.view.dtype, np.dtype(np.float32))
        npt.assert_almost_equal([0, 0, 1, 0], self.camera.view)
        npt.assert_almost_equal(0, self.camera.up.dot(self.camera.view))

    def test_view2(self):
        self.camera.view = [2, 0, 0, 0]
        self.assertIsInstance(self.camera.view, np.ndarray)
        npt.assert_almost_equal([1, 0, 0, 0], self.camera.view)
        npt.assert_almost_equal(0, self.camera.up.dot(self.camera.view))

    def test_up1(self):
        self.assertIsInstance(self.camera.up, np.ndarray)
        self.assertEqual(self.camera.up.dtype, np.dtype(np.float32))
        npt.assert_almost_equal([0, 1, 0, 0], self.camera.up)
        npt.assert_almost_equal(0, self.camera.up.dot(self.camera.view))

    def test_up2(self):
        self.camera.up = [2, 0, 0, 0]
        self.assertIsInstance(self.camera.up, np.ndarray)
        npt.assert_almost_equal([1, 0, 0, 0], self.camera.up)
        npt.assert_almost_equal(0, self.camera.up.dot(self.camera.view))

    def test_right1(self):
        self.assertIsInstance(self.camera.right, np.ndarray)
        self.assertEqual(self.camera.right.dtype, np.dtype(np.float32))
        npt.assert_almost_equal(0, self.camera.up.dot(self.camera.right))
        npt.assert_almost_equal(0, self.camera.view.dot(self.camera.right))

    def test_right2(self):
        with self.assertRaises(AttributeError):
            self.camera.right = (0, 1, 0, 0)

    def test_cast1(self):
        origin, direction = self.camera.cast([0.5], [0.5])
        self.assertIsInstance(origin, np.ndarray)
        self.assertIsInstance(direction, np.ndarray)
        self.assertEqual(origin.dtype, np.dtype(np.float32))
        self.assertEqual(direction.dtype, np.dtype(np.float32))
        npt.assert_almost_equal(origin, [[0, 0, 0.1, 1]])
        npt.assert_almost_equal(direction, [[0, 0, 1, 0]])

    def test_cast2(self):
        '''Cast should not write to arguments'''
        coord = np.linspace(0, 1, 10)
        coord.flags.writeable = False
        self.camera.cast(coord, coord)

    def test_cast3(self):
        '''Checks for out of bounds'''
        coord = np.linspace(0, 1.1, 10)
        with self.assertRaises(ValueError):
            self.camera.cast(coord, coord)

    def test_cast4(self):
        '''Checks for same size'''
        coord1 = np.linspace(0, 1.0, 10)
        coord2 = np.linspace(0, 1.0, 11)
        with self.assertRaises(ValueError):
            self.camera.cast(coord1, coord2)

ext: py

max_stars_repo_name: OEP/volpy

max_issues_repo_name: OEP/volpy

max_forks_repo_name: OEP/volpy

reason: import numpy


text:
# Copyright 2019-2020 QuantumBlack Visual Analytics Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND
# NONINFRINGEMENT. IN NO EVENT WILL THE LICENSOR OR OTHER CONTRIBUTORS
# BE LIABLE FOR ANY CLAIM, DAMAGES, OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF, OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#
# The QuantumBlack Visual Analytics Limited ("QuantumBlack") name and logo
# (either separately or in combination, "QuantumBlack Trademarks") are
# trademarks of QuantumBlack. The License does not grant you any right or
# license to the QuantumBlack Trademarks. You may not use the QuantumBlack
# Trademarks or any confusingly similar mark as a trademark for your product,
#     or use the QuantumBlack Trademarks in any other manner that might cause
# confusion in the marketplace, including but not limited to in advertising,
# on websites, or on software.
#
# See the License for the specific language governing permissions and
# limitations under the License.

import numpy as np
import pytest

from causalnex.discretiser.discretiser_strategy import (
    DecisionTreeSupervisedDiscretiserMethod,
)

class TestDecisionTree:
    def test_single_continuous(self, continuous_data):
        diabete = continuous_data.copy(deep=True)

        dt_single = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 2},
            mode="single",
        )
        tree_discretiser = dt_single.fit(
            feat_names=["s6"],
            dataframe=diabete,
            target_continuous=True,
            target="target",
        )
        discretiser_output = tree_discretiser.transform(diabete[["s6"]]).values

        ground_truth = np.array(
            [
                [1, 0, 1, 1, 0, 0, 0, 1, 1, 1, 0, 0, 0],
                [1, 0, 0, 1, 1, 1, 0, 1, 1, 1, 3, 0, 0],
                [1, 2, 0, 2, 1, 0, 1, 1, 0, 1, 1, 1, 1],
                [1, 2, 1, 0, 2, 1, 0, 0, 0, 1, 1, 1, 1],
                [1, 2, 1, 1, 1, 0, 0, 1, 0, 1, 1, 0, 1],
                [1, 1, 1, 1, 1, 2, 2, 1, 1, 2, 1, 1, 1],
                [0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 2, 1, 1],
                [1, 1, 0, 0, 1, 1, 2, 0, 0, 1, 2, 1, 1],
                [0, 1, 0, 1, 2, 2, 1, 0, 1, 2, 1, 1, 1],
                [3, 2, 1, 1, 1, 2, 2, 1, 1, 0, 0, 0, 2],
                [2, 0, 1, 1, 0, 2, 1, 1, 2, 1, 1, 3, 1],
                [0, 1, 0, 1, 2, 1, 0, 1, 1, 2, 1, 1, 2],
                [0, 1, 0, 1, 0, 2, 1, 2, 1, 1, 0, 2, 3],
                [1, 1, 0, 2, 0, 1, 1, 1, 1, 1, 1, 1, 0],
                [1, 2, 1, 1, 0, 0, 1, 1, 1, 1, 0, 1, 1],
                [1, 1, 2, 0, 1, 2, 1, 1, 1, 2, 1, 1, 2],
                [2, 1, 0, 1, 1, 1, 0, 2, 2, 2, 1, 1, 0],
                [0, 1, 0, 1, 1, 0, 1, 0, 1, 2, 1, 1, 0],
                [2, 2, 1, 1, 1, 2, 1, 2, 0, 0, 1, 0, 0],
                [0, 2, 1, 2, 2, 1, 2, 2, 1, 1, 0, 2, 1],
                [1, 1, 1, 1, 1, 2, 1, 1, 2, 0, 1, 1, 0],
                [2, 1, 2, 1, 1, 1, 1, 1, 1, 1, 1, 0, 1],
                [1, 1, 1, 0, 2, 0, 0, 1, 0, 1, 0, 0, 1],
                [1, 2, 2, 1, 1, 1, 0, 1, 2, 0, 2, 1, 1],
                [1, 1, 0, 1, 2, 2, 1, 2, 1, 2, 2, 2, 1],
                [2, 1, 1, 1, 2, 1, 1, 2, 1, 0, 0, 2, 1],
                [0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 3],
                [0, 0, 1, 2, 1, 0, 0, 0, 2, 2, 1, 2, 1],
                [2, 2, 2, 1, 2, 1, 0, 1, 1, 1, 0, 1, 1],
                [0, 0, 1, 1, 0, 2, 1, 0, 1, 1, 0, 0, 0],
                [2, 0, 2, 1, 1, 0, 0, 1, 0, 1, 1, 2, 2],
                [2, 1, 1, 0, 1, 2, 1, 0, 1, 2, 1, 1, 1],
                [2, 1, 1, 1, 2, 1, 1, 0, 1, 0, 2, 1, 3],
                [0, 1, 1, 2, 1, 1, 0, 0, 1, 2, 1, 1, 1],
            ]
        )  # ground truth is generated by manually use DecionTree to extract thresholds

        assert (ground_truth == discretiser_output.reshape(-1, 13)).all()

    def test_single_categorical(self, categorical_data):
        df = categorical_data.copy(deep=True)
        ground_truth = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2],
                [2, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 1, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )  # ground truth is generated by manually use DecionTree to extract thresholds

        dt_single = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 2},
            mode="single",
        )
        tree_discretiser = dt_single.fit(
            feat_names=["petal width (cm)"],
            dataframe=df,
            target_continuous=False,
            target="target",
        )
        discretiser_output = tree_discretiser.transform(df[["petal width (cm)"]]).values
        assert (ground_truth == discretiser_output.reshape(-1, 15)).all()

    def test_invalid_mode(self):
        with pytest.raises(KeyError):
            DecisionTreeSupervisedDiscretiserMethod(
                tree_params={"max_depth": 2}, mode="invalid"
            )

    def test_transform_no_feature(self, get_iris_data, caplog):
        ground_truth = get_iris_data[["sepal width (cm)"]]
        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            mode="multi",
            split_unselected_feat=False,
            tree_params={"max_depth": 3, "random_state": 2020},
        )
        tree_discretiser = dt_multi.fit(
            feat_names=["sepal length (cm)", "petal length (cm)"],
            dataframe=get_iris_data,
            target_continuous=False,
            target="target",
        )

        output = tree_discretiser.transform(get_iris_data[["sepal width (cm)"]])

        assert "The column is left unchanged" in caplog.text
        assert all(ground_truth == output)

    def test_keep_unselected_feature(self, get_iris_data):
        ground_truth = np.array(
            [
                [4, 2, 3, 3, 4, 6, 4, 4, 1, 3, 4, 4, 2, 2, 6],
                [6, 6, 4, 5, 5, 4, 4, 4, 3, 4, 2, 4, 4, 4, 3],
                [3, 4, 6, 6, 3, 3, 4, 4, 2, 4, 4, 0, 3, 4, 5],
                [2, 5, 3, 4, 3, 3, 3, 3, 0, 1, 1, 3, 0, 1, 1],
                [0, 2, 0, 1, 1, 3, 2, 1, 0, 1, 3, 1, 1, 1, 1],
                [2, 1, 2, 1, 1, 0, 0, 1, 1, 2, 4, 3, 0, 2, 1],
                [1, 2, 1, 0, 1, 2, 1, 1, 1, 1, 3, 1, 2, 1, 2],
                [2, 1, 1, 1, 4, 3, 1, 2, 1, 1, 3, 2, 5, 1, 0],
                [3, 1, 1, 1, 3, 3, 1, 2, 1, 2, 1, 5, 1, 1, 1],
                [2, 4, 3, 2, 3, 3, 3, 1, 3, 3, 2, 1, 2, 4, 2],
            ]
        )  # ground truth is generated by manually use DecionTree to extract thresholds

        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 3, "random_state": 2020},
            mode="multi",
            split_unselected_feat=True,
        )
        tree_discretiser = dt_multi.fit(
            feat_names=[
                "sepal length (cm)",
                "sepal width (cm)",
                "petal length (cm)",
                "petal width (cm)",
            ],
            dataframe=get_iris_data,
            target_continuous=False,
            target="target",
        )
        output = tree_discretiser.transform(get_iris_data[["sepal width (cm)"]]).values
        assert (ground_truth == output.reshape(-1, 15)).all()

    def test_multi_fit(self, get_iris_data):
        ground_truth_petal_length = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0],
                [0, 0, 2, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 2, 2, 2, 2],
                [2, 0, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                [2, 1, 2, 1, 2, 2, 0, 1, 2, 2, 2, 2, 2, 2, 2],
                [2, 2, 2, 0, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )  # ground truth is generated by manually use DecionTree to extract thresholds

        ground_truth_petal_width = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2],
                [2, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 1, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )  # ground truth is generated by manually use DecionTree to extract thresholds

        iris = get_iris_data.copy(deep=True)

        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 3, "random_state": 2020}, mode="multi"
        )

        tree_discretiser = dt_multi.fit(
            feat_names=[
                "sepal length (cm)",
                "sepal width (cm)",
                "petal length (cm)",
                "petal width (cm)",
            ],
            dataframe=iris,
            target_continuous=False,
            target="target",
        )

        discretiser_petal_length = tree_discretiser.transform(
            iris[["petal length (cm)"]]
        ).values
        discretiser_petal_width = tree_discretiser.transform(
            iris[["petal width (cm)"]]
        ).values
        assert (
            ground_truth_petal_length == discretiser_petal_length.reshape(-1, 15)
        ).all()
        assert (
            ground_truth_petal_width == discretiser_petal_width.reshape(-1, 15)
        ).all()

    def test_no_unselected_feature(self, get_iris_data):
        ground_truth = get_iris_data[["sepal width (cm)"]]
        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 3, "random_state": 2020},
            mode="multi",
            split_unselected_feat=False,
        )
        tree_discretiser = dt_multi.fit(
            feat_names=[
                "sepal length (cm)",
                "sepal width (cm)",
                "petal length (cm)",
                "petal width (cm)",
            ],
            dataframe=get_iris_data,
            target_continuous=False,
            target="target",
        )
        output = tree_discretiser.transform(get_iris_data[["sepal width (cm)"]])

        assert all(ground_truth == output)

    def test_default_args(self):
        dt_multi = DecisionTreeSupervisedDiscretiserMethod()
        params = dt_multi.get_params()
        assert params["tree_params"]["max_depth"] == 2

    def test_transform_all_single(self, get_iris_data):
        data = get_iris_data.copy(deep=True)
        sepal_length = np.array(
            [
                [1, 1, 0, 0, 1, 1, 0, 1, 0, 1, 1, 0, 0, 0, 2],
                [2, 1, 1, 2, 1, 1, 1, 0, 1, 0, 1, 1, 1, 1, 0],
                [0, 1, 1, 2, 1, 1, 2, 1, 0, 1, 1, 0, 0, 1, 1],
                [0, 1, 0, 1, 1, 3, 3, 3, 2, 3, 2, 3, 1, 3, 1],
                [1, 2, 2, 2, 2, 3, 2, 2, 3, 2, 2, 2, 3, 2, 3],
                [3, 3, 3, 2, 2, 2, 2, 2, 2, 1, 2, 3, 3, 2, 2],
                [2, 2, 2, 1, 2, 2, 2, 3, 1, 2, 3, 2, 3, 3, 3],
                [3, 1, 3, 3, 3, 3, 3, 3, 2, 2, 3, 3, 3, 3, 2],
                [3, 2, 3, 3, 3, 3, 3, 2, 3, 3, 3, 3, 3, 3, 2],
                [3, 3, 3, 2, 3, 3, 3, 2, 3, 3, 3, 3, 3, 3, 2],
            ]
        )
        sepal_width = np.array(
            [
                [2, 1, 1, 1, 2, 3, 2, 2, 0, 1, 2, 2, 1, 1, 3],
                [3, 3, 2, 2, 2, 2, 2, 2, 1, 2, 1, 2, 2, 2, 1],
                [1, 2, 3, 3, 1, 1, 2, 2, 1, 2, 2, 0, 1, 2, 2],
                [1, 2, 1, 2, 1, 1, 1, 1, 0, 0, 0, 1, 0, 0, 0],
                [0, 1, 0, 0, 0, 1, 1, 0, 0, 0, 1, 0, 0, 0, 0],
                [1, 0, 1, 0, 0, 0, 0, 0, 0, 1, 2, 1, 0, 1, 0],
                [0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 1, 0, 1],
                [1, 0, 0, 0, 2, 1, 0, 1, 0, 0, 1, 1, 2, 0, 0],
                [1, 0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 2, 0, 0, 0],
                [1, 2, 1, 1, 1, 1, 1, 0, 1, 1, 1, 0, 1, 2, 1],
            ]
        )
        petal_length = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 1, 1, 2, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 2, 1, 1],
                [1, 2, 2, 1, 1, 1, 1, 1, 2, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2],
                [2, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )
        petal_width = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2],
                [2, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 1, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )

        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            mode="single", tree_params={"max_depth": 2, "random_state": 2020}
        )
        tree_discretiser = dt_multi.fit(
            feat_names=[
                "sepal length (cm)",
                "sepal width (cm)",
                "petal length (cm)",
                "petal width (cm)",
            ],
            dataframe=data,
            target="target",
            target_continuous=False,
        )
        output_df = tree_discretiser.transform(data)

        assert (
            output_df["sepal length (cm)"].values.reshape(-1, 15) == sepal_length
        ).all()
        assert (
            output_df["sepal width (cm)"].values.reshape(-1, 15) == sepal_width
        ).all()
        assert (
            output_df["petal length (cm)"].values.reshape(-1, 15) == petal_length
        ).all()
        assert (
            output_df["petal width (cm)"].values.reshape(-1, 15) == petal_width
        ).all()

    def test_transform_all_multi(self, get_iris_data):
        data = get_iris_data.copy(deep=True)
        sepal_length = data["sepal length (cm)"]
        sepal_width = data["sepal width (cm)"]
        petal_length = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0],
                [0, 0, 2, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 2, 2, 2, 2],
                [2, 0, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
                [2, 1, 2, 1, 2, 2, 0, 1, 2, 2, 2, 2, 2, 2, 2],
                [2, 2, 2, 0, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )
        petal_width = np.array(
            [
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2],
                [2, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 1, 2, 2, 2, 1, 1],
                [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],
            ]
        )

        dt_multi = DecisionTreeSupervisedDiscretiserMethod(
            mode="multi", tree_params={"max_depth": 3, "random_state": 2020}
        )
        tree_discretiser = dt_multi.fit(
            feat_names=[
                "sepal length (cm)",
                "sepal width (cm)",
                "petal length (cm)",
                "petal width (cm)",
            ],
            dataframe=data,
            target="target",
            target_continuous=False,
        )
        output_df = tree_discretiser.transform(data)

        assert (output_df["sepal length (cm)"].values == sepal_length).all()
        assert (output_df["sepal width (cm)"].values == sepal_width).all()
        assert (
            output_df["petal length (cm)"].values.reshape(-1, 15) == petal_length
        ).all()
        assert (
            output_df["petal width (cm)"].values.reshape(-1, 15) == petal_width
        ).all()

    def test_transform_shuffled_indices(self, get_iris_data):
        data = get_iris_data.copy(deep=True)
        ground_truth = np.array(
            [
                [0, 0, 1, 0, 1],
                [0, 0, 0, 0, 0],
                [0, 1, 1, 2, 1],
                [2, 1, 1, 0, 1],
                [1, 2, 1, 0, 2],
                [2, 2, 2, 0, 0],
                [1, 1, 0, 2, 2],
                [1, 2, 0, 2, 2],
                [2, 2, 0, 1, 2],
                [1, 1, 2, 1, 1],
            ]
        )

        sample_data = data[["sepal length (cm)", "target"]].sample(
            50, random_state=2021
        )
        dt_single = DecisionTreeSupervisedDiscretiserMethod(
            tree_params={"max_depth": 2},
            mode="single",
        )
        dt_single.fit(
            feat_names=["sepal length (cm)"],
            dataframe=sample_data,
            target_continuous=False,
            target="target",
        )
        discretiser_output = dt_single.transform(sample_data[["sepal length (cm)"]])
        assert discretiser_output.isnull().values.sum() == 0
        assert (ground_truth == discretiser_output.values.reshape(10, 5)).all()


text:
"""
Custom model evaluation methods.

@author: Riley Smith
Created: 10-17-2021
"""
import numpy as np
from sklearn import metrics

class FixedFPREvaluation():
    """
    Class for evaluating classifier based on a fixed tolerable false alarm rate.

    The evaluation metric in this case is the percentage of retained true
    positives at or below the tolerable false alarm rate. Higher values are
    better. This class is intended to be used as the scoring function for
    Sklearn's GridSearchCV.
    """
    def __init__(self, fpr):
        """
        Parameters
        ----------
        fpr : float
            The tolerable false alarm rate (between 0 and 1).
        """
        self._fpr = fpr

    def __call__(self, estimator, X, y):
        """
        Assumes you are dealing with binary classification.

        Parameters
        ----------
        estimator : Sklearn estimator
            The estimator to evaluate.
        X : ndarray
            Validation data of shape (n_samples, n_features).
        y : ndarray
            Ground truth for X. Has shape (n_samples,)
        """
        # Get the probabilities from the estimator
        prob = estimator.predict_proba(X)
        # Get ROC curve for predicting home team first
        home_fpr, home_tpr, _ = metrics.roc_curve(y, prob[:,-1])
        # Find where fpr is below threshold and get max tpr there
        max_home_tpr = home_tpr[np.where(home_fpr < self._fpr)].max()
        # Weight by number of games in which home team won
        max_home_tpr *= y.mean()
        # Now do the same for away
        away_fpr, away_tpr, _ = metrics.roc_curve(1 - y, prob[:,0])
        max_away_tpr = away_tpr[np.where(away_fpr < self._fpr)].max()
        max_away_tpr *= (1 - y.mean())

        return max_home_tpr + max_away_tpr

class ThresholdedAccuracy():
    """
    Evaluate a classifier based on its accuracy above a certain predicted
    probability threshold.
    """
    def __init__(self, threshold):
        """
        Parameters
        ----------
        threshold : float
            The predicted probability threshold above which you will actually
            make a call on a game. Should be in the range (0,1).
        """
        self._threshold = threshold

    def __call__(self, estimator, X, y):
        """
        Assumes you are dealing with binary classification.

        Parameters
        ----------
        estimator : Sklearn estimator
            The estimator to evaluate.
        X : ndarray
            Validation data of shape (n_samples, n_features).
        y : ndarray
            Ground truth for X. Has shape (n_samples,)
        """
        # Get predictions from estimator
        preds = estimator.predict(X)
        # Get the probabilities from the estimator
        prob = estimator.predict_proba(X)
        # Grab only those that are above the threshold
        home_refined_pred = preds[np.where(prob[:,-1] > self._threshold)]
        home_refined_true = y[np.where(prob[:,-1] > self._threshold)]
        # Compute accuracy and weight by number of surviving games
        home_acc = metrics.accuracy_score(home_refined_true, home_refined_pred)
        home_acc *= home_refined_pred.size
        home_acc = 0 if np.isnan(home_acc) else home_acc
        # Do the same for away
        away_refined_pred = preds[np.where(prob[:,0] > self._threshold)]
        away_refined_true = y[np.where(prob[:,0] > self._threshold)]
        away_acc = metrics.accuracy_score(1 - away_refined_true, away_refined_pred)
        away_acc *= away_refined_pred.size
        away_acc = 0 if np.isnan(away_acc) else away_acc
        # Return the weighted average
        if home_refined_pred.size + away_refined_pred.size == 0:
            return np.nan
        return (home_acc + away_acc) / (home_refined_pred.size + away_refined_pred.size)

max_stars_repo_name: rileypsmith/nba-game-simulator

max_issues_repo_name: rileypsmith/nba-game-simulator

max_forks_repo_name: rileypsmith/nba-game-simulator


text:
import os
import torch
import numpy as np
import matplotlib.pyplot as plt
import glob

from PIL import Image
from tqdm import tqdm
from torch.utils import data
import torchvision

class VOC2012_Dataset(data.Dataset):
    n_classes = 21
    means = [0.485, 0.456, 0.406]
    stds = [0.229, 0.224, 0.225]

    def __init__(self, root, img_size=(512, 512), split='train', transforms=None):
        assert split in ['train', 'trainval', 'val']
        self.root = root
        self.img_size = img_size
        self.names = self.read_names(f'{self.root}/ImageSets/Segmentation/{split}.txt')
        self.img_paths = [f'{self.root}/JPEGImages/{name}.jpg' for name in self.names]
        self.label_paths = [f'{self.root}/EncodedSegMap/{name}.png' for name in self.names]
        self.transforms = transforms

        # Encode segmap
        existed_encoded_segmaps = set([os.path.basename(path) for path in glob.glob(f'{self.root}/EncodedSegMap/*.png')])
        targeted_segmaps = set([os.path.basename(path) for path in self.label_paths])
        if not targeted_segmaps.issubset(existed_encoded_segmaps):
            src_paths = [f'{self.root}/SegmentationClass/{name}.png' for name in self.names]
            self.pre_encode_segmap(src_paths, self.label_paths)

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

    def __getitem__(self, index):
        img = Image.open(self.img_paths[index])
        label = Image.open(self.label_paths[index])
        if self.transforms == None:
            img, label = self.default_transform_img_label(img, label)
        else:
            img, label = self.transforms(img, label)
        label[label == 255] = 0
        return img, label

    def default_transform_img_label(self, img, label):
        transforms = torchvision.transforms.Compose(
            [
                torchvision.transforms.ToTensor(),
                torchvision.transforms.Normalize(self.means, self.stds),
            ]
        )
        img = transforms(img.resize(self.img_size))
        label = np.array(label.resize(self.img_size, Image.NEAREST))
        label = torch.from_numpy(label).long()
        return img, label

    @staticmethod
    def get_pascal_labels():
        """Load the mapping that associates pascal classes with label colors
        Returns:
            np.ndarray with dimensions (21, 3)
        """
        return np.asarray(
            [
                [0, 0, 0],
                [128, 0, 0],
                [0, 128, 0],
                [128, 128, 0],
                [0, 0, 128],
                [128, 0, 128],
                [0, 128, 128],
                [128, 128, 128],
                [64, 0, 0],
                [192, 0, 0],
                [64, 128, 0],
                [192, 128, 0],
                [64, 0, 128],
                [192, 0, 128],
                [64, 128, 128],
                [192, 128, 128],
                [0, 64, 0],
                [128, 64, 0],
                [0, 192, 0],
                [128, 192, 0],
                [0, 64, 128],
            ]
        )

    @staticmethod
    def encode_segmap(mask):
        """Encode segmentation label images as pascal classes
        Args:
            mask (np.ndarray): raw segmentation label image of dimension
              (M, N, 3), in which the Pascal classes are encoded as colours.
        Returns:
            (np.ndarray): class map with dimensions (M,N), where the value at
            a given location is the integer denoting the class index.
        """
        mask = mask.astype(int)
        label_mask = np.zeros((mask.shape[0], mask.shape[1]), dtype=np.int16)
        for ii, label in enumerate(VOC2012_Dataset.get_pascal_labels()):
            label_mask[np.where(np.all(mask == label, axis=-1))[:2]] = ii
        label_mask = label_mask.astype(int)
        return label_mask

    @staticmethod
    def decode_segmap(label_mask):
        """Decode segmentation class labels into a color image
        Args:
            label_mask (np.ndarray): an (M,N) array of integer values denoting
              the class label at each spatial location.
            plot (bool, optional): whether to show the resulting color image
              in a figure.
        Returns:
            (np.ndarray, optional): the resulting decoded color image.
        """
        label_colours = VOC2012_Dataset.get_pascal_labels()
        r = label_mask.copy()
        g = label_mask.copy()
        b = label_mask.copy()
        for ll in range(0, VOC2012_Dataset.n_classes):
            r[label_mask == ll] = label_colours[ll, 0]
            g[label_mask == ll] = label_colours[ll, 1]
            b[label_mask == ll] = label_colours[ll, 2]
        rgb = np.zeros((label_mask.shape[0], label_mask.shape[1], 3))
        rgb[:, :, 0] = r / 255.0
        rgb[:, :, 1] = g / 255.0
        rgb[:, :, 2] = b / 255.0
        return rgb

    @staticmethod
    def denormalize(tensor):
        return UnNormalize(mean=VOC2012_Dataset.means, std=VOC2012_Dataset.stds)(tensor)

    def read_names(self, names_path):
        names = []
        with open(names_path) as f:
            lines = f.readlines()
            for line in lines:
                names.append(''.join(line.split()))
        return names

    def pre_encode_segmap(self, src_paths, dst_paths):
        os.makedirs(os.path.dirname(dst_paths[0]), exist_ok=True)
        pbar = tqdm(total=len(src_paths))
        for src_path, dst_path in zip(src_paths, dst_paths):
            label = np.asarray(Image.open(src_path).convert('RGB'))
            encoded_label = VOC2012_Dataset.encode_segmap(label)
            encoded_label = encoded_label.astype(np.uint8)
            encoded_label = Image.fromarray(encoded_label)
            encoded_label.save(dst_path, format="png")
            pbar.update()
        pbar.close()

class UnNormalize(object):
    def __init__(self, mean, std):
        self.mean = mean
        self.std = std
    def __call__(self, tensor):
        for t, m, s in zip(tensor, self.mean, self.std):
            t.mul_(s).add_(m)
        return tensor

if __name__ == '__main__':
    train_data = VOC2012_Dataset(root="G:/Codes/RealTime-Segementation/datasets/VOC2012")
    train_dataloader = data.DataLoader(train_data, batch_size=16, shuffle=True)
    for i, (img, label) in enumerate(train_dataloader):
        img = VOC2012_Dataset.denormalize(img[0]).numpy().transpose([1, 2, 0])
        print(label.shape, label.min(), label.max())
        label = VOC2012_Dataset.decode_segmap(label.numpy()[0])
        print(img.shape, img.min(), img.max())
        print(label.shape, label.min(), label.max())
        plt.figure(figsize=(10, 10))
        plt.subplot(121), plt.title("Image"), plt.imshow(img)
        plt.subplot(122), plt.title("SegMap"), plt.imshow(label)
        plt.show()
        break


text:
import itertools
import multiprocessing as mp
from contextlib import contextmanager
from typing import Optional, Dict, List, Union

import h5py
import tables
import numpy as np
import pandas as pd

from gmhazard_calc import utils
from gmhazard_calc import constants as const
from gmhazard_calc.im import IM
from .BaseDB import BaseDB, check_open

def get_station_ruptures(imdb_ffp: str, station: str):
    """Gets all ruptures names for a specific station

    Note I: Only for internal use by the add_rupture_lookup function
    Note II: Has to be outside of the class, as it is called using
    MP which does not support class or static methods
    """
    with IMDB.get_imdb(imdb_ffp) as db:
        if isinstance(db, IMDBNonParametric):
            im_data = db.im_data(station)
            if im_data is not None:
                return im_data.index.levels[0].values.astype(str)
        elif isinstance(db, IMDBParametric):
            im_data = db.im_data(station)
            if im_data is not None:
                return im_data.index.values.astype(str)

class IMDB(BaseDB):

    # Attributes of the databases
    IMDB_TYPE = "imdb_type"
    IMS_KEY = "ims"

    def __init__(
        self, db_ffp: str, writeable: bool = False, source_type: const.SourceType = None
    ):
        super().__init__(db_ffp, writeable=writeable)

        # Set the source type
        if source_type is None:
            with h5py.File(self.db_ffp, mode="r") as h5_store:
                if "source_type" in h5_store.attrs.keys():
                    self._source_type = const.SourceType(
                        h5_store.attrs["source_type"].decode()
                    )
                else:
                    raise TypeError(
                        "Database does not have a source_type, "
                        "must specify one in the command"
                    )
        else:
            self._source_type = const.SourceType(source_type)

    @property
    def source_type(self) -> const.SourceType:
        return self._source_type

    @property
    @check_open
    def imdb_type(self) -> const.IMDataType:
        """Either parametric or non_parametric"""
        im_format_type = self.attributes.get(self.IMDB_TYPE)

        if im_format_type is None:
            raise Exception(
                "The provided db file is not a valid IMBD as it is"
                " missing the im_type attribute, which specifies "
                "whether the data is in a parametric format or "
                "non-parametric format."
            )
        return const.IMDataType(im_format_type)

    @property
    @check_open
    def ims(self) -> np.ndarray:
        """Array of IMs stored in the db"""
        return self.attributes[self.IMS_KEY]

    @check_open
    def sites(self) -> pd.DataFrame:
        """Retrieves the sites which were processed during creation of this db.

        Note: This does not mean that data exists for every station, as some stations
        do not have any data associated with them, hence there is no
        entry in im_data for these.

        Returns
        -------
        pd.DataFrame:
            index = station name, columns = [lat, lon]
        """
        return self._db["sites"]

    @check_open
    def rupture_data(self, rupture_name: str, im: Optional[IM] = None):
        """
        Gets the IM values for all stations for a specific rupture

        Parameters
        ----------
        rupture_name: str
            Rupture of interest
        im: IM, optional

        Returns
        -------
        im_values: pd.DataFrame or pd.Series
            Returns dataframe if no im is specified, format
                multi index = (station, simulation), columns = IM values
            If an im is specified returns a series of format
                multi index = (station, simulation), value = im value
        """
        stations = None
        try:
            stations = self._db.get(self.get_rupture_lookup_path(rupture_name))
        except KeyError:
            pass

        if stations is None or stations.size == 0:
            return None
        stations = stations.values.astype(str)

        # Collect the data from each station
        # If this is ever to slow, this can be easily multi-processed using
        # pool and writing a function that just reads the IM data for a
        # given imdb_ffp and station.
        station_data = []
        for cur_station in stations:
            cur_im_data = self.im_data(cur_station, im=im).loc[rupture_name]
            if im is None and isinstance(cur_im_data, pd.Series):
                cur_im_data = cur_im_data.to_frame().T

            station_data.append(cur_im_data)

        df = pd.concat(station_data, axis=0, keys=stations, sort=False)
        return df

    def rupture_names(self) -> np.ndarray:
        """Returns an array with the names of all the ruptures
        for which there is a data in the IMDB

        Implemented by the specialisation classes
        """
        raise NotImplementedError

    def im_data(self, station: str, im: Optional[IM] = None):
        """Retrieves the IM data, implemented by
        the specialisation classes
        """
        raise NotImplementedError

    @check_open(writeable=True)
    def write_sites(self, site_df: pd.DataFrame) -> None:
        """Write the sites

        Parameters
        ----------
        site_df: pd.DataFrame
            index = station name, columns = [lat, lon]
        """
        if utils.check_names(["lon", "lat"], site_df.columns.values):
            self._db["sites"] = site_df

    @check_open(writeable=True)
    def write_im_data(self, station_name: str, im_df: pd.DataFrame) -> None:
        """Writes the IM data for the specified station

        Parameters
        ----------
        station_name: str
        im_df: pd.DataFrame
            The dataframe to write
        """
        self._db[self.get_im_data_path(station_name)] = im_df

    @check_open(writeable=True)
    def add_im_data(self, station_name: str, im_df: pd.DataFrame) -> None:
        """Adds the given IM df to the existing IM df for the specified station
        If no entry for this station exists one is created

        Parameters
        ----------
        station_name: str
        im_df: pd.DataFrame
            The dataframe to write
        """
        path = self.get_im_data_path(station_name)

        # Check if there is already an existing entry for this station
        # Using try/except instead of checking .keys as it is sig. faster
        cur_df = None
        try:
            cur_df = self._db[path]
        except KeyError:
            pass

        if cur_df is not None:
            # Check that the number of ruptures match
            assert cur_df.shape[0] == im_df.shape[0]

            # Delete the existing one
            self._db.remove(path)

            # Create updated
            cur_df = pd.concat([cur_df, im_df], axis=1)

        # Write
        self.write_im_data(station_name, cur_df)

    @check_open(writeable=True)
    def write_rupture_lookup(self, rupture_lookup: Dict[str, List[str]]):
        for cur_rup, cur_stations in rupture_lookup.items():
            # Save as series (index is meaningless)
            self._db[self.get_rupture_lookup_path(cur_rup)] = pd.Series(
                data=cur_stations
            )

    @staticmethod
    def get_im_data_path(station: str) -> str:
        """Returns the database path for the IM data of the specified station"""
        return f"/im_data/station_{station}"

    @staticmethod
    def get_rupture_lookup_path(rupture_name: str) -> str:
        """Returns the database path for the event based data links"""
        return f"/rupture_lookup/rupture_{rupture_name}"

    @staticmethod
    @contextmanager
    def get_imdb(imdb_ffp: str, writeable: bool = False) -> "IMDB":
        """Creates a contextmanager for the appropriate IMDB instance,
        based on the im_db_type
        """
        with h5py.File(imdb_ffp, mode="r") as h:
            imdb_type = h.attrs[IMDB.IMDB_TYPE].decode()
        if imdb_type == const.IMDataType.non_parametric.value:
            imdb = IMDBNonParametric(imdb_ffp, writeable=writeable)
        elif imdb_type == const.IMDataType.parametric.value:
            imdb = IMDBParametric(imdb_ffp, writeable=writeable)
        else:
            raise Exception(
                f"The specified IMDB has a im_db_type attribute of {imdb_type}, "
                f"which is not valid, has to be either "
                f"{const.IMDataType.non_parametric.value} or "
                f"{const.IMDataType.parametric.value}"
            )
        imdb.open()
        yield imdb
        imdb.close()

    @staticmethod
    def add_rupture_lookup(db_ffp: str, n_procs: int):
        """
        Add a lookup to get stations for each rupture and to get the full list of rupture names
        WARNING:
        Could take a long time for large DB's such as a Distributed Seismicity db
        Should only be used for a Fault db

        Parameters
        ----------
        db_ffp: str
            Full file path to the db file to add the rupture lookup
        n_procs: int
            Number of processes to use
        """
        with IMDB.get_imdb(db_ffp) as db:
            stations = db.sites().index.values

        # Iterate over all stations and get all ruptures for each station
        print("Collecting ruptures for each station")
        with mp.Pool(n_procs) as pool:
            result = pool.starmap(
                get_station_ruptures,
                [(db_ffp, cur_station) for cur_station in stations],
            )

        # Invert the result, i.e. for each rupture get all stations
        print("Computing rupture lookup")
        rupture_lookup = {}
        for cur_station, cur_ruptures in zip(stations, result):
            if cur_ruptures is not None:
                for cur_rup in cur_ruptures:
                    if cur_rup in rupture_lookup.keys():
                        rupture_lookup[cur_rup].append(cur_station)
                    else:
                        rupture_lookup[cur_rup] = [cur_station]

        # Write
        print("Writing")
        with IMDB.get_imdb(db_ffp, writeable=True) as db:
            db.write_rupture_lookup(rupture_lookup)
            db.write_attributes(
                rupture_names=np.asarray(list(rupture_lookup.keys()), dtype=str)
            )

    @check_open
    def get_stored_stations(self):
        return [
            stat.split("im_data/")[-1].replace("station_", "")
            for stat in self._db.keys()
        ]

class IMDBParametric(IMDB):
    def __init__(
        self, db_ffp: str, writeable: bool = False, source_type: const.SourceType = None
    ):
        super().__init__(db_ffp, writeable=writeable, source_type=source_type)

    @property
    def imdb_type(self) -> const.IMDataType:
        """Required for when database is opened in write mode"""
        if self.writeable:
            return const.IMDataType.parametric
        else:
            return super().imdb_type

    @check_open
    def _ruptures(self) -> pd.Series:
        """Returns a dataframe of all the ruptures in the IMDB

        Format: index = imdb_rupture_id, values = rupture_name
        """
        return self._db["ruptures"]

    @check_open()
    def rupture_names(self) -> Union[None, np.ndarray]:
        """Returns an array with the names of all the ruptures
        for which there is a data in the IMDB

        Note: This will only return values if rupture based
        lookup has been added to this IMDB, otherwise returns None
        """
        return self.get_attributes().get("rupture_names")

    @check_open
    def im_data(
        self, station: str, im: Optional[Union[List[IM], IM]] = None, incl_within_between_sigma: bool = False,
    ) -> Union[pd.DataFrame, pd.Series, None]:
        """Retrieves the IM parameters for the ruptures
        at a specific site

        Parameters
        ----------
        station: str
        im: IM or list[IM], optional
            IM(s) of interest
            if this is unspecified then it is equivalent to setting all IMs
        incl_within_between_sigma: bool
            Boolean flag to determine to either extract mu and total standard deviation or
            mu, between-event and within-event standard deviation

        Returns
        -------
        im_params, pd.DataFrame or None
            If an im is specified returns a df of format
                index = rupture, columns = [mu, sigma]
            Otherwise returns a dataframe of format
                index = rupture,
                columns = [im_1_mean, im_1_std, im_2_mean, im_2_std...]
            Returns None if there is no data in the IMDB for that station
        """
        df = None
        try:
            df = self._db.get(self.get_im_data_path(station))
        except KeyError:
            pass

        if df is None or df.size == 0:
            return None

        # Performance hack, replaces the following line of code
        # df.index = self._ruptures().loc[df.index.values, "rupture_name"].values.astype(str)
        with tables.open_file(self.db_ffp, mode="r") as fileh:
            lookup_indices = fileh.root.ruptures.table.read_coordinates(
                df.index.values
            )["values_block_0"].reshape(-1)
            df.index = (
                fileh.root.ruptures.meta.values_block_0.meta.table.read_coordinates(
                    lookup_indices
                )["values"].astype(str)
            )

        if im is not None:
            ims = im if isinstance(im, list) else [im]
            # Setting columns to extract from the DB
            if incl_within_between_sigma:
                single_im_columns = ["mu", "between_event_sigma", "within_event_sigma"]
                im_columns = list(
                    (itertools.chain(*[(f"{im}", f"{im}_sigma_inter", f"{im}_sigma_intra") for im in set(ims)]))
                )
            else:
                single_im_columns = ["mu", "sigma"]
                im_columns = list(
                    (itertools.chain(*[(f"{im}", f"{im}_sigma") for im in set(ims)]))
                )
            df = df.loc[:, im_columns]
            if len(ims) == 1:
                df.columns = single_im_columns
        return df

    @check_open(writeable=True)
    def write_attributes(
        self, erf_ffp: str = None, station_list_ffp: str = None, **kwargs
    ):
        """Writes the relevant parametric attributes in the database

        The erf_ffp and station_list_ffp, should be set
        when the database is written initially.
        """
        super().write_attributes(
            erf_ffp=erf_ffp,
            station_list_ffp=station_list_ffp,
            source_type=self._source_type.value,
            imdb_type=self.imdb_type.value,
            **kwargs,
        )

    @check_open(writeable=True)
    def write_rupture_data(self, rupture_df: pd.DataFrame) -> None:
        """Writes the rupture names to the database"""
        if utils.check_names(["rupture_name"], rupture_df.columns.values):
            self._db.put("ruptures", rupture_df, format="t")

class IMDBNonParametric(IMDB):
    def __init__(
        self,
        db_ffp: str,
        writeable: Optional[bool] = False,
        source_type: const.SourceType = None,
    ):
        super().__init__(db_ffp, writeable=writeable, source_type=source_type)
        self.writeable = writeable

        self._attrs = None

    @property
    def imdb_type(self) -> const.IMDataType:
        """Required for when database is opened in write mode"""
        if self.writeable:
            return const.IMDataType.non_parametric
        else:
            return super().imdb_type

    def rupture_names(self) -> np.ndarray:
        """Returns an array with the names of all the ruptures
        for which there is a data in the IMDB
        """
        ruptures = self.get_attributes().get("rupture_names")
        if ruptures is not None:
            return ruptures

        return np.unique(
            np.array(
                list(
                    np.char.split(
                        self.simulations().values.astype(str), "_", maxsplit=1
                    )
                )
            )[:, 0]
        )

    @check_open
    def simulations(self) -> pd.Series:
        """Retrieves the simulations in this imdb

        Returns
        -------
        pd.Series
            Index = simulation index, values = simulation names
        """
        return self._db["simulations"]

    @check_open
    def im_data(
        self, station: str, im: Optional[Union[List[IM], IM]] = None
    ) -> Union[None, pd.DataFrame, pd.Series]:
        """Retrieves the IM dataframe for all
        simulations for the specified station

        Parameters
        ----------
        station: str
        im: IM or list[IM], optional
            IM(s) of interest
            if this is unspecified then it is equivalent to setting all IMs

        Returns
        -------
        im_values: pd.DataFrame or pd.Series
            If a single im is specified returns a series of format
                multi index = (rupture, simulation), value = im value
            Otherwise Returns dataframe with the specified format
                multi index = (rupture, simulation), columns = IM values
        """
        df = None
        try:
            df = self._db.get(self.get_im_data_path(station))
        except KeyError:
            pass

        if df is None or df.size == 0:
            return None

        simulations = self.simulations()
        df["realisation"] = simulations.iloc[df.index.values].values
        df["fault"] = np.stack(np.char.split(df.realisation.values.astype(str), "_", maxsplit=1))[:, 0]
        df.sort_values(["fault", "realisation"], inplace=True)

        # Create hierarchical dataframe index (fault, realisation)
        df.index = pd.MultiIndex.from_frame(df.loc[:, ["fault", "realisation"]])
        df.drop(columns=["fault", "realisation"], inplace=True)

        if im is not None:
            return df[im]
        return df

    @check_open(writeable=True)
    def write_attributes(self, **kwargs):
        """Writes the relevant parametric attributes in the database"""
        super().write_attributes(
            imdb_type=np.string_(self.imdb_type.value),
            source_type=np.string_(self._source_type.value),
            **kwargs,
        )

    @check_open(writeable=True)
    def write_simulations(self, simulations: pd.Series) -> None:
        """Writes the simulations

        Parameters
        ----------
        simulations: pd.Series
            index = simulation id, values = simulation name
        """
        if not isinstance(simulations, pd.Series):
            raise ValueError(
                "Has to be a pd.Series with id as index and simulation names as values"
            )
        self._db["simulations"] = simulations

max_stars_repo_name: ucgmsim/seistech

max_issues_repo_name: ucgmsim/seistech

max_forks_repo_name: ucgmsim/seistech


text:
# Src: Andrew's tune_tf2

import os
import os.path as osp
import numpy as np
import ray
from ray import tune
from yacs.config import CfgNode as CN
import torch

from src.config.default import get_cfg_defaults, unflatten
from src.runner import Runner

class tuneNDT(tune.Trainable):
    """ A wrapper class that allows `tune` to interface with NDT.
    """

    def setup(self, config):
        yacs_cfg = self.convert_tune_cfg(config)
        self.epochs_per_generation = yacs_cfg.TRAIN.TUNE_EPOCHS_PER_GENERATION
        self.warmup_epochs = yacs_cfg.TRAIN.TUNE_WARMUP
        self.runner = Runner(config=yacs_cfg)
        self.runner.load_device()
        self.runner.load_train_val_data_and_masker()
        num_hidden = self.runner.setup_model(self.runner.device)
        self.runner.load_optimizer(num_hidden)

    def step(self):
        num_epochs = self.epochs_per_generation
        # the first generation always completes ramping (warmup)
        if self.runner.count_updates < self.warmup_epochs:
            num_epochs += self.warmup_epochs
        for i in range(num_epochs):
            metrics = self.runner.train_epoch()
        return metrics

    def save_checkpoint(self, tmp_ckpt_dir):
        path = osp.join(tmp_ckpt_dir, f"{self.runner.config.VARIANT}.{self.runner.count_checkpoints}.pth")
        self.runner.save_checkpoint(path)
        return path

    def load_checkpoint(self, path):
        self.runner.load_checkpoint(path)

    def reset_config(self, new_config):
        new_cfg_node = self.convert_tune_cfg(new_config)
        self.runner.update_config(new_cfg_node)
        return True

    def convert_tune_cfg(self, flat_cfg_dict):
        """Converts the tune config dictionary into a CfgNode for LFADS.
        """
        cfg_node = get_cfg_defaults()

        flat_cfg_dict['CHECKPOINT_DIR'] = osp.join(self.logdir, 'ckpts')
        flat_cfg_dict['TENSORBOARD_DIR'] = osp.join(self.logdir, 'tb')
        flat_cfg_dict['LOG_DIR'] = osp.join(self.logdir, 'logs')
        flat_cfg_dict['TRAIN.TUNE_MODE'] = True
        cfg_update = CN(unflatten(flat_cfg_dict))
        cfg_node.merge_from_other_cfg(cfg_update)

        return cfg_node


text:
import abc
import math

import numpy as np

class CodeLength(metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def length(self, x):
        pass

def zero2small(v, small=1e-5):
    return v if v > 0 else small

class SDNML(CodeLength):
    def __init__(self, discounting_param=0.1, order=2):
        self.discounting_param = discounting_param
        self.order = order
        self._stats = dict(e2_sum=0, tau=np.atleast_2d(0.1).T)
        self._xs = [0 for _ in range(self.order)]
        self._time = 1

    def _update_stats(self, xs):
        if 'v' not in self._stats:
            self._stats['v'] = np.full(
                (self.order, self.order), 1/self.order)
        if 'm' not in self._stats:
            self._stats['m'] = np.full(
                (self.order, self.order), 1/self.order)
        r = self.discounting_param  # alias
        c = r*xs.T.dot(self._stats['v']).dot(xs)
        v = (1/(1-r))*self._stats['v']
        v -= (
            (r/(1-r)) *
            (self._stats['v'].dot(xs).dot(xs.T).dot(self._stats['v']))/(1-r+c))
        m = (1-r)*self._stats['m']+r*xs*xs.T
        a = v.dot(m)
        e = xs-a.T.dot(xs)
        d = c/(1-r+c)
        self._stats['c'] = c
        self._stats['a'] = a
        self._stats['v'] = v
        self._stats['m'] = m
        self._stats['d'] = d
        self._stats['e2_sum'] += e.T.dot(e)
        tau = self._stats['e2_sum']/self._time
        self._stats['prev_tau'] = self._stats['tau']
        self._stats['tau'] = tau

    def length(self, x):
        """Calculate SDNML code-length.

        Parameters
        ----------
        x : float
            Sample.

        Returns
        -------
        code_length : float
            Code length of the sample.
        """
        self._xs.pop(0)
        self._xs.append(x)
        xs = np.atleast_2d(self._xs).T
        self._update_stats(xs)
        code_length = 0
        if self._time > 1:
            code_length = (math.log(math.pi)/2) \
                          - math.log(zero2small(1-self._stats['d']))
            code_length -= math.lgamma((self._time-1)/2)
            code_length += math.lgamma((self._time)/2)
            code_length += (math.log(self._time-1) +
                            math.log(zero2small(self._stats['prev_tau'])))/2
            code_length += (self._time)/2 * (
                math.log(zero2small((self._time)*self._stats['tau'])) -
                math.log(zero2small((self._time-1)*self._stats['prev_tau']))
            )
        self._time += 1
        return code_length

max_stars_repo_name: hyusuk/rtchange

max_issues_repo_name: hyusuk/rtchange

max_forks_repo_name: hyusuk/rtchange


text:
import numpy as np

LOWER_WHITE_THRES = [0,200,0]
HIGHER_WHITE_THRES = [255,255,255]

LOWER_YELLOW_THRES = [20, 80, 80]
HIGHER_YELLOW_THRES = [30, 255, 255]

GAUSSIAN_KERNEL = 3  # ORG 3
CANNY_LOW_THRES = 70  # ORG 70
CANNY_HIGH_THRES = 200  # ORG 200

LEFT_ROI = 0.095
RIGHT_ROI = 0.96
MIDDLE_LEFT_ROI = 0.47
MIDDLE_RIGHT_ROI = 0.55
MIDDLE_HEIGHT = 0.59

HOUGH_ROU = 1  # pixel Units
HOUGH_THETA = np.pi / 180  # Radian
HOUGH_THRESHOLD = 15  # Min Number of votes / intersection in a given grid
HOUGH_MIN_LINE_LENGT = 10  # After what distance should a line considered as a line
HOUGH_MAX_LINE_GAP = 40  # maximum distance between segments that is allowed to be connected into a single line.

MAX_LANE_HEIGHT = 0.62 # ORG 0.61

LEFT_LINE_COLOR = [255, 0, 0]
RIGHT_LINE_COLOR = [0, 0, 255]
LINE_THICKNESS = 2

LANE_THICKNESS = 8
LANE_COLOR = [0, 255, 0]

MAX_LEN_QUE = 10 # Take the average of last 10 Frames


text:
"""This module implements the Tensor entity"""

# INTEL CONFIDENTIAL
#
# Copyright (C) 2021 Intel Corporation
#
# This software and the related documents are Intel copyrighted materials, and
# your use of them is governed by the express license under which they were provided to
# you ("License"). Unless the License provides otherwise, you may not use, modify, copy,
# publish, distribute, disclose or transmit this software or the related documents
# without Intel's prior written permission.
#
# This software and the related documents are provided as is,
# with no express or implied warranties, other than those that are expressly stated
# in the License.

from typing import Tuple

import numpy as np

from ote_sdk.entities.metadata import IMetadata

class TensorEntity(IMetadata):
    """
    Represents a metadata of tensor type in OTE.

    :param name: name of metadata
    :param numpy: the numpy data of the tensor
    """

    def __init__(self, name: str, numpy: np.ndarray):
        self.name = name
        self._numpy = numpy

    @property
    def numpy(self) -> np.ndarray:
        """Returns the numpy representation of the tensor."""
        return self._numpy

    @numpy.setter
    def numpy(self, value):
        self._numpy = value

    @property
    def shape(self) -> Tuple[int, ...]:
        """Returns the shape of the tensor."""
        return self._numpy.shape

    def __eq__(self, other):
        if isinstance(other, TensorEntity):
            return np.array_equal(self.numpy, other.numpy)
        return False

    def __str__(self):
        return f"{self.__class__.__name__}(name={self.name}, shape={self.shape})"

    def __repr__(self):
        return f"{self.__class__.__name__}(name={self.name}, numpy={self.numpy})"