import json import boto3 import faiss import numpy as np import pandas as pd import pyarrow as pa import pyarrow.compute as compute import streamlit as st from audience.functions.find_artist_data import get_artist_data from audience.functions.reorder_artist_data import reorder_artists @st.cache # make it more efficient to load pickled files into memory def preparing_environment() -> pd.DataFrame: # Dataframes for this app can be recreated using project related notebooks. # Inside these notebooks you will find cells that will produce required parquet files. # load objects for just genre model main_artist_df = pd.read_parquet("/var/app/models/algo_input_df_f.parquet") main_artist_df["search_string"] = main_artist_df["C_ARTIST_NAME"].str.lower() # load objects for showing related artists # related_artist_df = pd.read_parquet('related_artists_df_f.parquet') # tfidf_matrix_truncated = load('tfidf_matrix_truncated.npy') # faiss requirement is float32 data type # tfidf_matrix_truncated = np.float32(tfidf_matrix_truncated) return main_artist_df # return related_artist_df, main_artist_df # https://docs.streamlit.io/knowledge-base/deploy/authentication-without-sso def check_password() -> bool: """Returns `True` if the user had the correct password.""" def password_entered() -> None: """Checks whether a password entered by the user is correct.""" if ( st.session_state["username"] in st.secrets["passwords"] and st.session_state["password"] == st.secrets["passwords"][st.session_state["username"]] ): st.session_state["password_correct"] = True st.session_state["logged_in_user"] = st.session_state["username"] del st.session_state["password"] # don't store username + password del st.session_state["username"] else: st.session_state["password_correct"] = False if "password_correct" not in st.session_state: # First run, show inputs for username + password. st.text_input("Username", on_change=password_entered, key="username") st.text_input( "Password", type="password", on_change=password_entered, key="password" ) return False elif not st.session_state["password_correct"]: # Password not correct, show input + error. st.text_input("Username", on_change=password_entered, key="username") st.text_input( "Password", type="password", on_change=password_entered, key="password" ) if len(st.session_state["password"]) > 0: st.error("😕 User not known or password incorrect") return False else: # Password correct. return True def make_clickable(link: str) -> str: # target _blank to open new window # extract clickable text to display for your link return f'Open Spotify' # if we find matching username and password then we will show app if check_password(): # START OF ACTUAL APP CONTENT AFTER VALID PASSWORD HAS BEEN PROVIDED # load model parameters # Opening JSON file with open('/var/app/models/model_parameters.json', 'r') as openfile: model_params_dict = json.load(openfile) model_file_rows = model_params_dict['nr_of_artists'] model_file_features = model_params_dict['nr_of_features'] model_version = model_params_dict['model_prepared'] # get the logged-in user logged_in_user = st.session_state["logged_in_user"] dynamodb = boto3.resource("dynamodb") feedback_table = dynamodb.Table("dev-streamlit-audience-feedback") # Before starting script make sure that S3 bucket also contains required files # These files can be created using project related notebooks. # Depending on what kind of similarities have been created, fill dictionaries accordingly. approaches = { "c": ["C_GENRES", "C_FAN_COUNTRY_CODE", "C_BAND", "C_PRONOUN", "C_GENDER"], } index_sizes = { "c": 200, "e": 200, } descriptors = { "c": "This approaches is based on genre, top 3 streaming locations and band, pronoun and gender indicators", "e": "This approaches is based on genre, top 3 streaming locations and band, pronoun and gender indicators " "and artist popularity", } filenames = { "c": "c_lsh_faiss", "e": "e_lsh_faiss", } # index = faiss.read_index("/var/app/models/e_lsh_faiss") index = faiss.read_index(f'/var/app/models/{model_version}_lsh_faiss') nr_of_artists = 0 mmap_related = pa.memory_map("/var/app/models/related_artists_df.arrow") table_mmap_related = pa.ipc.RecordBatchFileReader(mmap_related).read_all() algo_input_df = preparing_environment() st.markdown( "

Similar artists demo


", unsafe_allow_html=True, ) st.markdown( "STEP 1: Using left side menu, select approach to use. Name of the approach shows what data " "features are used for finding similarity between artists.", unsafe_allow_html=True, ) st.markdown( "STEP 2: Then search for artist of interest. You can also use partial name. " "If there are multiple profiles for one artist then select the one " "with highest popularity. Read more about Spotify popularity " "" "here. Click on the select button. After that main screen will be filled with similar artists.", unsafe_allow_html=True, ) st.markdown( "STEP 3: Results. These are sorted by descending order. Meaning that most similar and " " most popular artists are at the top", unsafe_allow_html=True, ) # st.markdown( # "STEP 4: Filters. You can narrow down inital results by applying additional filters." # "These filters can be found at the bottom of left side menu.", # unsafe_allow_html=True, # ) st.markdown( "STEP 4: Listening. By clicking on the link under in the Spotify link column you will be" " redirected to the Spotify website where you can listen for that artist.
", unsafe_allow_html=True, ) with st.sidebar: # define filters inside left side of app window st.markdown("Specify inputs", unsafe_allow_html=True) algo_list = ["genre, streaming location, band, gender, pronoun"] algo_selection = st.selectbox("Select algorithm:", algo_list) artist = st.text_input("Type part of artist name to search for:", "") artist = artist.lower() if artist: st.write( "Search string used: ", artist, ". Top 5 results sorted by popularity.", unsafe_allow_html=True, ) df_result_search = ( algo_input_df[ algo_input_df["search_string"].str.contains(artist, na=False) ][["SPOTIFY_ARTIST_ID", "C_ARTIST_NAME", "C_POPULARITY"]] .sort_values(by=["C_POPULARITY"], ascending=False) .head() ) for header, column in zip( ["Artist", "Popularity", ""], st.columns((2, 1, 1)) ): column.subheader(header) for i, row in df_result_search.iterrows(): col1, col2, col3 = st.columns((2, 1, 1)) col1.write(row["C_ARTIST_NAME"]) col2.write(row["C_POPULARITY"]) if col3.button("Select", key=row["SPOTIFY_ARTIST_ID"]): st.session_state["spotify_id"] = row["SPOTIFY_ARTIST_ID"] spotify_id = st.session_state.get("spotify_id") # spotify_id = st.text_input('Copy here the spotify id:', '') # nr_of_artists = st.slider('How many similar artist to show?(max 50)', 5, 50, 10, 5) nr_of_artists = 30 # apply additional filters to first similarity results # st.write('Apply additional filters to results if needed:') # options_gender = algo_input_df["C_GENDER"].unique().tolist() # options_gender = pa.compute.unique(table_mmap_main["C_GENDER"]) # multi_options_gender = st.multiselect( # "Choose artist gender", options_gender # ) # options_pronoun = algo_input_df["C_PRONOUN"].unique().tolist() # options_pronoun = pa.compute.unique(table_mmap_main["C_PRONOUN"]) # multi_options_pronoun = st.multiselect( # "Choose artist pronoun", options_pronoun # ) # options_band = algo_input_df["C_BAND"].unique().tolist() # options_band = pa.compute.unique(table_mmap_main["C_BAND"]) # multi_options_band = st.multiselect("Is artist band", options_band) # country_filter = st.text_input("Type 2 letter country code:", "") # country_filter = country_filter.upper() if artist and spotify_id: memory_mapped_array = np.memmap( "/var/app/models/tfidf_matrix_truncated_raw.npy", dtype="float64", mode="r+", # shape=(5371054, 200), shape=(model_file_rows, model_file_features), ) st.write("Current similarity is based on ", algo_selection, ".") # select what data to use if algo_selection == "genre": selected_algo = "a" elif algo_selection == "genre, streaming location": selected_algo = "b" else: selected_algo = "e" try: artist_desc, artist_index = get_artist_data(algo_input_df, spotify_id) if selected_algo == "e": artist_row_data = memory_mapped_array[[artist_index]] artist_row = np.float32(artist_row_data) # distances, nearest_artists = index.search(tfidf_matrix_truncated[[artist_index]], nr_of_artists) distances, nearest_artists = index.search(artist_row, nr_of_artists) else: artist_row_data = memory_mapped_array[[artist_index]] artist_row = np.float32(artist_row_data) # distances, nearest_artists = index.search(tfidf_matrix_truncated[[artist_index]], nr_of_artists) distances, nearest_artists = index.search(artist_row, nr_of_artists) nearest_artists = nearest_artists.tolist()[0] response_tag = algo_input_df.loc[nearest_artists][ [ "C_ARTIST_NAME", "SPOTIFY_ARTIST_ID", "C_POPULARITY", "C_FAN_COUNTRY_CODE", "C_GENRES", "C_PRONOUN", "C_GENDER", "C_BAND", ] ] response_tag.columns = [ "ARTIST NAME", "SPOTIFY ID", "POPULARITY", "TOP COUNTRIES", "GENRES", "PRONOUN", "GENDER", "BAND", ] st.json(artist_desc) st.write("Showing most similar artists:") # create hyperlink using Spotify id response_tag["SPOTIFY ID"] = response_tag["SPOTIFY ID"].apply( make_clickable ) artists_from_algo = response_tag["ARTIST NAME"].tolist() # check for filters multi_options_gender = False multi_options_pronoun = False multi_options_band = False country_filter = False if multi_options_gender: response_tag = response_tag[ response_tag["Gender"].isin(multi_options_gender) ] if multi_options_pronoun: response_tag = response_tag[ response_tag["Pronoun"].isin(multi_options_pronoun) ] if multi_options_band: response_tag = response_tag[ response_tag["BAND"].isin(multi_options_band) ] if country_filter: response_tag = response_tag[ response_tag["TOP COUNTRIES"].str.contains(country_filter, na=True) ] st.write(country_filter) # Move 10 most popular artists at the top based on popularity from artist itself response_tag = reorder_artists( response_tag, artist_desc.get("popularity", 50) ) response_tag.rename( columns={ "ARTIST NAME": "Artist", "SPOTIFY ID": "Spotify link", "POPULARITY": "Spotify popularity", "TOP COUNTRIES": "Mostly listened in countries", "GENRES": "Related genres", "PRONOUN": "Pronoun", "GENDER": "Gender", "BAND": "Band", }, inplace=True, ) band_map_dict = {"0": "no", "1": "yes"} response_tag["Band"] = response_tag["Band"].map(band_map_dict) response_tag = response_tag.to_html(escape=False) st.write(response_tag, unsafe_allow_html=True) except KeyError: st.markdown( "Did not find matching Spotify artist.", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) st.write("Showing most similar artists according to Spotify:") try: rel_artists = table_mmap_related.filter( compute.equal(table_mmap_related["MAIN_ARTIST"], spotify_id) ) rel_artists = rel_artists.to_pandas() rel_artists.drop(columns=["MAIN_ARTIST"], inplace=True) # create hyperlink using Spotify id rel_artists["RELATED_ARTIST_ID"] = rel_artists["RELATED_ARTIST_ID"].apply( make_clickable ) rel_artists.rename( columns={ "ARTIST_NAME": "Artist", "RELATED_ARTIST_ID": "Spotify link", "FOLLOWERS_LATEST": "Nr of Spotify followers", }, inplace=True, ) artist_from_spotify = rel_artists["Artist"].tolist() rel_artists = rel_artists.style.set_precision(0) rel_artists = rel_artists.to_html(escape=False) st.write(rel_artists, unsafe_allow_html=True) # st.dataframe(rel_artists.style.set_precision(0)) common_artists = set(artists_from_algo).intersection(artist_from_spotify) st.markdown("
", unsafe_allow_html=True) st.write( f"Number of overlapping artists from both similarity results: {len(common_artists)}" ) if len(common_artists) > 0: st.write(",\n".join(common_artists)) # here user can give their feedback with st.form("my_form"): st.write("How do you rate this recommendation?") stars = st.selectbox( "How many stars you would give to this recommendation:", (1, 2, 3, 4, 5), 4, ) option = st.selectbox( "What would you improve?", ( "Nothing to improve", "Spotify’s recommendation match better my expectations", "Recommendation contains only well known/popular artists", "Recommendation contains too many unknown/low popularity artists", "Recommendation does not list artists that I know are similar to artist what I searched for", ), ) comments = st.text_area( label="Additional comments:", help="any additional comments(maximum 400 chars)", max_chars=400, ) submitted = st.form_submit_button("Submit") if submitted: st.write("Thank you for submitting feedback for this artist") response = feedback_table.put_item( # Data to be inserted Item={ "user": f"USER#{logged_in_user}", "artist": f"ARTIST#{spotify_id}", "feedback": option, "rating": stars, "comment": comments, } ) except NameError: st.markdown( "Did not find matching related artists from Spotify.", unsafe_allow_html=True, )