Show current song

README.md

@@ -153,6 +153,7 @@ h/H/help/?: Show this help message
 r/R/record: Record a station
 f/F/fav: Add station to favorite list
 rf/RF/recordfile: Specify a filename for the recording.
+s: Start/stop playing the station
 ```
 
 

radioactive/__main__.py

@@ -3,6 +3,7 @@
 import signal
 import sys
 from time import sleep
+from typing import Optional
 
 from zenlog import log
 
@@ -28,7 +29,7 @@
 
 # globally needed as signal handler needs it
 # to terminate main() properly
-player = None
+player: Optional[Player] = None
 
 
 def final_step(options, last_station, alias, handler):
@@ -41,7 +42,12 @@ def final_step(options, last_station, alias, handler):
     if options["curr_station_name"].strip() == "":
         options["curr_station_name"] = "N/A"
 
-    player = Player(options["target_url"], options["volume"], options["loglevel"])
+    def curry_current_station_name_state_changed(callback_player: Optional[Player]):
+        # TODO clean this up, as soon as there's a type for stations,
+        # which holds at least the station and url
+        handle_current_play_panel(callback_player, options["curr_station_name"])
+
+    player = Player(options["target_url"], options["volume"], options["loglevel"],[curry_current_station_name_state_changed])
 
     handle_save_last_station(
         last_station, options["curr_station_name"], options["target_url"]
@@ -52,8 +58,6 @@ def final_step(options, last_station, alias, handler):
             alias, options["curr_station_name"], options["target_url"]
         )
 
-    handle_current_play_panel(options["curr_station_name"])
-
     if options["record_stream"]:
         handle_record(
             options["target_url"],
@@ -65,6 +69,7 @@ def final_step(options, last_station, alias, handler):
         )
 
     handle_listen_keypress(
+        player,
         alias,
         target_url=options["target_url"],
         station_name=options["curr_station_name"],
@@ -286,4 +291,8 @@ def signal_handler(sig, frame):
 signal.signal(signal.SIGINT, signal_handler)
 
 if __name__ == "__main__":
-    main()
+    try:
+        main()
+    finally:
+        if player:
+            player.stop()

radioactive/player.py

@@ -1,4 +1,5 @@
 """ FFplay process handler """
+from __future__ import annotations
 
 import os
 import signal
@@ -7,6 +8,7 @@
 import threading
 from shutil import which
 from time import sleep
+from typing import Callable, List
 
 import psutil
 from zenlog import log
@@ -35,32 +37,74 @@ def kill_background_ffplays():
 
 
 class Player:
-
     """FFPlayer handler, it holds all the attributes to properly execute ffplay
     FFmepg required to be installed separately
     """
 
-    def __init__(self, URL, volume, loglevel):
-        self.url = URL
+    def __init__(
+        self,
+        url: str,  # TODO pass the station in the future
+        volume: int,  # [0..100] everything else is interpreted as 100 by ffplay
+        loglevel: str,
+        state_changed: List[Callable[[Player], None]] = None
+    ):
+        self.url = url
         self.volume = volume
-        self.is_playing = False
         self.process = None
         self.exe_path = None
+        self._stream_title: str = ""
         self.program_name = "ffplay"  # constant value
+        self._metadata_program = "ffprobe"  # constant value
+        self._exe_metadata_path = None
         self.loglevel = loglevel
+        self._state_changed = state_changed
 
-        log.debug("player: url => {}".format(self.url))
+        log.debug(f"player: url => {self.url}")
         # check if FFplay is installed
         self.exe_path = which(self.program_name)
-        log.debug("FFplay: {}".format(self.exe_path))
-
         if self.exe_path is None:
-            log.critical("FFplay not found, install it first please")
+            log.critical(f"{self.program_name} not found, install it first please")
             sys.exit(1)
+        else:
+            log.debug(f"{self.program_name}: {self.exe_path}")
 
-        self.start_process()
+        self._exe_metadata_path = which(self._metadata_program)
+        if self._exe_metadata_path is None:
+            log.critical(f"{self._metadata_program} not found, install it first please")
+            sys.exit(1)
+        else:
+            log.debug(f"{self.program_name}: {self._exe_metadata_path}")
+        self._start_process()
+
+    def read_title(self):
+        ffprobe_command = [
+            self._exe_metadata_path,
+            "-v", "error",
+            "-select_streams", "a:0",
+            # TODO maybe this tag is different on different stations
+            "-show_entries", "format_tags=StreamTitle",
+            "-of", "default=noprint_wrappers=1:nokey=1",
+            self.url
+        ]
+        try:
+            ffprobe_proc = subprocess.Popen(
+                ffprobe_command,
+                shell=False,
+                stdout=subprocess.PIPE,  # Capture standard output
+                stderr=subprocess.PIPE,  # Capture standard error
+                text=True,  # Use text mode to capture strings
+            )
+            title = ""
+            for line in ffprobe_proc.stdout:
+                title += line
+            # truncate trailing \n
+            return title[:-1] if title[-1] == "\n" else title
+        except Exception as e:
+            # Handle exceptions that might occur during process setup
+            log.error(f"Error while reading current track: {e}")
+            return f"Error while reading current track: {e}"
 
-    def start_process(self):
+    def _start_process(self):
         ffplay_commands = [
             self.exe_path,
             "-volume",
@@ -86,19 +130,32 @@ def start_process(self):
                 stderr=subprocess.PIPE,  # Capture standard error
                 text=True,  # Use text mode to capture strings
             )
-            self.is_running = True
-            log.debug("player: ffplay => PID {} initiated".format(self.process.pid))
+            log.debug(
+                f"player: {self.program_name} => PID {self.process.pid} initiated")
             # Create a thread to continuously capture and check error output
             error_thread = threading.Thread(target=self.check_error_output)
             error_thread.daemon = True
             error_thread.start()
+            # Create a thread to update the stream title information
+            title_update_thread = threading.Thread(target=self.update_title)
+            title_update_thread.daemon = True
+            title_update_thread.start()
 
         except Exception as e:
             # Handle exceptions that might occur during process setup
-            log.error("Error while starting radio: {}".format(e))
+            log.error(f"Error while starting radio: {e}")
+
+    def update_title(self):
+        while self.is_playing:
+            new_title = self.read_title()
+            if self._stream_title != new_title:
+                self._stream_title = new_title
+                self._informAboutChange()
+            sleep(1)
+        self._stream_title = ""
 
     def check_error_output(self):
-        while self.is_running:
+        while self.is_playing:
             stderr_result = self.process.stderr.readline()
             if stderr_result:
                 print()  # pass a blank line to command for better log messages
@@ -109,10 +166,8 @@ def check_error_output(self):
                     # only showing the server response
                     log.error(stderr_result.split(": ")[1])
                 except Exception as e:
-                    log.debug("Error: {}".format(e))
+                    log.debug(f"Error: {e}")
                     pass
-
-                self.is_running = False
                 self.stop()
             sleep(2)
 
@@ -147,32 +202,50 @@ def is_active(self):
             log.debug("Process not found")
             return False
         except Exception as e:
-            log.error("Error while checking process status: {}".format(e))
+            log.error(f"Error while checking process status: {e}")
             return False
 
     def play(self):
         """Play a station"""
         if not self.is_playing:
-            pass  # call the init function again ?
+            self._start_process()
+
+    @property
+    def is_playing(self):
+        return self.process
+
+    @property
+    def stream_title(self):
+        return self._stream_title
 
     def stop(self):
         """stop the ffplayer"""
-
         if self.is_playing:
+            ffplay_proc = self.process
+            self.process = None
+            self._stream_title = ""
+            self._informAboutChange()
             try:
-                self.process.terminate()  # Terminate the process gracefully
-                self.process.wait(timeout=5)  # Wait for process to finish
+                ffplay_proc.terminate()  # Terminate the process gracefully
+                ffplay_proc.wait(timeout=5)  # Wait for process to finish
                 log.info("Radio playback stopped successfully")
             except subprocess.TimeoutExpired:
                 log.warning("Radio process did not terminate, killing...")
-                self.process.kill()  # Kill the process forcefully
+                ffplay_proc.kill()  # Kill the process forcefully
             except Exception as e:
-                log.error("Error while stopping radio: {}".format(e))
+                log.error(f"Error while stopping radio: {e}")
                 raise
-            finally:
-                self.is_playing = False
-                self.process = None
         else:
             log.debug("Radio is not currently playing")
             current_pid = os.getpid()
             os.kill(current_pid, signal.SIGINT)
+
+    def _informAboutChange(self):
+        for changed in self._state_changed:
+            changed(self)
+
+    # TODO pass the station in the future
+    def switch_url(self, url: str):
+        self.stop()
+        self.url = url
+        self.play()

radioactive/utilities.py

@@ -3,6 +3,7 @@
 import datetime
 import os
 import sys
+from typing import Optional
 
 from pick import pick
 from rich import print
@@ -13,7 +14,7 @@
 from zenlog import log
 
 from radioactive.last_station import Last_station
-from radioactive.player import kill_background_ffplays
+from radioactive.player import kill_background_ffplays, Player
 from radioactive.recorder import record_audio_auto_codec, record_audio_from_url
 
 RED_COLOR = "\033[91m"
@@ -260,6 +261,7 @@ def handle_save_last_station(last_station, station_name, station_url):
 
 
 def handle_listen_keypress(
+    player: Optional[Player],
     alias,
     target_url,
     station_name,
@@ -272,7 +274,7 @@ def handle_listen_keypress(
     log.info("Press '?' to see available commands\n")
     while True:
         user_input = input("Enter a command to perform an action: ")
-        if user_input == "r" or user_input == "R" or user_input == "record":
+        if user_input in ["r", "R", "record"]:
             handle_record(
                 target_url,
                 station_name,
@@ -281,7 +283,7 @@ def handle_listen_keypress(
                 record_file_format,
                 loglevel,
             )
-        elif user_input == "rf" or user_input == "RF" or user_input == "recordfile":
+        elif user_input in ["rf", "RF", "recordfile"]:
             # if no filename is provided try to auto detect
             # else if ".mp3" is provided, use libmp3lame to force write to mp3
 
@@ -311,32 +313,37 @@ def handle_listen_keypress(
                     loglevel,
                 )
 
-        elif user_input == "f" or user_input == "F" or user_input == "fav":
+        elif user_input in ["f", "F", "fav"]:
             handle_add_to_favorite(alias, station_name, station_url)
 
-        elif user_input == "q" or user_input == "Q" or user_input == "quit":
+        elif user_input in ["q", "Q", "quit"]:
             kill_background_ffplays()
             sys.exit(0)
-        elif user_input == "w" or user_input == "W" or user_input == "list":
+        elif user_input in ["w", "W", "list"]:
             alias.generate_map()
             handle_favorite_table(alias)
-
-        elif (
-            user_input == "h"
-            or user_input == "H"
-            or user_input == "?"
-            or user_input == "help"
-        ):
+        elif user_input in ["s", "S"]:
+            if player:
+                if player.is_playing:
+                    player.stop()
+                else:
+                    player.play()
+        elif user_input in ["h", "H", "?", "help"]:
             log.info("h/help/?: Show this help message")
             log.info("q/quit: Quit radioactive")
             log.info("r/record: Record a station")
             log.info("f/fav: Add station to favorite list")
             log.info("rf/recordfile: Specify a filename for the recording")
+            log.info("s: Start/stop playing the station")
             # TODO: u for uuid, link for url, p for setting path
 
 
-def handle_current_play_panel(curr_station_name=""):
-    panel_station_name = Text(curr_station_name, justify="center")
+def handle_current_play_panel(player: Optional[Player], curr_station_name=""):
+    print()
+    now_playing = curr_station_name
+    if player:
+        now_playing += f"\n{player.read_title()}"
+    panel_station_name = Text(now_playing, justify="center")
 
     station_panel = Panel(panel_station_name, title="[blink]:radio:[/blink]", width=85)
     console = Console()