Update app.py

app.py

@@ -1,3 +1,5 @@
+import requests
+import streamlit
 from PIL import Image
 from utils import *
 from app_utils import *
@@ -23,18 +25,19 @@ def log_to_spotify():
             client_credentials_manager = SpotifyClientCredentials()
             sp = spotipy.Spotify(client_credentials_manager=client_credentials_manager)
             user_id = None
+            auth_manager = None
         else:
-            sp, user_id = new_get_client(session=st.session_state)
+            sp, user_id, auth_manager = new_get_client(session=st.session_state)
         if sp != None:
             legit_genres = sp.recommendation_genre_seeds()['genres']
-            st.session_state['login'] = (sp, user_id, legit_genres)
+            st.session_state['login'] = (sp, user_id, legit_genres, auth_manager)
             st.success('You are logged in.')
         else:
             legit_genres = None
     else:
-        sp, user_id, legit_genres = st.session_state['login']
+        sp, user_id, legit_genres, auth_manager = st.session_state['login']
         st.success('You are logged in.')
-    return sp, user_id, legit_genres
+    return sp, user_id, legit_genres, auth_manager
 
 
 @st.cache(suppress_st_warning=True)
@@ -61,27 +64,28 @@ def get_filtered_user_playlists(user_links):
     if n_playlists <= 1:
         return all_uris
     else:
-        # let the user uncheck playlists
-        st.markdown("Check boxes to select playlists from the selected users."
-                    "Note: to check all, first uncheck all (bug).")
-        columns = st.columns(np.ones(5))
-        with columns[1]:
-            check_all_playlists = st.button('Check all')
-        with columns[3]:
-            uncheck_all_playlists = st.button('Uncheck all')
-
-        if 'checkboxes' not in st.session_state.keys():
-            st.session_state['checkboxes_playlists'] = [True] * n_playlists
-
-        empty_checkboxes = wall_of_checkboxes(all_names, max_width=5)
-        if check_all_playlists:
-            st.session_state['checkboxes_playlists'] = [True] * n_playlists
-        if uncheck_all_playlists:
-            st.session_state['checkboxes_playlists'] = [False] * n_playlists
-        for i_emc, emc in enumerate(empty_checkboxes):
-            st.session_state['checkboxes_playlists'][i_emc] = emc.checkbox(all_names[i_emc], value=st.session_state['checkboxes_playlists'][i_emc])
-
-        filter_playlist = centered_button(st.button, 'Update user playlists', n_columns=5)
+        with st.expander("##### Select user playlists (default all)"):
+            # let the user uncheck playlists
+            st.markdown("Check boxes to select playlists from the selected users."
+                        "Note: to check all, first uncheck all (bug).")
+            columns = st.columns(np.ones(5))
+            with columns[1]:
+                check_all_playlists = st.button('Check all')
+            with columns[3]:
+                uncheck_all_playlists = st.button('Uncheck all')
+
+            if 'checkboxes' not in st.session_state.keys():
+                st.session_state['checkboxes_playlists'] = [True] * n_playlists
+
+            empty_checkboxes = wall_of_checkboxes(all_names, max_width=5)
+            if check_all_playlists:
+                st.session_state['checkboxes_playlists'] = [True] * n_playlists
+            if uncheck_all_playlists:
+                st.session_state['checkboxes_playlists'] = [False] * n_playlists
+            for i_emc, emc in enumerate(empty_checkboxes):
+                st.session_state['checkboxes_playlists'][i_emc] = emc.checkbox(all_names[i_emc], value=st.session_state['checkboxes_playlists'][i_emc])
+
+            filter_playlist = centered_button(st.button, 'Update user playlists', n_columns=5)
         if filter_playlist:
             return list(np.array(all_uris)[np.where(st.session_state['checkboxes_playlists'])])
         else:
@@ -187,31 +191,41 @@ def select_songs(legit_genres):
 
 def customize_widgets(genres_labels):
     st.subheader("Step 3: Customize it!")
-    st.markdown("##### Which genres?")
-    st.markdown("Check boxes to select genres, see how many tracks were selected below. Note: to check all, first uncheck all (bug).")
-    columns = st.columns(np.ones(5))
-    with columns[1]:
-        check_all = st.button('Check all')
-    with columns[3]:
-        uncheck_all = st.button('Uncheck all')
-
-    if 'checkboxes' not in st.session_state.keys():
-        st.session_state['checkboxes'] = [True] * len(genres_labels)
-
-    empty_checkboxes = wall_of_checkboxes(genres_labels, max_width=5)
-    if check_all:
-        st.session_state['checkboxes'] = [True] * len(genres_labels)
-    if uncheck_all:
-        st.session_state['checkboxes'] = [False] * len(genres_labels)
-    for i_emc, emc in enumerate(empty_checkboxes):
-        st.session_state['checkboxes'][i_emc] = emc.checkbox(genres_labels[i_emc], value=st.session_state['checkboxes'][i_emc])
+    st.markdown('##### Which genres?')
+
+    expanded = True if 'expanded_genres' in st.session_state else False
+    with st.expander("Unroll to select (default all)", expanded=expanded):
+        st.session_state['expanded_genres'] = True
+        st.markdown("Check boxes to select genres, see how many tracks were selected below. Note: to check all, first uncheck all (bug).")
+        columns = st.columns(np.ones(5))
+        with columns[1]:
+            check_all = st.button('Check all')
+        with columns[3]:
+            uncheck_all = st.button('Uncheck all')
+
+        if 'checkboxes' not in st.session_state.keys():
+            st.session_state['checkboxes'] = [True] * len(genres_labels)
+
+        empty_checkboxes = wall_of_checkboxes(genres_labels, max_width=5)
+        if check_all:
+            st.session_state['checkboxes'] = [True] * len(genres_labels)
+        if uncheck_all:
+            st.session_state['checkboxes'] = [False] * len(genres_labels)
+        for i_emc, emc in enumerate(empty_checkboxes):
+            st.session_state['checkboxes'][i_emc] = emc.checkbox(genres_labels[i_emc], value=st.session_state['checkboxes'][i_emc])
 
     st.markdown("##### What's the mood?")
-    valence = st.slider('Valence (0 negative, 100 positive)', min_value=0, max_value=100, value=100, step=1) / 100
-    energy = st.slider('Energy (0 low, 100 high)', min_value=0, max_value=100, value=100, step=1) / 100
-    danceability = st.slider('Danceability (0 low, 100 high)', min_value=0, max_value=100, value=100, step=1) / 100
+    valence = st.slider('Valence (0 negative, 100 positive)', min_value=0, max_value=100, value=60, step=1) / 100
+    energy = st.slider('Energy (0 low, 100 high)', min_value=0, max_value=100, value=60, step=1) / 100
+    danceability = st.slider('Danceability (0 low, 100 high)', min_value=0, max_value=100, value=60, step=1) / 100
     target_mood = np.array([valence, energy, danceability]).reshape(1, 3)
-    return target_mood
+    streamlit.markdown('##### Shall we explore?')
+    streamlit.write("Set the strength of music exploration:\n"
+                    "* 0%: all songs are selected from the music sources\n"
+                    "* 100%: all songs are new.")
+    exploration = st.slider('Exploration (0%, 100%)', min_value=0, max_value=100, value=50, step=1) / 100
+
+    return target_mood, exploration
 
 @st.cache
 def filter_songs_by_genre(checkboxes, genres_labels, genres):
@@ -223,6 +237,7 @@ def filter_songs_by_genre(checkboxes, genres_labels, genres):
     genre_selected_indexes = np.array(sorted(set(genre_selected_indexes)))
     return genre_selected_indexes
 
+@st.cache
 def find_best_songs_for_mood(all_tracks_audio_features, genre_selected_indexes, target_mood):
     candidate_moods = np.array([np.array(all_tracks_audio_features[feature])[genre_selected_indexes] for feature in ['valence', 'energy', 'danceability']]).T
     distances = np.sqrt(((candidate_moods - target_mood) ** 2).sum(axis=1))
@@ -230,6 +245,72 @@ def find_best_songs_for_mood(all_tracks_audio_features, genre_selected_indexes,
     n_candidates = distances.shape[0]
     return min_dist_indexes, n_candidates
 
+@st.cache
+def run_exploration(selected_tracks_uris, playlist_length, exploration, all_tracks_uris, target_mood, reauthenticate):
+    # sample exploration songs
+    if exploration > 0:
+        n_known = int(playlist_length * (1 - exploration))
+        n_new = playlist_length - n_known
+        print(f'Number of new songs: {n_new}, known songs: {n_known}')
+        known_songs = selected_tracks_uris[:n_known]
+        seed_songs = selected_tracks_uris[-n_new:]
+        dict_args = dict()  # enforce bounds on recommendations' moods
+        for i_m, m in enumerate(['valence', 'energy', 'danceability']):
+            dict_args[f'min_{m}'] = max(0, target_mood[i_m] - 0.1)
+            dict_args[f'max_{m}'] = min(1, target_mood[i_m] + 0.1)
+        dict_args_loose = dict()  # enforce bounds on recommendations' moods
+        for i_m, m in enumerate(['valence', 'energy', 'danceability']):
+            dict_args_loose[f'min_{m}'] = max(0, target_mood[i_m] - 0.2)
+            dict_args_loose[f'max_{m}'] = min(1, target_mood[i_m] + 0.2)
+        dict_args_looser = dict()  # enforce bounds on recommendations' moods
+        for i_m, m in enumerate(['valence', 'energy', 'danceability']):
+            dict_args_loose[f'min_{m}'] = max(0, target_mood[i_m] - 0.3)
+            dict_args_loose[f'max_{m}'] = min(1, target_mood[i_m] + 0.3)
+        new_songs = []
+        counter_seed = 0
+        counter_failure = 0
+        while len(new_songs) < n_new:
+            try:
+                print(seed_songs[counter_seed])
+                print(dict_args)
+                reco = sp.recommendations(seed_tracks=[seed_songs[counter_seed]], market="from_token", **dict_args)['tracks']
+                if len(reco) == 0:
+                    print('Using loose bounds')
+                    reco = sp.recommendations(seed_tracks=[seed_songs[counter_seed]], market="from_token", **dict_args_loose)['tracks']
+                    if len(reco) == 0:
+                        print('Using looser bounds')
+                        reco = sp.recommendations(seed_tracks=[seed_songs[counter_seed]], market="from_token", **dict_args_looser)['tracks']
+                        if len(reco) == 0:
+                            print('Removing bounds')
+                            reco = sp.recommendations(seed_tracks=[seed_songs[counter_seed]], market="from_token")['tracks']
+                assert len(reco) > 0
+                for r in reco:
+                    if r['uri'] not in all_tracks_uris:
+                        new_songs.append(r['uri'])
+                        break
+            except:
+                pass
+            print(counter_seed, len(new_songs))
+            counter_seed = (counter_seed + 1) % len(seed_songs)
+
+        assert len(new_songs) == n_new
+        assert len(known_songs) == n_known
+        selected_tracks_uris = np.array(list(known_songs) + new_songs)
+        np.random.shuffle(selected_tracks_uris)
+    return selected_tracks_uris
+
+@st.cache
+def sample_playlist(n_candidates, playlist_length, genre_selected_indexes, min_dist_indexes, all_tracks_uris):
+    # give more freedom to randomize the playlist
+    if n_candidates > 5 * playlist_length:
+        selected_tracks_indexes = genre_selected_indexes[min_dist_indexes[:int(playlist_length * 2)]]
+
+    else:
+        selected_tracks_indexes = genre_selected_indexes[min_dist_indexes[:playlist_length]]
+    selected_tracks_uris = all_tracks_uris[selected_tracks_indexes]
+    np.random.shuffle(selected_tracks_uris)
+    selected_tracks_uris = selected_tracks_uris[:playlist_length]
+    return selected_tracks_uris
 
 def run_app():
     global sp
@@ -243,13 +324,19 @@ def run_app():
                 "* **You're in control**: you provide a source of candidate songs, select a list of genres and choose the mood for the playlist.")
     fake = centered_button(st.button, "Let's go", n_columns=7, disabled=True)
 
-    sp, user_id, legit_genres = log_to_spotify()
+    sp, user_id, legit_genres, auth_manager = log_to_spotify()
+
+    def reauthenticate():
+        global sp
+        auth_manager.get_access_token(st.experimental_get_query_params()['code'])
+        sp = spotipy.Spotify(auth_manager=auth_manager)
+        return sp
 
     if 'login' in st.session_state or debug:
         all_tracks_uris, all_tracks_audio_features, genres, genres_labels = select_songs(legit_genres)
 
         if all_tracks_uris is not None:
-            target_mood = customize_widgets(genres_labels)
+            target_mood, exploration = customize_widgets(genres_labels)
             custom_button = centered_button(st.button, 'Run customization', n_columns=5)
             if custom_button or 'run_custom' in st.session_state.keys():
                 st.session_state['run_custom'] = True
@@ -265,27 +352,29 @@ def run_app():
                 init_time = time.time()
                 if genre_selected_indexes is not None:
                     min_dist_indexes, n_candidates = find_best_songs_for_mood(all_tracks_audio_features, genre_selected_indexes, target_mood)
-
                     print(f'7. filter by mood: {time.time() - init_time:.2f}')
                     init_time = time.time()
+
                     if n_candidates < 25:
                         st.warning('Please add more music sources or select more genres.')
                     else:
                         playlist_length = st.number_input(f'Pick a playlist length, given {n_candidates} candidates.', min_value=5,
-                                                          value=min(10, n_candidates//5), max_value=n_candidates//5)
+                                                          value=min(10, n_candidates//3), max_value=n_candidates//3)
 
-                        selected_tracks_indexes = genre_selected_indexes[min_dist_indexes[:playlist_length]]
-                        selected_tracks_uris = all_tracks_uris[selected_tracks_indexes]
-                        np.random.shuffle(selected_tracks_uris)
+                        selected_tracks_uris = sample_playlist(n_candidates, playlist_length, genre_selected_indexes, min_dist_indexes, all_tracks_uris)
+                        print(f'8. Sample songs: {time.time() - init_time:.2f}')
+                        init_time = time.time()
 
                         playlist_name = st.text_input('Playlist name', value='Mood Playlist')
                         if playlist_name == '':
                             st.warning('Please enter a playlist name.')
                         else:
-                            print(f'8. build playlist: {time.time() - init_time:.2f}')
-                            init_time = time.time()
                             generation_button = centered_button(st.button, 'Generate playlist', n_columns=5)
                             if generation_button:
+                                selected_tracks_uris = run_exploration(selected_tracks_uris, playlist_length, exploration, all_tracks_uris, target_mood.flatten(), reauthenticate)
+                                print(f'9. run exploration: {time.time() - init_time:.2f}')
+                                init_time = time.time()
+
                                 target_mood = np.array(target_mood).flatten() * 100
                                 description = f'Emotion Playlist for Valence: {int(target_mood[0])}, ' \
                                               f'Energy: {int(target_mood[1])}, ' \