from huggingface_hub import hf_hub_download import os import time import gradio as gr import yt_dlp import sys import uuid import traceback import tensorflow import csv import json embed_html1 = '' # NO GPU ? os.environ['CUDA_VISIBLE_DEVICES'] = '-1' os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3" # tensorflow.config.threading.set_intra_op_parallelism_threads(8) # tensorflow.config.threading.set_inter_op_parallelism_threads(8) #print(f"Available GPUs {tensorflow.config.list_physical_devices('GPU')}") # python_path = hf_hub_download( repo_id=os.environ["REPO_ID"], repo_type="space", filename=os.environ["MODEL_FILE"], use_auth_token=os.environ["TOKEN"], ) # print("PATH : ", python_path) sys.path.append(os.environ["PRIVATE_DIR"]) from models import * max_results = 100 max_output = 50 # global (faster) ind = get_index() ind_filenames = get_audio_names() catalog = get_catalog() url_dict = get_durl_myma() dict_catalog = get_dict_catalog() fixation_id_to_file_name = {} for file_name, infos in dict_catalog.items(): # we want only main versions if infos["Parent fixation id"].strip(): continue fixation_id_to_file_name[infos["Fixation id"].strip()] = file_name child_to_parent_filename = {} count = count_failed = 0 for file_name, infos in dict_catalog.items(): if not infos["Parent fixation id"].strip(): continue count += 1 try: child_to_parent_filename[file_name] = fixation_id_to_file_name[ infos["Parent fixation id"].strip() ] except Exception as e: print(f"No parent for {file_name} : {e}") count_failed += 1 print(f"{count_failed} tracks have no parent / {count} tracks") parent_file_names = set(list(fixation_id_to_file_name.values())) file_name_to_url = {} for file_url in url_dict.values(): file_name = os.path.splitext(os.path.basename(file_url))[0] if file_name not in parent_file_names: continue file_name_to_url[file_name] = file_url parent_file_names = [] fixation_id_to_file_name = [] with open("fixation_id_to_file_name.json", "w") as w, open( "child_to_parent_filename.json", "w" ) as w2, open("file_name_to_url.json", "w") as w3: json.dump(fixation_id_to_file_name, w) json.dump(child_to_parent_filename, w2) json.dump(file_name_to_url, w3) parent_file_names = [] fixation_id_to_file_name = {} def download_audio(id_video): id = id_video.split("?v=")[-1][:11] file_name = f"youtube_video_{id}.mp3" with yt_dlp.YoutubeDL( {"extract_audio": True, "format": "bestaudio", "outtmpl": file_name} ) as video: video.download(id_video) embed_html_all = embed_html1 + id + embed_html2 return file_name, embed_html_all def process_url(input_path): # setup the client # try : file_name, embed_html_all = download_audio(input_path) return process(file_name, embed_html_all) # except: # return "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "" def process_file(input_path): return process(input_path, "")[1:] def process(file_name, embed_html_all): """try :""" emb, _ = get_embed(file_name) try: os.remove(file_name) except: print(traceback.format_exc()) # => global timestart = time.time() D, I = do_search(emb, ind) print("search time :", time.time() - timestart) print("DEBUG DISTANCES : ", D) tops = get_topN(I, ind_filenames, url_dict, catalog, max_results) formated = [{"f": "Choose a result to play", "t": ""}] output_csv = f"{file_name}_results.csv" with open(output_csv, "w") as w: writer = csv.writer(w) header = False already = set() for position, top in enumerate(tops): if len(formated) / 2 >= max_output: break file = os.path.splitext(os.path.basename(top))[0] try: file = child_to_parent_filename[file] top = file_name_to_url[file] except KeyError: pass if file in already: continue already.add(file) file_name = file if file in dict_catalog: if not header: writer.writerow(list(dict_catalog[file].keys())) header = True writer.writerow(dict_catalog[file].values()) file_name = dict_catalog[file]["Track name"] try: file_name += " - " + dict_catalog[file]["Composer1 full name"] except: pass try: file_name += " - " + dict_catalog[file]["Album name"] except: pass else: writer.writerow([file, "no metadata provided"]) formated.append({"f": f"{position+1} - {file_name}", "t": top}) return (embed_html_all, output_csv, formated) """except: return embed_html_all, "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "", "Erreur Input", "" """ with gr.Blocks() as demo: with gr.Row(): with gr.Column(): with gr.Row(): # gr.HTML(embed_html) html = gr.HTML() with gr.Row(): with gr.Column(): audio_url_input = gr.Textbox( placeholder="YouTube video URL", label="YouTube video URL" ) analyze_url_btn = gr.Button("Search from URL") with gr.Row(): with gr.Column(): audio_input_file = gr.Audio(type="filepath", label="Audio Input") analyze_file_btn = gr.Button("Search from file") with gr.Row(): with gr.Column(): csv_results = gr.File(label="Results as CSV") with gr.Row(): with gr.Column(): results = gr.JSON(visible=False) select_results = gr.Dropdown(label="Results", choices=[]) audio_player = gr.Audio(None, label="Results player") @select_results.select(inputs=select_results, outputs=audio_player) def change_audio(value): if value: return gr.Audio(value, label="Results player") return gr.Audio(None, label="Results player") @results.change( inputs=results, outputs=select_results, ) def update_select(json_results): try: print("change dropdown") return gr.Dropdown( label="Results", choices=[(k["f"], k["t"]) for k in json_results], value=None, ) except: return gr.Dropdown( choices=[], label="Results", ) @audio_input_file.change( outputs=[ audio_url_input, results, select_results, csv_results, audio_player, html, ] ) def cleanup_on_file(): print("cleanup on file change") return ( gr.Textbox( placeholder="YouTube video URL", label="YouTube video URL" ), gr.JSON([{"f": "Choose a result to play", "t": ""}], visible=False), gr.Dropdown(label="Results", choices=[]), gr.File(None, label="Results as CSV"), gr.Audio(None, label="Results player"), gr.HTML(""), ) @audio_url_input.change( outputs=[ audio_input_file, results, select_results, csv_results, audio_player, html, ] ) def cleanup_on_url(): print("cleanup on url change") return ( gr.Audio(None, type="filepath", label="Audio Input"), gr.JSON([{"f": "Choose a result to play", "t": ""}], visible=False), gr.Dropdown(choices=[], label="Results"), gr.File(None, label="Results as CSV"), gr.Audio(None, label="Results player"), gr.HTML(""), ) analyze_url_btn.click( process_url, inputs=[audio_url_input], outputs=[html, csv_results, results], ) gr.Examples( examples=[ "https://www.youtube.com/watch?v=aNzCDt2eidg", "https://www.youtube.com/watch?v=NBE-uBgtINg", "https://www.youtube.com/watch?v=5NV6Rdv1a3I", "https://www.youtube.com/watch?v=OiC1rgCPmUQ", # "https://www.youtube.com/watch?v=dRX0wDNK6S4", # "https://www.youtube.com/watch?v=Guzu9aAeDIc", ], inputs=[audio_url_input], outputs=[html, csv_results, results], fn=process_url, cache_examples=False, examples_per_page=20, run_on_click=True, ) analyze_file_btn.click( process_file, inputs=[audio_input_file], outputs=[csv_results, results], ) demo.launch(debug=False)