diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a96caa..5599d1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,14 @@ +## 4.1.0 + +1. Significant improvements on app loading speed :zap: +2. Recording quality improved 🎶 +3. Faster station lookup +4. Optimize interaction with external player, improving play/pause time +5. Improved fuzzy find logic 🔍 +6. Stability improvements for windows +7. Critical bug fixes 🐛 +8. Minor UI fixes + ## 4.0.2 1. More information on the shazam identified track diff --git a/radioactive/__main__.py b/radioactive/__main__.py index 81fa8dc..782da50 100755 --- a/radioactive/__main__.py +++ b/radioactive/__main__.py @@ -3,44 +3,11 @@ import os import signal import sys +import threading from time import sleep -import psutil -from zenlog import log - -from radioactive.alias import Alias -from radioactive.app import App -from radioactive.ffplay import Ffplay, kill_background_ffplays -from radioactive.handler import Handler -from radioactive.help import show_help -from radioactive.history import History -from radioactive.last_station import Last_station -from radioactive.parser import parse_options -from radioactive.paths import get_pid_path -from radioactive.utilities import ( - check_sort_by_parameter, - handle_add_station, - handle_add_to_favorite, - handle_current_play_panel, - handle_direct_play, - handle_favorite_table, - handle_history_table, - handle_listen_keypress, - handle_play_last_station, - handle_play_random_station, - handle_record, - handle_save_last_station, - handle_save_to_history, - handle_search_stations, - handle_station_selection_menu, - handle_station_uuid_play, - handle_update_screen, - handle_user_choice_from_search_result, - handle_welcome_screen, -) - -# globally needed as signal handler needs it -# to terminate main() properly +# Globally needed as signal handler needs them. +# These are assigned inside main() or final_step(). ffplay = None player = None @@ -49,7 +16,18 @@ def final_step(options, last_station, alias, handler, history, station_list=None global ffplay # always needed global player - # check target URL + from zenlog import log + + from radioactive.ffplay import Ffplay + from radioactive.utilities import ( + handle_add_to_favorite, + handle_current_play_panel, + handle_listen_keypress, + handle_record, + handle_save_last_station, + handle_save_to_history, + ) + target_url = (options.get("target_url") or "").strip() if target_url == "": log.info("Type 's' to search for a station or '?' for help") @@ -123,28 +101,62 @@ def final_step(options, last_station, alias, handler, history, station_list=None def main(): + from zenlog import log + log.level("info") - app = App() + from radioactive.app import App + from radioactive.parser import parse_options + app = App() options = parse_options() - VERSION = app.get_version() + # --- Fast early exits: avoid all heavy imports --- if options["version"]: log.info("RADIO-ACTIVE : version {}".format(VERSION)) sys.exit(0) - handler = Handler() - alias = Alias() - alias.generate_map() - last_station = Last_station() - history = History() + from radioactive.help import show_help if options["show_help_table"]: show_help() sys.exit(0) + # --- Deferred heavy imports (saves ~260ms when not needed) --- + import psutil + + from radioactive.alias import Alias + from radioactive.ffplay import kill_background_ffplays + from radioactive.handler import Handler + from radioactive.history import History + from radioactive.last_station import Last_station + from radioactive.paths import get_pid_path + from radioactive.utilities import ( + check_sort_by_parameter, + handle_add_station, + handle_add_to_favorite, + handle_current_play_panel, + handle_direct_play, + handle_favorite_table, + handle_history_table, + handle_play_last_station, + handle_play_random_station, + handle_record, + handle_search_stations, + handle_station_selection_menu, + handle_station_uuid_play, + handle_update_modal, + handle_update_screen, + handle_user_choice_from_search_result, + handle_welcome_screen, + ) + + alias = Alias() + alias.generate_map() + last_station = Last_station() + history = History() + if options["flush_fav_list"]: sys.exit(alias.flush()) @@ -226,6 +238,11 @@ def cleanup(): handle_welcome_screen() + # Run update check in background so it never blocks the interactive prompt. + # The modal will be shown if an update is detected. + _update_thread = threading.Thread(target=app.is_update_available, daemon=True) + _update_thread.start() + # ------------------ SCHEDULED RECORDING MODE ------------------ # from radioactive.feature_flags import RECORDING_FEATURE @@ -314,6 +331,7 @@ def cleanup(): log.info(f"Target URL: {options['target_url']}") else: # We need to use handler to validate UUID + handler = Handler() options["curr_station_name"], options["target_url"] = ( handle_station_uuid_play(handler, options["search_station_uuid"]) ) @@ -400,7 +418,15 @@ def cleanup(): options["sort_by"] = check_sort_by_parameter(options["sort_by"]) - handle_update_screen(app) + # Construct Handler as late as possible — right before we actually need the API. + # This avoids the ~50ms init cost for flag-only paths (--list, --kill, etc.) + handler = Handler() + + # Update check is already running in the background thread started above; + # wait briefly (non-blocking) so it can print before we proceed to prompts + _update_thread.join(timeout=0.4) + if app.update_available: + handle_update_modal(app) # ----------- country ----------- # if options["discover_country_code"]: @@ -560,6 +586,8 @@ def cleanup(): def signal_handler(sig, frame): + from zenlog import log + log.debug("You pressed Ctrl+C!") log.debug("Stopping the radio") if ffplay and ffplay.is_playing: diff --git a/radioactive/actions.py b/radioactive/actions.py index be6231e..d6e390c 100644 --- a/radioactive/actions.py +++ b/radioactive/actions.py @@ -188,7 +188,12 @@ def handle_record( outfile_path = os.path.join(record_file_path, tmp_filename) process = record_audio_from_url( - target_url, outfile_path, force_mp3, loglevel, duration + target_url, + outfile_path, + force_mp3, + loglevel, + duration, + station_name=curr_station_name, ) return process, outfile_path @@ -205,10 +210,9 @@ def handle_add_station(alias) -> None: if left.strip() == "" or right.strip() == "": log.error("Empty inputs not allowed") - sys.exit(1) + return alias.add_entry(left, right) log.info("New entry: {}={} added\n".format(left, right)) - sys.exit(0) def handle_add_to_favorite(alias, station_name: str, station_uuid_url: str) -> None: @@ -337,37 +341,14 @@ def handle_direct_play( return station_name, station_name_or_url else: log.debug("Direct play: station name provided") - # station name from fav list - # search for the station in fav list and return name and url - + # search in favorites first response = alias.search(station_name_or_url) + + # if not found, check history if not response and history: log.debug("Not found in favorites, checking history") - # history object should have a search method or we iterate - # looking at history.py might be good if hasattr(history, "search"): response = history.search(station_name_or_url) - else: - # fallback iteration - for entry in history.get_list(): - name = entry.get("name", "").strip() - val = entry.get("uuid_or_url", "").strip() - # also check stationuuid if it exists (older history) - sid = entry.get("stationuuid", "").strip() - - token = station_name_or_url.strip().lower() - log.debug( - f"Comparing history entry: '{name.lower()}' with token: '{token}'" - ) - - if ( - name.lower() == token - or val == station_name_or_url.strip() - or sid == station_name_or_url.strip() - ): - log.debug(f"History match found: {name}") - response = entry - break if not response: log.debug(f"Search failed for: {station_name_or_url}") @@ -376,7 +357,7 @@ def handle_direct_play( ) return None, None else: - log.debug(f"Direct play: {response}") + log.debug(f"Direct play found: {response}") return response["name"], response.get("uuid_or_url") or response.get( "stationuuid" ) diff --git a/radioactive/alias.py b/radioactive/alias.py index 557fe38..58b6238 100644 --- a/radioactive/alias.py +++ b/radioactive/alias.py @@ -9,6 +9,7 @@ def __init__(self): from radioactive.paths import get_alias_path self.alias_map = [] + self.alias_dict = {} # for O(1) lookups self.found = False self.alias_path = get_alias_path() @@ -33,21 +34,22 @@ def generate_map(self): if os.path.exists(self.alias_path): log.debug(f"Alias file at: {self.alias_path}") try: - with open(self.alias_path, "r+") as f: + with open(self.alias_path, "r") as f: alias_data = f.read().strip() if alias_data == "": log.debug("Empty alias list") return alias_list = alias_data.splitlines() + self.alias_map = [] + self.alias_dict = {} for alias in alias_list: - if alias.strip() == "": - # empty line pass - continue - temp = alias.split("==") - left = temp[0] - right = temp[1] - # may contain both URL and UUID - self.alias_map.append({"name": left, "uuid_or_url": right}) + if "==" in alias: + temp = alias.split("==") + left = temp[0].strip() + right = temp[1].strip() + entry = {"name": left, "uuid_or_url": right} + self.alias_map.append(entry) + self.alias_dict[left.lower()] = entry except Exception as e: log.debug(f"could not get / parse alias data: {e}") @@ -55,40 +57,41 @@ def generate_map(self): log.debug("Alias file does not exist") def search(self, entry): - """searches for an entry in the fav list with the name - the right side may contain both url or uuid , need to check properly - """ - log.debug("Alias search: {}".format(entry)) - if len(self.alias_map) > 0: - log.debug("looking under alias file") - for alias in self.alias_map: - if alias["name"].strip().lower() == entry.strip().lower(): - log.debug( - "Alias found: {} == {}".format( - alias["name"], alias["uuid_or_url"] - ) - ) - self.found = True - return alias - - log.debug("Alias not found") - else: - log.debug("Empty Alias file") + """searches for an entry in the fav list with the name (fast O(1))""" + if not entry: + return None + + target = entry.strip().lower() + if target in self.alias_dict: + res = self.alias_dict[target] + log.debug(f"Alias found: {res['name']} == {res['uuid_or_url']}") + self.found = True + return res + + log.debug(f"Alias not found: {entry}") return None def add_entry(self, left, right): """Adds a new entry to the fav list""" + # Ensure map is current before adding self.generate_map() if self.search(left) is not None: log.warning("An entry with same name already exists, try another name") return False - else: - with open(self.alias_path, "a+") as f: + + try: + with open(self.alias_path, "a") as f: f.write("{}=={}\n".format(left.strip(), right.strip())) log.info("Current station added to your favorite list") - # Refresh map after writing - self.generate_map() + + # Update internal state directly to avoid re-reading the whole file + entry = {"name": left.strip(), "uuid_or_url": right.strip()} + self.alias_map.append(entry) + self.alias_dict[left.strip().lower()] = entry return True + except Exception as e: + log.error(f"Could not add entry: {e}") + return False def flush(self): """deletes all the entries in the fav list""" diff --git a/radioactive/app.py b/radioactive/app.py index 98d2c37..007aff8 100644 --- a/radioactive/app.py +++ b/radioactive/app.py @@ -12,15 +12,18 @@ else: import importlib_metadata as metadata +# from zenlog import log + class App: def __init__(self): try: self.__VERSION__ = metadata.version("radio-active") except metadata.PackageNotFoundError: - self.__VERSION__ = "4.0.2" # change this on every update # + self.__VERSION__ = "4.1.0" # change this on every update # self.pypi_api = "https://pypi.org/pypi/radio-active/json" self.remote_version = "" + self.update_available = False def get_version(self): """get the version number as string""" @@ -47,11 +50,14 @@ def is_update_available(self): tup_remote = tuple(map(int, self.remote_version.split("."))) if tup_remote > tup_local: + self.update_available = True return True + self.update_available = False return False except Exception: - print("Could not fetch remote version number") + # log.debug("Could not fetch remote version number") + pass def get_release_notes(self, local_version, remote_version): """Fetch and parse release notes for all versions greater than local_version.""" diff --git a/radioactive/ffplay.py b/radioactive/ffplay.py index 8277a0c..c6248ca 100644 --- a/radioactive/ffplay.py +++ b/radioactive/ffplay.py @@ -18,27 +18,32 @@ def kill_background_ffplays() -> None: """ Kill all background 'ffplay' processes started by this user. + Optimized to use faster platform-specific tools when available. """ + try: + if sys.platform != "win32": + # Fast path for Unix-like systems + # -x ensures exact match for process name 'ffplay' + subprocess.run( + ["pkill", "-x", "ffplay"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + return + except Exception: + pass + + # Fallback to psutil if pkill fails or on Windows all_processes = psutil.process_iter(attrs=["pid", "name"]) - count = 0 - # Iterate through the processes and terminate those named "ffplay" for process in all_processes: try: if process.info["name"] == "ffplay": pid = process.info["pid"] p = psutil.Process(pid) p.terminate() - count += 1 - log.info(f"Terminated ffplay process with PID {pid}") - if p.is_running(): - p.kill() - log.debug(f"Forcefully killing ffplay process with PID {pid}") + log.debug(f"Terminated ffplay process with PID {pid}") except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): - # Handle exceptions, such as processes that no longer exist or access denied - log.debug("Could not terminate a ffplay processes!") - if count == 0: - pass - # log.info("No background radios are running!") + pass class Ffplay: @@ -90,8 +95,9 @@ def start_process(self) -> None: self.process = subprocess.Popen( ffplay_commands, shell=False, - stdout=subprocess.PIPE, + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + stdin=subprocess.PIPE, text=True, ) @@ -127,7 +133,6 @@ def _check_error_output(self) -> None: break except Exception: break - sleep(0.5) def _handle_error(self, stderr_result: str) -> None: """Log the error message.""" @@ -154,26 +159,11 @@ def terminate_parent_process(self) -> None: log.debug(f"Could not kill parent process: {e}") def is_active(self) -> bool: - """Check if the ffplay process is currently active/running.""" + """Check if the ffplay process is currently active/running (fast).""" if not self.process: - log.warning("Process is not initialized") return False - try: - proc = psutil.Process(self.process.pid) - if proc.status() == psutil.STATUS_ZOMBIE: - log.debug("Process is a zombie") - return False - - if proc.status() in [psutil.STATUS_RUNNING, psutil.STATUS_SLEEPING]: - return True - - log.warning("Process is not in an expected state") - return False - - except (psutil.NoSuchProcess, Exception) as e: - log.debug(f"Process not found or error checking status: {e}") - return False + return self.process.poll() is None def play(self) -> None: """Resume or start playback.""" @@ -187,10 +177,10 @@ def stop(self) -> None: try: self.process.terminate() try: - self.process.wait(timeout=3) + self.process.wait(timeout=0.2) except subprocess.TimeoutExpired: self.process.kill() - self.process.wait(timeout=2) + self.process.wait(timeout=0.2) log.debug("Radio playback stopped successfully") except Exception as e: diff --git a/radioactive/handler.py b/radioactive/handler.py index 1e4ca9d..479e509 100644 --- a/radioactive/handler.py +++ b/radioactive/handler.py @@ -5,10 +5,9 @@ import datetime import json import sys +import threading from typing import Any, Dict, List, Optional, Union -import requests_cache -from pyradios import RadioBrowser from rich.console import Console from rich.table import Table from zenlog import log @@ -147,31 +146,42 @@ def __init__(self): # When RadioBrowser can not be initiated properly due to no internet (probably) try: + import requests_cache + from pyradios import RadioBrowser + expire_after = datetime.timedelta(days=DEFAULT_CACHE_RETENTION_DAYS) session = requests_cache.CachedSession( cache_name="cache", backend="sqlite", expire_after=expire_after ) self.API = RadioBrowser(session=session) + self._country_map = None # Lazy cache for country names -> codes except Exception as e: log.debug(f"Error initializing RadioBrowser: {e}") log.critical("Something is wrong with your internet connection") sys.exit(1) + def _get_country_map(self) -> Dict[str, str]: + """Lazy load and cache country mapping.""" + if self._country_map is None: + try: + log.debug("Fetching country list for cache...") + countries = self.API.countries() + self._country_map = { + c["name"].lower(): c["iso_3166_1"] + for c in countries + if "name" in c and "iso_3166_1" in c + } + except Exception as e: + log.debug(f"Could not fetch country list: {e}") + return {} + return self._country_map + def get_country_code(self, name: str) -> Optional[str]: """ - Get the ISO 3166-1 alpha-2 country code for a given country name. - - Args: - name (str): The name of the country. - - Returns: - str: The country code if found, None otherwise. + Get the ISO 3166-1 alpha-2 country code for a given country name (cached). """ - self.countries = self.API.countries() - for country in self.countries: - if country["name"].lower() == name.lower(): - return country["iso_3166_1"] - return None + cmap = self._get_country_map() + return cmap.get(name.lower()) def validate_uuid_station(self) -> List[Dict[str, Any]]: """ @@ -185,8 +195,13 @@ def validate_uuid_station(self) -> List[Dict[str, Any]]: log.debug(json.dumps(self.response[0], indent=3)) self.target_station = self.response[0] - # register a valid click to increase its popularity - self.vote_for_uuid(self.target_station["stationuuid"]) + # register a valid click to increase its popularity (non-blocking) + t = threading.Thread( + target=self.vote_for_uuid, + args=(self.target_station["stationuuid"],), + daemon=True, + ) + t.start() return self.response @@ -229,7 +244,7 @@ def search_by_station_name( except Exception as e: log.debug(f"Error in search_by_station_name: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] # ------------------------- UUID ------------------------ # def play_by_station_uuid(self, uuid: str) -> List[Dict[str, Any]]: @@ -248,7 +263,7 @@ def play_by_station_uuid(self, uuid: str) -> List[Dict[str, Any]]: except Exception as e: log.debug(f"Error in play_by_station_uuid: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] # -------------------------- COUNTRY ----------------------# def discover_by_country( @@ -290,13 +305,13 @@ def discover_by_country( except Exception as e: log.debug(f"Error searching by country name: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] else: - log.error("Not a valid country name") - sys.exit(1) + log.error(f"'{country_code_or_name}' is not a valid country name") + return [] - # display the result - print_table( + # display and return result + return print_table( response, [ "Station:name@30", @@ -307,7 +322,6 @@ def discover_by_country( sort_by, filter_with, ) - return response # ------------------- by state --------------------- @@ -324,7 +338,7 @@ def discover_by_state( except Exception as e: log.debug(f"Error discover_by_state: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] return print_table( response, @@ -354,7 +368,7 @@ def discover_by_language( except Exception as e: log.debug(f"Error discover_by_language: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] return print_table( response, @@ -382,7 +396,7 @@ def discover_by_tag( except Exception as e: log.debug(f"Error discover_by_tag: {e}") log.error("Something went wrong. please try again.") - sys.exit(1) + return [] return print_table( response, diff --git a/radioactive/help.py b/radioactive/help.py index d2c554a..437d564 100644 --- a/radioactive/help.py +++ b/radioactive/help.py @@ -1,13 +1,13 @@ from os import path -from rich.console import Console -from rich.table import Table - user = path.expanduser("~") def show_help(): """Show help message as table""" + from rich.console import Console + from rich.table import Table + console = Console() table = Table(show_header=True, header_style="bold magenta") diff --git a/radioactive/history.py b/radioactive/history.py index e3faba0..ac7fe9c 100644 --- a/radioactive/history.py +++ b/radioactive/history.py @@ -1,5 +1,6 @@ import json import os +from typing import Dict, Optional from zenlog import log @@ -19,7 +20,7 @@ def load(self): self.history_list = json.load(f) else: self.history_list = [] - except Exception as e: + except (json.JSONDecodeError, Exception) as e: log.debug(f"Error loading history: {e}") self.history_list = [] @@ -28,22 +29,19 @@ def append(self, station): Add a station to history. station: dict with name, uuid, url keys mostly """ - # remove existing entry with same name or uuid to avoid duplicates - # and bring it to top + curr_name = station.get("name", "").strip().lower() + curr_uuid = ( + station.get("stationuuid") or station.get("uuid_or_url") or "" + ).strip() + + # Deduplicate and bring to top new_list = [] for s in self.history_list: prev_name = s.get("name", "").strip().lower() - curr_name = station.get("name", "").strip().lower() - prev_uuid = (s.get("stationuuid") or s.get("uuid_or_url") or "").strip() - curr_uuid = ( - station.get("stationuuid") or station.get("uuid_or_url") or "" - ).strip() - # check name if prev_name == curr_name: continue - # check uuid if available if prev_uuid and curr_uuid and prev_uuid == curr_uuid: continue @@ -59,11 +57,31 @@ def append(self, station): self.save() def save(self): + """Atomic save of history file to prevent corruption.""" try: - with open(self.history_path, "w") as f: + temp_path = f"{self.history_path}.tmp" + with open(temp_path, "w") as f: json.dump(self.history_list, f, indent=4) + os.replace(temp_path, self.history_path) except Exception as e: log.warning(f"Could not save history: {e}") + if os.path.exists(temp_path): + os.remove(temp_path) def get_list(self): return self.history_list + + def search(self, token: str) -> Optional[Dict]: + """Search for a station in history by name, url, or uuid.""" + if not token: + return None + token = token.strip().lower() + + for entry in self.history_list: + name = entry.get("name", "").strip().lower() + url = entry.get("uuid_or_url", "").strip().lower() + uuid = (entry.get("stationuuid") or "").strip().lower() + + if name == token or url == token or uuid == token: + return entry + return None diff --git a/radioactive/recorder.py b/radioactive/recorder.py index c94777c..64151ce 100644 --- a/radioactive/recorder.py +++ b/radioactive/recorder.py @@ -4,6 +4,19 @@ def record_audio_auto_codec(input_stream_url): + """Detect the audio codec of a stream and map it to a standard extension.""" + # Standard mapping of codec_name to file extension + CODEC_MAP = { + "aac": "aac", + "mp3": "mp3", + "opus": "opus", + "vorbis": "ogg", + "flac": "flac", + "wav": "wav", + "pcm_s16le": "wav", + "pcm_s24le": "wav", + } + try: # Run FFprobe to get the audio codec information ffprobe_command = [ @@ -19,55 +32,69 @@ def record_audio_auto_codec(input_stream_url): input_stream_url, ] - codec_info = subprocess.check_output(ffprobe_command, text=True) + # 5s is usually enough for metadata headers + codec_info = subprocess.check_output(ffprobe_command, text=True, timeout=5) - # Determine the file extension based on the audio codec - audio_codec = codec_info.strip() - audio_codec = audio_codec.split("\n")[0] - return audio_codec + audio_codec = codec_info.strip().split("\n")[0].lower() + # Return mapped extension or the codec name as fallback + return CODEC_MAP.get(audio_codec, audio_codec) - except subprocess.CalledProcessError as e: - log.error(f"Error: could not fetch codec {e}") + except FileNotFoundError: + log.error("ffprobe not found! Install FFmpeg to use recording.") + return None + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + log.debug(f"Could not fetch codec via ffprobe: {e}") return None -def record_audio_from_url(input_url, output_file, force_mp3, loglevel, duration=None): +def record_audio_from_url( + input_url, + output_file, + force_mp3, + loglevel, + duration=None, + station_name=None, + track_name=None, +): """ Record audio from a URL using FFmpeg. - Returns the subprocess.Popen object to allow UI tracking. """ log.debug(f"Recording audio from {input_url} to {output_file}") try: ffmpeg_command = [ "ffmpeg", + "-y", # Overwrite if exists "-i", input_url, "-vn", "-stats", ] - ffmpeg_command.append("-c:a") if force_mp3: - ffmpeg_command.append("libmp3lame") + ffmpeg_command.extend(["-c:a", "libmp3lame", "-q:a", "2"]) else: - ffmpeg_command.append("copy") + ffmpeg_command.extend(["-c:a", "copy"]) + + # Add metadata if provided + if station_name: + ffmpeg_command.extend(["-metadata", f"service_name={station_name}"]) + ffmpeg_command.extend(["-metadata", f"publisher={station_name}"]) + if track_name: + ffmpeg_command.extend(["-metadata", f"title={track_name}"]) ffmpeg_command.append("-loglevel") if loglevel == "debug": ffmpeg_command.append("info") else: - ffmpeg_command.append("error") - ffmpeg_command.append("-hide_banner") + ffmpeg_command.extend(["error", "-hide_banner"]) if duration: seconds = int(duration) * 60 - ffmpeg_command.append("-t") - ffmpeg_command.append(str(seconds)) + ffmpeg_command.extend(["-t", str(seconds)]) ffmpeg_command.append(output_file) - # Run FFmpeg command in background to allow UI tracking - # Use DEVNULL to prevent hangs and terminal corruption + # Use stdin=PIPE to allow sending 'q' for graceful termination process = subprocess.Popen( ffmpeg_command, stdin=subprocess.PIPE, @@ -77,12 +104,9 @@ def record_audio_from_url(input_url, output_file, force_mp3, loglevel, duration= return process except FileNotFoundError: - log.error( - "FFmpeg not found! Please install FFmpeg to use the recording feature." - ) + log.error("FFmpeg not found! Please install it to use recording.") + return None except Exception as ex: + log.debug(f"FFmpeg startup error: {ex}") log.error(f"Error while starting recording: {ex}") return None - except Exception as ex: - log.debug("Error: {}".format(ex)) - log.error(f"An error occurred: {ex}") diff --git a/radioactive/ui.py b/radioactive/ui.py index c45313c..13fb0af 100644 --- a/radioactive/ui.py +++ b/radioactive/ui.py @@ -35,6 +35,7 @@ def handle_welcome_screen() -> None: def handle_update_screen(app) -> None: """ Check for updates and print a message if available. + Used for non-modal background notification. Args: app: The App instance to check for updates. @@ -44,9 +45,8 @@ def handle_update_screen(app) -> None: remote_version = app.get_remote_version() update_msg = ( - f"\t[blink]An update available, run [green][italic]pip install radio-active==" - + remote_version - + f"[/italic][/green][/blink]\n" + f"\t[blink]An update available, run [green][italic]pipx upgrade radio-active" + f"[/italic][/green][/blink]\n" ) # Add release notes for all missing versions if available @@ -66,6 +66,60 @@ def handle_update_screen(app) -> None: log.debug("Update not available") +def handle_update_modal(app) -> None: + """ + Show a modal popup for update notification with release notes. + + Args: + app: The App instance. + """ + try: + from rich.align import Align + from rich.console import Console + from rich.panel import Panel + + local_version = app.get_version() + remote_version = app.get_remote_version() + + update_msg = ( + f"[bold green]A new version of radio-active is available![/bold green]\n\n" + f"Current version: [yellow]v{local_version}[/yellow]\n" + f"Latest version: [bold green]v{remote_version}[/bold green]\n\n" + f"To update, run:\n[italic]pipx upgrade radio-active[/italic]\n" + ) + + # Add release notes if available + release_notes = app.get_release_notes(local_version, remote_version) + if release_notes: + update_msg += f"\n[bold yellow]What's new since v{local_version}:[/bold yellow]\n{release_notes}" + else: + update_msg += f"\nSee all changes: https://github.com/dpnkrpl/radio-active/blob/main/CHANGELOG.md" + + console = Console() + with console.screen(): + info_panel = Panel( + update_msg, + title="[bold white]🚀 Update Available[/bold white]", + subtitle="Press Enter to continue", + border_style="green", + padding=(1, 4), + width=100, + expand=False, + ) + + # Center the panel visually + console.print("\n" * 4) + console.print(Align.center(info_panel)) + + try: + console.input() + except (EOFError, KeyboardInterrupt): + pass + + except Exception as e: + log.debug(f"Error in update modal: {e}") + + def handle_favorite_table(alias) -> None: """ Print the user's favorite list in a table. diff --git a/radioactive/utilities.py b/radioactive/utilities.py index 0c0b508..1f81aeb 100644 --- a/radioactive/utilities.py +++ b/radioactive/utilities.py @@ -61,6 +61,7 @@ handle_history_table, handle_recording_popup, handle_show_station_info, + handle_update_modal, handle_update_screen, handle_welcome_screen, handle_zen_mode, @@ -94,11 +95,10 @@ def handle_station_selection_menu(handler, last_station, alias) -> Tuple[str, st station_selection_names.append( f"{last_station_info['name'].strip()} (last played station)" ) - try: - station_selection_urls.append(last_station_info["stationuuid"]) - except Exception as e: - log.debug(f"Error: {e}") - station_selection_urls.append(last_station_info["uuid_or_url"]) + uuid = last_station_info.get("stationuuid") or last_station_info.get( + "uuid_or_url", "" + ) + station_selection_urls.append(uuid) fav_stations = alias.alias_map for entry in fav_stations: @@ -148,7 +148,7 @@ def get_key(): try: return ch.decode("utf-8") - except: + except (UnicodeDecodeError, ValueError): return "" else: import termios @@ -212,7 +212,9 @@ def handle_vim_style_prompt(alias, history) -> str: buffer = "" - def get_display(text, matches=[]): + def get_display(text, matches=None): + if matches is None: + matches = [] # The prompt part prompt_text = Text("command : ", style="green") prompt_text.append(text, style="bold cyan") @@ -459,9 +461,9 @@ def handle_user_choice_from_search_result(handler, response) -> Tuple[str, str]: log.debug("Next station requested, picking first one") if user_input in ["r", "R", "random"]: - # pick a random integer withing range - user_input = randint(1, len(response) - 1) - log.debug(f"Radom station id: {user_input}") + # pick a random integer within range (inclusive of last result) + user_input = randint(1, len(response)) + log.debug(f"Random station id: {user_input}") # elif user_input in ["f", "F", "fuzzy"]: # fuzzy find all the stations, and return the selected station id # user_input = fuzzy_find(response) @@ -477,13 +479,13 @@ def handle_user_choice_from_search_result(handler, response) -> Tuple[str, str]: return handle_station_uuid_play(handler, target_response["stationuuid"]) else: log.error("Please enter an ID within the range") - sys.exit(1) + return None, None except ValueError: - log.error("Please enter an valid ID number") - sys.exit(1) + log.error("Please enter a valid ID number") + return None, None except Exception as e: log.error(f"Error: {e}") - sys.exit(1) + return None, None def handle_listen_keypress( @@ -646,18 +648,22 @@ def stop_playback(): else: # child os.setsid() - # Redirect standard file descriptors + # Redirect standard file descriptors, keeping named refs for cleanup sys.stdin.close() - sys.stdout = open(os.devnull, "w") - sys.stderr = open(os.devnull, "w") + _devnull_out = open(os.devnull, "w") + _devnull_err = open(os.devnull, "w") + sys.stdout = _devnull_out + sys.stderr = _devnull_err # child should not listen to keypresses anymore import signal try: signal.pause() - except: + except (AttributeError, OSError): while True: time.sleep(100) + _devnull_out.close() + _devnull_err.close() sys.exit(0) except AttributeError: log.error("Background mode is only supported on Unix-like systems") @@ -894,9 +900,10 @@ def stop_playback(): if len(target_list) == 1: log.info("Station is already playing!") break - player.stop() - player.url = new_target_url - player.play() + if player: + player.stop() + player.url = new_target_url + player.play() handle_current_play_panel(new_station_name) # Save the new station as last played and add to history handle_save_last_station( @@ -908,6 +915,7 @@ def stop_playback(): station_url = new_target_url station_name = new_station_name target_url = new_target_url + auto_fetcher.update_url(target_url) break else: raise Exception("Could not resolve station URL") @@ -926,31 +934,36 @@ def stop_playback(): ) elif user_input == "v+": - new_vol = min(player.volume + 10, 100) - player.set_volume(new_vol) + if player: + new_vol = min(player.volume + 10, 100) + player.set_volume(new_vol) + else: + log.info("Nothing is playing") elif user_input == "v-": - new_vol = max(player.volume - 10, 0) - player.set_volume(new_vol) + if player: + new_vol = max(player.volume - 10, 0) + player.set_volume(new_vol) + else: + log.info("Nothing is playing") elif user_input.startswith("v "): - try: - vol_str = user_input.split(" ")[1].strip() - new_vol = int(vol_str) - if 0 <= new_vol <= 100: - player.set_volume(new_vol) - else: - log.error("Volume must be between 0 and 100") - except Exception: - log.error("Invalid volume format. Use 'v 50'") + if player: + try: + vol_str = user_input.split(" ")[1].strip() + new_vol = int(vol_str) + if 0 <= new_vol <= 100: + player.set_volume(new_vol) + else: + log.error("Volume must be between 0 and 100") + except Exception: + log.error("Invalid volume format. Use 'v 50'") + else: + log.info("Nothing is playing") - elif user_input == "?": + elif user_input in ["?", "help"]: handle_runtime_help_menu() - elif user_input in ["q", "Q", "quit"]: - player.stop() - sys.exit(0) - elif user_input.strip() != "": # Fuzzy match station from aliases or history if not a direct command # Try to see if it's a station name the user typed @@ -971,6 +984,7 @@ def stop_playback(): station_url = url station_name = name target_url = url + auto_fetcher.update_url(target_url) except SystemExit: # Direct play sys.exit(1) on failure, we want to stay in loop pass