import logging import os import io import asyncio LOG_FILE = 'textual.log' LOGGING_LEVEL = logging.DEBUG LOG_FORMAT = '[%(asctime)s] %(levelname)s in %(module)s: %(message)s' # Clear Log File if os.path.exists(LOG_FILE): os.remove(LOG_FILE) CACHE = True os.makedirs(".cache/art/", exist_ok=True) os.makedirs(".cache/audio/", exist_ok=True) logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT) import SubSonic os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" import pygame from rich.table import Table, Column from textual import on, events, work from textual.reactive import reactive from textual.app import App, ComposeResult, RenderResult from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label from textual.widget import Widget from textual.containers import Horizontal, Vertical, Container, VerticalScroll from textual.screen import Screen from textual.widgets.option_list import Option from textual.timer import Timer from rich_pixels import Pixels from PIL import Image u, p = [s.strip() for s in open('auth', 'r').read().split(',')] conn = SubSonic.SubpyConn(u, p, "https://music.0x01fe.net") pygame.mixer.init() class AlbumSongs(Screen): BINDINGS = [ ('escape', 'die()', "Back"), ('p', 'play()', 'Play Song'), ('q' 'queue()', 'Queue Song') ] selected_song_id: str | None = None songs = None def __init__(self, album_id: str): self.album_id = album_id super().__init__() def compose(self) -> ComposeResult: yield DataTable(id="albumSongs") yield Footer() def on_mount(self) -> None: self.songs = conn.getAlbumSongs(self.album_id) songsTable: DataTable = self.query_one(DataTable) songsTable.cursor_type = 'row' songsTable.add_columns("#", "Title", "Artist", "Duration", "Bitrate") for song in self.songs: songsTable.add_row(song.trackNumber, song.title, song.artist, song.duration, song.bitRate, key=song.id) @on(DataTable.RowHighlighted) def row_highlighted(self, event: DataTable.RowHighlighted) -> None: self.selected_song_id = event.row_key.value def action_play(self) -> None: song_index = self._find_selected_song() app.play_song(self.songs[song_index]) app.song_queue = self.songs[song_index + 1:] app.update_queue_table() def action_queue(self) -> None: app.queue_song(self._find_selected_song()) # Returns index of selected song def _find_selected_song(self) -> int: for i, song in enumerate(self.songs): if song.id == self.selected_song_id: return i # Pops and uninstalls self def action_die(self) -> None: app.pop_screen() app.uninstall_screen("albumSongs") class ArtistAlbums(Screen): BINDINGS = [ ('escape', 'die()', "Back") ] def __init__(self, artist_id: str): self.artist_id = artist_id super().__init__() def compose(self) -> ComposeResult: yield DataTable(id="Albums") yield Footer() def on_mount(self) -> None: albums = conn.getArtistAlbums(self.artist_id) albumTable: DataTable = self.query_one(DataTable) albumTable.cursor_type = 'row' albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") for album in albums: albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) @on(DataTable.RowSelected) async def album_selected(self, event: DataTable.RowSelected) -> None: app.install_screen(AlbumSongs(event.row_key.value), "albumSongs") app.push_screen("albumSongs") # Pops and uninstalls self def action_die(self) -> None: app.pop_screen() app.uninstall_screen("artistAlbums") class Artists(Screen): def compose(self) -> ComposeResult: yield OptionList(id="Artists") yield Footer() def on_mount(self) -> None: artists = conn.getArtists() artistList = self.query_one(OptionList) for artist in artists: artistList.add_option(Option(artist.name, artist.id)) @on(OptionList.OptionSelected) def artist_selected(self, event: OptionList.OptionSelected) -> None: app.install_screen(ArtistAlbums(event.option_id), 'artistAlbums') app.push_screen('artistAlbums') class Albums(Screen): def compose(self) -> ComposeResult: yield DataTable() yield Footer() def on_mount(self) -> None: albums = conn.getAllAlbums() albumTable = self.query_one(DataTable) albumTable.cursor_type = 'row' albumTable.add_columns("Title", "Songs", "Plays", "Year", "Time") for album in albums: albumTable.add_row(album.title, album.songCount, album.playCount, album.year, album.duration, key=album.id) @on(DataTable.RowSelected) def album_selected(self, event: DataTable.RowSelected) -> None: app.install_screen(AlbumSongs(event.row_key.value), 'albumSongs') app.push_screen('albumSongs') class Playlists(Screen): playlists: list[SubSonic.Playlist] | None = None selected_playlist_id: str = None def compose(self) -> ComposeResult: yield Header(icon="") yield DataTable() yield Footer() def on_mount(self) -> None: app.sub_title = "Playlists" self.playlists = conn.getPlaylists() playlistTable = self.query_one(DataTable) playlistTable.cursor_type = "row" playlistTable.add_columns("Name", "Songs", "Time", "Owner") for playlist in self.playlists: playlistTable.add_row(playlist.name, playlist.songCount, playlist.duration, playlist.owner, key=playlist.id) @on(DataTable.RowHighlighted) def row_highlighted(self, event: DataTable.RowHighlighted) -> None: self.selected_playlist_id = event.row_key.value @on(DataTable.RowSelected) def playlist_selected(self, event: DataTable.RowSelected) -> None: app.install_screen(PlaylistSongs(self._find_selected_playlist()), "playlistSongs") app.push_screen('playlistSongs') def _find_selected_playlist(self) -> SubSonic.Playlist: for playlist in self.playlists: if playlist.id == self.selected_playlist_id: return playlist class PlaylistSongs(Screen): BINDINGS = [ ('escape', 'die()', "Back"), ('p', 'play()', 'Play Song'), ('q', 'queue()', 'Queue Song') ] selected_song_id: str | None = None def __init__(self, playlist: SubSonic.Playlist): self.playlist = playlist self.songs = conn.getPlaylist(playlist.id) super().__init__() def compose(self) -> ComposeResult: self.sub_title = self.playlist.name yield Header(icon="") yield DataTable() yield Footer() def on_mount(self) -> None: songTable: DataTable = self.query_one(DataTable) songTable.cursor_type = "row" songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") for song in self.songs: songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) # Pops and uninstalls self def action_die(self) -> None: app.pop_screen() app.uninstall_screen("playlistSongs") @on(DataTable.RowHighlighted) def row_highlighted(self, event: DataTable.RowHighlighted) -> None: self.selected_song_id = event.row_key.value def action_play(self) -> None: song_index = self._find_selected_song() app.play_song(self.songs[song_index]) app.song_queue = self.songs[song_index + 1:] app.update_queue_table() def action_queue(self) -> None: song_id: int = self._find_selected_song() app.queue_song(self.songs[song_id]) # Returns index of selected song def _find_selected_song(self) -> int: for i, song in enumerate(self.songs): if song.id == self.selected_song_id: return i class Favourites(Screen): BINDINGS = [ ('p', 'play()', 'Play Song'), ('q', 'queue()', 'Queue Song') ] songs: list[SubSonic.Song] | None = None selected_song_id: str | None = None def compose(self) -> ComposeResult: yield Header(icon="") yield DataTable() yield Footer() def on_mount(self) -> None: app.sub_title = "Favourites" self.songs = conn.getStarredSongs() songTable = self.query_one(DataTable) songTable.cursor_type = "row" songTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") for song in self.songs: songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=song.id) @on(DataTable.RowHighlighted) def row_highlighted(self, event: DataTable.RowHighlighted) -> None: self.selected_song_id = event.row_key.value def action_play(self) -> None: song_index = self._find_selected_song() app.play_song(self.songs[song_index]) app.song_queue = self.songs[song_index + 1:] app.update_queue_table() def action_queue(self) -> None: app.queue_song(self._find_selected_song()) # Returns index of selected song def _find_selected_song(self) -> int: for i, song in enumerate(self.songs): if song.id == self.selected_song_id: return i # TODO fix coverart class CoverArtPixels(Container): cover = reactive[Image.Image | None](None) def render(self) -> RenderResult: if self.cover: width, height = self.size width -= 4 height -= 4 if width > height: width = height elif height > width: height = width return Pixels.from_image(self.cover, resize=(width, height)) return '' @work(exclusive=True, thread=True) def get_cover_art(self, cover_art_id: str) -> None: image = None cache_path = f".cache/art/{cover_art_id}.png" if CACHE and os.path.exists(cache_path): try: image = Image.open(cache_path) self.cover = image except Exception as e: logging.error(f"Failed to load cached cover art {cache_path}: {e}") if image is None: try: cover_art_data = conn.getCoverArtRaw(cover_art_id) image = Image.open(io.BytesIO(cover_art_data)) if CACHE: try: image.save(cache_path, "PNG") except Exception as e: logging.error(f"Failed to save cover art to cache {cache_path}: {e}") self.cover = image except Exception as e: logging.error(f"Failed to get cover art: {e}") self.cover = None return class QueueTable(Widget): BINDINGS = [ ("p", "play_queue_song()", "Play") ] song_ids: list[str] = [] selected_row_key: int = 0 def compose(self) -> ComposeResult: yield DataTable(id="songsqueue", cursor_type='row') def on_mount(self) -> None: table = self.query_one(DataTable) table.add_columns("Title", "Album", "Artist", "Plays", "Duration") self.update_queue() def update_queue(self) -> None: table = self.query_one(DataTable) table.clear() queue: list[SubSonic.Song] = app.user_queue.copy() queue.extend(app.song_queue) self.song_ids = [] for i, song in enumerate(queue): table.add_row(song.title, song.album, song.artist, str(song.playCount), str(song.duration), key=str(i)) self.song_ids.append(song.id) @on(DataTable.RowHighlighted) def set_selected_row_key(self, e: DataTable.RowHighlighted) -> None: if e.row_key.value: self.selected_row_key = int(e.row_key.value) def action_play_queue_song(self) -> None: # Reconstruct the queue to get the selected song object queue: list[SubSonic.Song] = app.user_queue.copy() queue.extend(app.song_queue) if self.selected_row_key >= len(queue): return # Avoid index out of bounds # Add currently playing song to history if app.currently_playing: app.play_history.append(app.currently_playing) # Add all songs before the selected one to history songs_to_skip = queue[:self.selected_row_key] app.play_history.extend(songs_to_skip) song_to_play = queue[self.selected_row_key] # The new queue is everything after the selected song new_queue = queue[self.selected_row_key + 1:] # Play the song app.play_song(song_to_play) # The user queue is cleared, and the song queue is the rest of the songs app.user_queue = [] app.song_queue = new_queue # Update the visual queue table self.update_queue() # Fine to be a static renderable because it's not big enough to need a scroll class NowPlayingInfo(Container): song = reactive[SubSonic.Song | None](None, layout=True) # Place is from pygame.mixer.music.get_pos and is in ms place: reactive[int] = reactive(0) def render(self) -> RenderResult: if self.song: if self.song.title and self.song.artist: # Convert seconds to MM:SS duration_min, duration_sec = divmod(self.song.duration, 60) duration_formatted = f"{duration_min:02d}:{duration_sec:02d}" place_in_seconds = self.place // 1000 place_min, place_sec = divmod(place_in_seconds, 60) place_formatted = f"{place_min:02d}:{place_sec:02d}" s = f"Song: {self.song.title}\nAlbum: {self.song.album}\nArtist: {self.song.artist}\nProgress: {place_formatted} / {duration_formatted}\n" return s return "Nothing Playing" class NowPlaying(Widget): def compose(self) -> ComposeResult: yield NowPlayingInfo(id="textinfo") yield CoverArtPixels(id="coverart") def update_song(self, song: SubSonic.Song) -> None: info = self.query_one(NowPlayingInfo) info.song = song if song.albumId: cover_art = self.query_one(CoverArtPixels) album = conn.getAlbumSongs(song.albumId) if album: cover_id: str | None = album[0].coverArtId if cover_id: cover_art.get_cover_art(cover_id) def set_place(self, place: int) -> None: info = self.query_one(NowPlayingInfo) info.place = place # https://textual.textualize.io/widgets/data_table/#labeled-rows # Add this ^ to queue sometime plz class MainView(Screen): CSS_PATH = 'main.tcss' def compose(self) -> ComposeResult: yield Header(icon="") with Horizontal(): with Vertical(classes="info", id='info'): yield NowPlaying(id="now-playing") yield ProgressBar() with Vertical(classes="column"): yield Label("Queue") yield QueueTable(id="queue") yield Footer() def on_mount(self) -> None: app.sub_title = "Overview" @on(events.ScreenResume) def screen_resumed(self, event: events.ScreenResume) -> None: app.sub_title = "Overview" queue_table = self.query_one(QueueTable) queue_table.update_queue() app.update_now_playing() class TextualSubpyPlayer(App): BINDINGS = [ ("r", "rewind()", "⏪"), ("space", "play_pause_song()", "⏯"), ("s", "skip()", "⏭"), ("f1", "switch_mode('main')", "Overview"), ("f2", "switch_mode('artists')", "Artists"), ("f3", "switch_mode('albums')", "Albums"), ("f4", "switch_mode('favourites')", "Favourites"), ("f5", "switch_mode('playlists')", "Playlists") ] MODES = { "main" : MainView, "artists" : Artists, "albums" : Albums, "favourites" : Favourites, "playlists" : Playlists } currently_playing: SubSonic.Song | None = reactive(SubSonic.Song()) song_queue: list[SubSonic.Song] = reactive([]) user_queue: list[SubSonic.Song] = reactive([]) play_history: list[SubSonic.Song] = [] is_paused: bool = False place: int = 0 check_playing_timer: Timer | None = None def on_mount(self) -> None: self.title = "SubPy" self.switch_mode('main') def on_key(self, event: events.Key) -> None: # logging.debug(f'Key Pressed {event.name}') pass @work(exclusive=True) async def play_song(self, song: SubSonic.Song) -> None: self.currently_playing = song self.is_paused = False logging.debug(f"Playing song {song.title} by {song.artist}.") self.notify(f"Playing \"{song.title}\" by {song.artist}.") audio_data = None audio_cache_path = f'.cache/audio/{song.id}.ogg' if CACHE and os.path.exists(audio_cache_path): with open(audio_cache_path, 'rb') as file: audio_data = io.BytesIO(file.read()) else: # Run the blocking stream call in a separate thread r = await asyncio.to_thread(conn.stream, song.id) audio_content = r.content if CACHE: # Also run the file writing in a thread await asyncio.to_thread(self._cache_audio, audio_cache_path, audio_content) audio_data = io.BytesIO(audio_content) pygame.mixer.music.load(audio_data) pygame.mixer.music.play() self.check_playing_timer = self.set_interval(1 / 5, self.check_done_playing, pause=False) self.progess_timer = self.set_interval(0.3, self.update_progress, pause=False) self.update_queue_table() self.update_now_playing() # playingLabel: Label = self.query_one(Label) # playingLabel. # Set callback loop for progress bar # self.progress_timer = self.set_interval(1 / 10, self.update_progress, pause=False) def _cache_audio(self, path: str, content: bytes) -> None: """Helper method to write audio to cache.""" with open(path, "wb") as file: file.write(content) def check_done_playing(self) -> None: if not pygame.mixer.music.get_busy() and not self.is_paused: self.play_history.append(self.currently_playing) queue = self.user_queue queue.extend(self.song_queue) next_song = queue.pop(0) self.play_song(next_song) self.check_playing_timer.stop() def action_skip(self) -> None: if self.check_playing_timer: self.check_playing_timer.stop() if self.currently_playing: self.play_history.append(self.currently_playing) queue = self.user_queue queue.extend(self.song_queue) self.play_song(queue.pop(0)) def action_rewind(self) -> None: if not self.play_history: return if self.check_playing_timer: self.check_playing_timer.stop() if self.currently_playing: self.song_queue.insert(0, self.currently_playing) song_to_play = self.play_history.pop() self.play_song(song_to_play) self.update_queue_table() def update_progress(self) -> None: if isinstance(self.screen, MainView) and self.currently_playing: now_playing = self.screen.query_one(NowPlaying) now_playing.set_place(pygame.mixer.music.get_pos()) def queue_song(self, song: SubSonic.Song) -> None: logging.debug(f'Queue\'d song "{song.title}" by {song.artist}.') self.notify(f'Queue\'d song "{song.title}" by {song.artist}.') self.update_user_queue(song) def action_play_pause_song(self) -> None: self.play_pause_song() def play_pause_song(self) -> None: if not self.currently_playing: return None if not self.is_paused: self.notify("Playback paused") self.place = pygame.mixer.music.get_pos() pygame.mixer.music.pause() self.is_paused = True else: self.notify("Resuming Playback") pygame.mixer.music.unpause() self.is_paused = False def update_now_playing(self) -> None: if isinstance(self.screen, MainView) and self.currently_playing: logging.debug("Updating now playing widget") now_playing = self.screen.query_one(NowPlaying) now_playing.update_song(self.currently_playing) def update_queue_table(self) -> None: if isinstance(self.screen, MainView): logging.debug("Updating queue through app") queue_table = self.screen.query_one(QueueTable) queue_table.update_queue() def update_user_queue(self, song: SubSonic.Song) -> None: self.user_queue.append(song) self.update_queue_table() def update_song_queue(self, song: SubSonic.Song) -> None: self.song_queue.append(song) self.update_queue_table() if __name__ == '__main__': app = TextualSubpyPlayer() app.run()