import multiprocessing import configparser import logging import asyncio import random import math import os import re import io LOG_FILE = 'textual.log' LOGGING_LEVEL = logging.DEBUG LOG_FORMAT = '[%(asctime)s] %(levelname)s in %(module)s: %(message)s' # Read config file config = configparser.ConfigParser() config.read('config.ini') CACHE = config.getboolean('STREAMING', 'Cache') BITRATE = config.get('STREAMING', 'Bitrate') BITRATE = re.findall(r'[0-9]+', BITRATE)[0] BITRATE = int(BITRATE) if BITRATE else BITRATE IMAGE_MODE = config.getboolean('GENERAL', 'GUI') import SubSonic os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide" import pygame import pynput from rich.text import Text from textual import on, events, work from textual.binding import Binding from textual.reactive import reactive from textual.app import App, ComposeResult, RenderResult from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label, Input, Placeholder, Static, Rule from textual.widget import Widget 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 from PIL import Image, ImageFile ImageFile.LOAD_TRUNCATED_IMAGES = True 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'), ('m', 'shuffle_mix()', 'Shuffle Play') ] selected_song_id: str | None = None songs: list[SubSonic.Song] 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") def action_shuffle_mix(self) -> None: q = self.songs.copy() random.shuffle(q) app.play_song(q.pop(0)) app.song_queue = q app.update_queue_table() 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 AlbumPanel(Widget, can_focus=True, can_focus_children=True): album: SubSonic.Album def compose(self) -> ComposeResult: yield CoverArtPixels(classes="albumArt") yield Label(classes="albumName") def set_album(self, album: SubSonic.Album) -> None: self.album = album coverArt = self.query_one(CoverArtPixels) coverArt.get_cover_art(self.album.id) label = self.query_one(Label) label.update(self.album.album) @on(events.Click) def clicked(self, event: events.Click) -> None: app.install_screen(AlbumSongs(self.album.id), 'albumSongs') app.push_screen('albumSongs') def clear(self) -> None: self.album = None coverArt = self.query_one(CoverArtPixels) coverArt.cover = None label = self.query_one(Label) label.update('') class ArtistAlbumsGUI(Screen): BINDINGS = [ ('escape', 'die()', "Back"), Binding('a', 'page_backward()', 'Prev. Page'), Binding('d', 'page_forward()', 'Next Page') ] artist: SubSonic.Artist albums: list[SubSonic.Album] page: int = 1 pages: int = 1 albums_displayed: int def __init__(self, artist: SubSonic.Artist): self.artist = artist super().__init__() def compose(self) -> ComposeResult: with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") yield Label(classes='artistNameHeader') yield Rule() with Horizontal(): yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") with Horizontal(): yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") yield AlbumPanel(classes="albumPanel") yield Label(id='page-counter') yield Footer() def on_mount(self) -> None: self.albums = conn.getArtistAlbums(self.artist.id) album_panels = self.query(AlbumPanel) artist_header = self.query_one(Label) artist_header.update(self.artist.name) self.pages = math.ceil(len(self.albums) / len(album_panels)) self.update_pager() self.albums_displayed = 0 for album_panel in album_panels: album_panel.set_album(self.albums[self.albums_displayed]) self.albums_displayed += 1 if self.albums_displayed >= len(self.albums): return def update_pager(self) -> None: pager = self.query_one('#page-counter', Label) pager.update(f'{self.page} ... {self.pages}') def _clear_panels(self) -> None: album_panels = self.query(AlbumPanel) for panel in album_panels: panel.clear() def action_die(self) -> None: app.pop_screen() app.uninstall_screen("artistAlbums") def action_page_forward(self) -> None: album_panels = self.query(AlbumPanel) if self.page + 1 > self.pages: return self.page += 1 self.update_pager() self._clear_panels() for album_panel in album_panels: if self.albums_displayed >= len(self.albums): return album_panel.set_album(self.albums[self.albums_displayed]) self.albums_displayed += 1 def action_page_backward(self) -> None: album_panels = self.query(AlbumPanel) if self.page - 1 <= 0: return # count how many panels actually are displaying an album real_displayed = 0 for album_panel in album_panels: if album_panel.album: real_displayed += 1 self.page -= 1 self.update_pager() self._clear_panels() self.albums_displayed -= len(album_panels) + real_displayed for album_panel in album_panels: if self.albums_displayed >= len(self.albums): return album_panel.set_album(self.albums[self.albums_displayed]) self.albums_displayed += 1 class Artists(Screen): artists: list[SubSonic.Artist] def compose(self) -> ComposeResult: yield Header(icon='') with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") with Vertical(id="artistContainer"): yield Input(placeholder="Search") yield OptionList(id="Artists") yield Footer() def on_mount(self) -> None: app.sub_title = 'Artists' self.artists = conn.getArtists() artistList = self.query_one(OptionList) for artist in self.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') @on(Input.Submitted) def search_submit(self, event: Input.Submitted) -> None: search_value = event.value.lower() artist_list = self.query_one(OptionList) artist_list.clear_options() for artist in self.artists: if search_value in artist.name.lower(): artist_list.add_option(Option(artist.name, artist.id)) class ArtistLabel(Label, can_focus=True): def render(self) -> RenderResult: if self.has_focus: return Text(str(self.content), style="white on blue") else: return self.content class ArtistArt(Static): art = reactive[Image.Image | None](None) def render(self) -> RenderResult: if self.art: width, height = self.size height *= 2 if width > height: width = height elif height > width: height = width try: return Pixels.from_image(self.art, resize=(width, height)) except Exception as e: logging.error(f'Couldn\'t load image {e}') return '' @work(exclusive=True, thread=True) def get_artist_image(self, artist_id: str) -> None: artist_id = 'ar-' + artist_id image = None cache_path = f".cache/art/artists/{artist_id}.png" if CACHE and os.path.exists(cache_path): try: self.art = Image.open(cache_path) except Exception as e: logging.error(f"Failed to load cached artist art {cache_path}: {e}") if image is None: try: artist_art_data = conn.getCoverArtRaw(artist_id, size=300) image = Image.open(io.BytesIO(artist_art_data)) if CACHE: try: image.save(cache_path, 'PNG') except Exception as e: logging.error(f"Failed to save artist art to cache {cache_path}: {e}") self.art = image except Exception as e: logging.error(f"Failed to get artist art: {e}") self.art = None class ArtistPanel(Widget, can_focus=True, can_focus_children=True): artist: SubSonic.Artist | None def compose(self) -> ComposeResult: yield ArtistArt(classes="artistArt") yield ArtistLabel(classes="artistName") def set_artist(self, artist: SubSonic.Artist) -> None: self.artist = artist art: ArtistArt = self.query_one(ArtistArt) art.get_artist_image(self.artist.id) label = self.query_one(Label) label.update(self.artist.name) @on(events.Click) def clicked(self, event: events.Click) -> None: if not self.artist: return logging.debug(f'ArtistPanel {self.artist.name} was clicked.') app.install_screen(ArtistAlbumsGUI(self.artist), 'artistAlbums') app.push_screen('artistAlbums') def clear(self) -> None: self.artist = None art: ArtistArt = self.query_one(ArtistArt) art.art = None label = self.query_one(Label) label.update('') class ArtistsGUI(Screen): BINDINGS = [ Binding('a', "page_backward()", "Prev. Page"), Binding('d', "page_forward()", "Next Page") ] artists: list[SubSonic.Artist] artists_displayed: int = 0 page: int = 1 pages: int = 1 def compose(self) -> ComposeResult: yield Header(icon='') with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") yield Input(placeholder="Search", id="artistSearchBar") with Horizontal(): yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") with Horizontal(): yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") with Horizontal(): yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield ArtistPanel(classes="artistPanel") yield Label(id="page-counter") yield Footer() def on_mount(self) -> None: app.sub_title = 'Artists' self.artists = conn.getArtists() artist_panels = self.query(ArtistPanel) self.artists_displayed = 0 self.pages = math.ceil(len(self.artists)/len(artist_panels)) self.update_pager() for artist_panel in artist_panels: if self.artists_displayed >= len(self.artists) - 1: return artist_panel.set_artist(self.artists[self.artists_displayed]) self.artists_displayed += 1 def update_pager(self) -> None: pager = self.query_one("#page-counter", Label) pager.update(f"{self.page} ... {self.pages}") def action_page_forward(self) -> None: artist_panels = self.query(ArtistPanel) if self.page + 1 > self.pages: return self.page += 1 self.update_pager() self._clear_panels() for artist_panel in artist_panels: if self.artists_displayed >= len(self.artists) - 1: return artist_panel.set_artist(self.artists[self.artists_displayed]) self.artists_displayed += 1 def action_page_backward(self) -> None: artist_panels = self.query(ArtistPanel) if self.page - 1 <= 0: return # Count artist panels with info in them real_displayed = 0 for artist_panel in artist_panels: if artist_panel.artist: real_displayed += 1 self.page -= 1 self.update_pager() self._clear_panels() self.artists_displayed -= len(artist_panels) + real_displayed for artist_panel in artist_panels: artist_panel.set_artist(self.artists[self.artists_displayed]) self.artists_displayed += 1 def _clear_panels(self) -> None: artist_panels = self.query(ArtistPanel) for artist_panel in artist_panels: artist_panel.clear() @on(Input.Submitted) def search_submit(self, event: Input.Submitted) -> None: search_value = event.value.lower() logging.debug(search_value) artist_panels = self.query(ArtistPanel) self.artists_displayed = 0 filtered_artists = [] for artist in self.artists: if search_value in artist.name.lower(): filtered_artists.append(artist) self.page = 1 self.pages = math.ceil(len(filtered_artists) / len(artist_panels)) self.update_pager() self._clear_panels() for artist_panel in artist_panels: if self.artists_displayed >= len(filtered_artists): return artist_panel.set_artist(filtered_artists[self.artists_displayed]) self.artists_displayed += 1 class Albums(Screen): albums: list[SubSonic.Album] def compose(self) -> ComposeResult: yield Header(icon='') with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("* F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") with Vertical(id="albumContainer"): yield Input(placeholder="Search", id="albumSearchBar") yield DataTable(id="albumList") yield Footer() def on_mount(self) -> None: app.sub_title = "Albums" self.albums = conn.getAllAlbums() albumTable = self.query_one(DataTable) albumTable.cursor_type = 'row' albumTable.add_columns("Title", "Artist", "Songs", "Plays", "Year", "Time") for album in self.albums: albumTable.add_row(album.title, album.artist, 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') @on(Input.Submitted) def search_submit(self, event: Input.Submitted) -> None: search_value = event.value.lower() album_table = self.query_one(DataTable) album_table.clear() for album in self.albums: if search_value in album.title.lower() or search_value in album.artist.lower(): album_table.add_row(album.title, album.artist, album.songCount, album.playCount, album.year, album.duration, key=album.id) class Playlists(Screen): playlists: list[SubSonic.Playlist] | None = None selected_playlist_id: str = None def compose(self) -> ComposeResult: yield Header(icon="") with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("* F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") 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'), ('m', 'shuffle_mix()', 'Shuffle Play') ] selected_song_id: str | None = None songs: list[SubSonic.Song] 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 i, song in enumerate(self.songs): songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=str(i)) # 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 = self.songs[int(event.row_key.value)].id 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 def action_shuffle_mix(self) -> None: q = self.songs.copy() random.shuffle(q) app.play_song(q.pop(0)) app.song_queue = q app.update_queue_table() class Favourites(Screen): BINDINGS = [ ('p', 'play()', 'Play Song'), ('q', 'queue()', 'Queue Song'), ('m', 'shuffle_mix()', 'Shuffle Play') ] songs: list[SubSonic.Song] selected_song_id: str | None = None def compose(self) -> ComposeResult: yield Header(icon="") with Horizontal(id="headerbs"): yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("* F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") 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: song_index = self._find_selected_song() app.queue_song(self.songs[song_index]) # 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 def action_shuffle_mix(self) -> None: q = self.songs.copy() random.shuffle(q) app.play_song(q.pop(0)) app.song_queue = q app.update_queue_table() class CoverArtPixels(Container): cover = reactive[Image.Image | None](None) def render(self) -> RenderResult: if self.cover: width, height = self.size height *= 2 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: self.cover = Image.open(cache_path) 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, size=300) 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 class QueueTable(Widget): BINDINGS = [ ("p", "play_queue_song()", "Play"), ("f", "star_song()", "Star Song") ] 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() async def action_star_song(self) -> None: queue: list[SubSonic.Song] = app.user_queue.copy() queue.extend(app.song_queue) song = queue[self.selected_row_key] asyncio.create_task(asyncio.to_thread(conn.star, song.id)) # 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(id="headerbs"): yield Label(Text("* F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb") yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb") yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb") yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb") yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb") 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() # Startup keyboard listener for media keys def listen_media_keys(queue: multiprocessing.Queue): media_log_file = 'media_keys.log' logger = logging.getLogger('media_keys') logger.setLevel(LOGGING_LEVEL) formatter = logging.Formatter('[%(asctime)s] %(levelname)s in %(module)s: %(message)s') handler = logging.FileHandler(media_log_file) handler.setFormatter(formatter) logger.addHandler(handler) logger.info('media_key logger initialized') try: def on_press(key: pynput.keyboard.Key | pynput.keyboard.KeyCode | None) -> None: logger.debug(f'Process Key: {key}') match key: case pynput.keyboard.Key.media_play_pause: logger.info('Media play/pause key pressed') queue.put('play_pause') case pynput.keyboard.Key.media_next: logger.info('Media next key pressed') queue.put('next') case pynput.keyboard.Key.media_previous: logger.info('Media rewind pressed') queue.put('prev') case pynput.keyboard.Key.media_volume_down: logger.info('Media volume down') queue.put('vdown') case pynput.keyboard.Key.media_volume_up: logger.info('Media volume up') queue.put('udown') with pynput.keyboard.Listener(on_press=on_press) as listener: logger.info('Media key listener started') listener.join() except Exception as e: pass logger.error(f'Media key listener failed: {e}') class TextualSubpyPlayer(App): BINDINGS = [ ("r", "rewind()", "⏪"), ("space", "play_pause_song()", "⏯"), ("s", "skip()", "⏭"), ("d", "volume_down()", "🕩"), ("u", "volume_up()", "🕪"), ("F", "star_c_song()", "Star Current Song"), Binding('f1', "switch_mode('main')", 'Overview', show=False), Binding('f2', "switch_mode('artists')", "Artists", show=False), Binding('f3', "switch_mode('albums')", "Albums", show=False), Binding('f4', "switch_mode('favourites')", "Favourites", show=False), Binding('f5', "switch_mode('playlists')", "Playlists", show=False) ] MODES = { "main" : MainView, "artists" : ArtistsGUI, "albums" : Albums, "favourites" : Favourites, "playlists" : Playlists } currently_playing: SubSonic.Song | None = reactive(SubSonic.Song()) is_playing: bool = False song_queue: list[SubSonic.Song] = reactive([]) user_queue: list[SubSonic.Song] = reactive([]) play_history: list[SubSonic.Song] = [] is_paused: bool = False place: int = 0 music_command_queue: multiprocessing.Queue check_playing_timer: Timer | None = None check_music_commands: Timer def on_mount(self) -> None: self.title = "SubPy" self.switch_mode('main') self.check_music_commands = self.set_interval(2/3, self._music_player_loop, pause=False) @work(exclusive=True) async def play_song(self, song: SubSonic.Song) -> None: self.currently_playing = song self.is_paused = False self.is_playing = True 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()) # If file was found in cache record scrobble asyncio.create_task(asyncio.to_thread(conn.scrobble, song.id, submission=False)) else: # Run the blocking stream call in a separate thread if BITRATE: r = await asyncio.to_thread(conn.stream, song.id, maxBitRate=BITRATE) else: 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 _music_player_loop(self) -> None: if self.music_command_queue.empty(): return command = self.music_command_queue.get_nowait() match command: case "play_pause": self.action_play_pause_song() case "next": self.action_skip() case "prev": self.action_rewind() case "vdown": self.action_volume_down() case "udown": self.action_volume_up() 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 and self.currently_playing: self.play_history.append(self.currently_playing) self.is_played = False next_song: SubSonic.Song | None = None if self.user_queue: next_song = self.user_queue.pop(0) elif self.song_queue: next_song = self.song_queue.pop(0) if next_song: self.play_song(next_song) if self.check_playing_timer: 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) next_song: SubSonic.Song | None = None if self.user_queue: next_song = self.user_queue.pop(0) elif self.song_queue: next_song = self.song_queue.pop(0) if next_song: self.play_song(next_song) 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 action_volume_down(self) -> None: pygame.mixer.music.set_volume(max(pygame.mixer.music.get_volume() - 0.05, 0.0)) def action_volume_up(self) -> None: pygame.mixer.music.set_volume(min(pygame.mixer.music.get_volume() + 0.05, 1.0)) def play_pause_song(self) -> None: if not self.currently_playing or not self.is_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() async def action_star_c_song(self) -> None: if self.currently_playing: await asyncio.create_task(asyncio.to_thread(conn.star, self.currently_playing.id)) self.notify(f'Favourite song "{self.currently_playing.title}".') def on_key(self, event: events.Key) -> None: logging.debug(event.key) if __name__ == '__main__': os.makedirs(".cache/art/", exist_ok=True) os.makedirs(".cache/audio/", exist_ok=True) os.makedirs(".cache/art/artists/", exist_ok=True) # Clear Log File if os.path.exists(LOG_FILE): os.remove(LOG_FILE) logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT) q = multiprocessing.Queue(-1) p = multiprocessing.Process(target=listen_media_keys, args=(q,)) p.start() app = TextualSubpyPlayer() app.music_command_queue = q app.run() p.terminate()