import logging import os import io LOG_FILE = 'textual.log' LOGGING_LEVEL = logging.DEBUG LOG_FORMAT = '[%(asctime)s] %(levelname)s in %(module)s: %(message)s' # Clear Log File os.remove(LOG_FILE) logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT) import SubSonic os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" import pygame from textual import on, events, work from textual.app import App, ComposeResult, RenderResult from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label from textual.containers import Horizontal, Vertical, Container from textual.screen import Screen from textual.widgets.option_list import Option from textual.timer import Timer from rich_pixels import Pixels u = 'pseudo' p = 'telegram9dock-MICHEL' 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:] 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:] 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 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:] 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 : Maybe should just take bytes instead of saving it as a file class CoverArtPixels(Container): def __init__(self, cover_art_id: str): super().__init__() self.cover_art_id = cover_art_id conn.getCoverArt(cover_art_id, path=f'art/{cover_art_id}.png') def render(self) -> RenderResult: # Make the size square width, height = self.size if width > height: width = height elif height > width: height = width return Pixels().from_image_path(f'art/{self.cover_art_id}.png', resize=(width, height)) class MainView(Screen): CSS_PATH = 'main.tcss' def compose(self) -> ComposeResult: yield Header(icon="") with Horizontal(): with Vertical(classes="info", id='info'): yield Label("Playing") yield DataTable(id="currently_playing") yield ProgressBar() with Vertical(classes="column"): yield Label("Queue") yield DataTable(id="queue") yield Footer() def on_mount(self) -> None: app.sub_title = "Overview" queueTable: DataTable = self.query_one("#queue") queueTable.cursor_type = "row" queueTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate") currentlyPlaying: DataTable = self.query_one('#currently_playing') currentlyPlaying.cursor_type = "row" currentlyPlaying.add_columns("", "") @on(events.ScreenResume) async def screen_resumed(self, event: events.ScreenResume) -> None: app.sub_title = "Overview" # Update Queue Table queueTable: DataTable = self.query_one('#queue') self._clear_table('#queue') queue: list[SubSonic.Song] = [] queue.extend(app.user_queue) queue.extend(app.song_queue) for song in queue: queueTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate) # Update Currently Playing Table self._clear_table('#currently_playing') if app.currently_playing: song = app.currently_playing currentlyPlaying: DataTable = self.query_one('#currently_playing') currentlyPlaying.add_row("Title", song.title) currentlyPlaying.add_row("Artist", song.artist) currentlyPlaying.add_row("Album", song.album) currentlyPlaying.add_row("Plays", song.playCount if song.playCount else 0) currentlyPlaying.add_row("Duration", song.duration) currentlyPlaying.add_row("Track Number", song.trackNumber) currentlyPlaying.add_row("Bit Rate", song.bitRate) await self.query_one('#info').mount(CoverArtPixels(app.currently_playing.id)) def _clear_table(self, table_id: str) -> None: table: DataTable = self.query_one(table_id) row_keys = [] for row_key, row in table.rows.items(): row_keys.append(row_key) for row_key in row_keys: table.remove_row(row_key) class TextualSubpyPlayer(App): BINDINGS = [ ("space", "play_pause_song()", "⏯️"), ("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 = None song_queue: list[SubSonic.Song] = [] user_queue: 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}.") r = conn.stream(song.id) audio_data = io.BytesIO(r.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) # 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 check_done_playing(self) -> None: if not pygame.mixer.music.get_busy() and not self.is_paused: 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 update_progress(self) -> None: # self.query_one(ProgressBar).advance(1) 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.user_queue.append(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.place = pygame.mixer.music.get_pos() pygame.mixer.music.pause() self.is_paused = True else: pygame.mixer.music.unpause() self.is_paused = False if __name__ == '__main__': app = TextualSubpyPlayer() app.run()