from src.fw_api import current_instance, get_radios, post_radio_session, get_track_radio, list_libraries, favorite_track, get_audio_file from src.fw_libraries import libraries from src.fw_tags import list_tags from src.utils import download_track from src.mpv_control import player, track_url_to_uuid, player_fw_storage from src.settings import get_config from pyfzf.pyfzf import FzfPrompt from loguru import logger import threading import time fzf = FzfPrompt() @logger.catch def list_radios(): radios = get_radios() count = radios.get('count') results = radios.get('results') view = [] for i in results: index = results.index(i) id_radio = i.get('id') name = i.get('name') descr = i.get('description') radio_option = name if descr and descr != "": radio_option += f' | {descr}' view.append(f'{index}.{radio_option}') # Radios avalaible only for auth user if current_instance.s.headers.get('Authorization'): view.append('Favourites') view.append('Less listened') view.extend(['Tag', 'Random', 'Libraries', 'Users', 'Recently Added']) selected = fzf.prompt( view, f'--header \'Found {count} radios\' --read0', delimiter="\0")[0].split('.', 1) if 'Favourites' in selected: radio_load(id_radio, 'favorites', name='your favorites tracks') elif 'Tag' in selected: tag = list_tags() radio_load(type_radio='tag', name=f'by tag: {tag}', related_object=tag) elif 'Random' in selected: radio_load(id_radio, 'random', name='totally random') elif 'Libraries' in selected: id_radio, type_radio, name_radio, related_obj = libraries() radio_load(id_radio, type_radio, name_radio, related_obj) elif 'Users' in selected: libs = list_libraries()['results'] libraries_listing = [] for lib_i in libs: lib_name = lib_i.get('actor').get('full_username') if lib_name not in libraries_listing: libraries_listing.append(lib_name) libraries_listing.append('Custom') lib_addr = fzf.prompt(libraries_listing)[0] if lib_addr == 'Custom': print('Input remote user library (ex. nick@funkwhale.domain.example: ') lib_addr = input() radio_load(None, 'actor-content', lib_addr, lib_addr) elif 'Recently Added' in selected: radio_load(id_radio, 'recently-added', name='Newest content on the network') elif 'Less listened' in selected: radio_load(id_radio, 'less-listened', name="Less listened tracks") else: id_selected = selected[0] id_radio = results[int(id_selected)].get('id') name_radio = results[int(id_selected)].get('name') radio_load(id_radio, name=name_radio) def radio_generator(radio_session_id): count_t = 0 while radio_session_id != '': time.sleep(1) if not radio_event_gen.wait(0): break count_t += 1 if count_t >= 60: count_t = 0 playlist_remaining = len(player.playlist) - \ player.playlist_current_pos if playlist_remaining <= 2: radio_get_track(radio_session_id) logger.info('Radio generator stopped') radio_event_gen = threading.Event() @logger.catch def radio_load(id_radio=None, type_radio='custom', name=None, related_object=None): show_like_button = get_config('show_like_button') player.volume = get_config('mpv_volume') requested_radio = { 'custom_radio': id_radio, 'radio_type': type_radio, 'related_object_id': related_object } radio_session_id = post_radio_session(requested_radio).get('id') for i in range(0, 2): radio_get_track(radio_session_id) radio_event_gen.set() radio_task = threading.Thread( target=radio_generator, args=(radio_session_id,), daemon=True) radio_task.start() player_items_menu = ['Next', 'Prev', 'Pause', 'Download', 'Info'] if show_like_button: player_items_menu.append('Like') player_items_menu.extend(['Hide artist', 'Exit']) while True: try: if player.pause: player_items_menu[2] = 'Play' else: player_items_menu[2] = 'Pause' select = fzf.prompt(player_items_menu, f"--header=\'Radio {name} playing...\'")[0] if select == 'Next': playlist_remaining = len( player.playlist) - player.playlist_current_pos if playlist_remaining <= 2: threading.Thread(target=radio_get_track, args=( radio_session_id,), daemon=True).start() player.playlist_next() elif select == 'Prev': player.playlist_prev() elif select in ('Pause', 'Play'): if player.pause: player.pause = False else: player.pause = True elif select == 'Download': name_downloaded = download_track(player.stream_open_filename) elif select == 'Info': track = player_fw_storage.storage.get(track_url_to_uuid()) for i in ('title', 'fid', 'license', 'album', 'artist'): if i in ('album', 'artist'): name_aa = track.get(i).get('name') if not name_aa: name_aa = track[i]['title'] print(i + ': ' + name_aa) key = track.get(i) if key and isinstance(key, str): print(i + ': ' + key) print('Direct link: ' + player.stream_open_filename) input() elif select == 'Like': favorite_track(player_fw_storage.storage.get( track_url_to_uuid())['id']) elif select == 'Exit': try: radio_event_gen.clear() except: logger.exception('Error stopping Thread radio generator') pass player.playlist_clear() player.stop() player_fw_storage.storage = {} break except: try: radio_event_gen.clear() except: logger.exception('Error stopping Thread radio generator') pass player.playlist_clear() player.stop() player_fw_storage.storage = {} logger.exception('Radio force stopped') break def radio_get_track(radio_session_id): radio_context = get_track_radio({'session': radio_session_id}) if not radio_context: return if isinstance(radio_context, str): logger.error(radio_context) if radio_context == "Radio doesn't have more candidates": radio_event_gen.clear() return if radio_context.get('error'): logger.error(radio_context.get('error')) return else: track = radio_context.get('track') listen_url = track['listen_url'] player_fw_storage.storage[track_url_to_uuid(listen_url)] = track player.loadfile(get_audio_file( listen_url, listen_url=True), 'append-play')