import asyncio import random import re from dataclasses import dataclass import pyperclip from textual import on from textual.widgets import DataTable, Input, Label, Button, RadioSet, RadioButton from textual.widget import Widget from textual.app import ComposeResult from textual.binding import Binding from textual.screen import ModalScreen from textual.containers import Container, Horizontal, Vertical import SubSonic def fix_duration(seconds: int) -> str: min, sec = divmod(seconds, 60) min = max(0, min) sec = max(0, sec) formatted = f"{min}:{sec:02d}" return formatted class RatingModal(ModalScreen[int]): """Modal for getting a numerical rating for an Album, Artist, or Song""" BINDINGS = [ ('escape', 'die', 'Close') ] thing_title: str selected_rating: int = 0 def __init__(self, thing_title: str): self.thing_title = thing_title super().__init__() def on_mount(self) -> None: label: Label = self.query_one(Label) label.update(f'Select a rating for {self.thing_title}') def compose(self) -> ComposeResult: with Vertical(): yield Label() with Horizontal(): with RadioSet(): yield RadioButton("0", value = True) yield RadioButton("1") yield RadioButton("2") yield RadioButton("3") yield RadioButton("4") yield RadioButton("5") with Horizontal(): yield Button("Submit") @on(Button.Pressed) def selection_submit(self, event: Button.Pressed) -> None: self.dismiss(self.selected_rating) @on(RadioSet.Changed) def selection_changed(self, event: RadioSet.Changed) -> None: self.selected_rating = event.radio_set.pressed_index def action_die(self) -> None: self.dismiss() class ShareLinkModal(ModalScreen): """Modal for copying a generated share link""" BINDINGS = [ ('escape', 'die', 'Close') ] link: str def __init__(self, link: str): self.link = link super().__init__() def on_mount(self) -> None: label: Label = self.query_one(Label) label.update(self.link) def compose(self) -> ComposeResult: with Horizontal(): yield Label() yield Button("Copy Link") @on(Button.Pressed) def copy_link(self) -> None: pyperclip.copy(self.link) def action_die(self) -> None: self.dismiss() class PlaylistSelectionModal(ModalScreen[str]): """Modal for selecting a playlist""" playlists: list[SubSonic.Playlist] = [] selected_playlist_id: str = "" BINDINGS = [ ('escape', 'die', 'Back') ] def compose(self) -> ComposeResult: with Container(): yield Label("Select a playlist") yield DataTable() def on_mount(self) -> None: self.playlists = self.app.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: self.dismiss(self.selected_playlist_id) def action_die(self) -> None: self.dismiss("") @dataclass class SongTableColumnConfig: track_number: bool = False title: bool = True album: bool = True artist: bool = True plays: bool = True duration: bool = True rating: bool = False bitrate: bool = False # Maps config field -> (column header, song attribute) COLUMN_MAPPING: list[tuple[str, str, str]] = [ ('track_number', '#', 'trackNumber'), ('title', 'Title', 'title'), ('album', 'Album', 'album'), ('artist', 'Artist', 'artist'), ('plays', 'Plays', 'playCount'), ('duration', 'Duration', 'duration'), ('rating', 'Rating', 'rating'), ('bitrate', 'Bitrate', 'bitRate') ] """ A custom DataTable designed to hold and display songs TODO @0x01fe : needs optional paging on init """ class SongTable(Widget): BINDINGS = [ ('p', 'play()', 'Play Song'), ('q', 'queue()', 'Queue Song'), ('m', 'shuffle()', 'Shuffle Play'), ('A', 'add_to_playlist()', 'Add to Playlist'), ('f', 'star_song()', 'Star Song'), ('S', 'share_song()', 'Share Song'), ('R', 'rate_song()', 'Rate Song'), Binding('/', 'filter()', 'Search', show=False) ] columns: SongTableColumnConfig songs: list[SubSonic.Song] = [] filter_active: bool = False selected_song_id: str | None = None max_col_width: int | None = None OVERFLOW_INDICATOR: str = "..." def __init__(self, columns: SongTableColumnConfig | None = SongTableColumnConfig(), songs: list[SubSonic.Song] | None = [], max_col_width: int | None = None, **kwargs): self.columns = columns self.songs = songs self.max_col_width = max_col_width super().__init__(**kwargs) def compose(self) -> ComposeResult: yield Input(placeholder="Search", id="filter") yield DataTable() def _format_cell(self, config_attr: str, value): """Format cell values for display in the DataTable.""" if config_attr == "duration": try: return fix_duration(int(value or 0)) except (TypeError, ValueError): return fix_duration(0) if self.max_col_width and type(value) == str and len(value) > self.max_col_width: value = value[:self.max_col_width - len(self.OVERFLOW_INDICATOR)] value += self.OVERFLOW_INDICATOR return value def _row_data_for_song(self, song: SubSonic.Song) -> list: """Build DataTable row values based on configured columns.""" row_data = [] for config_attr, _, song_attr in COLUMN_MAPPING: if getattr(self.columns, config_attr): value = getattr(song, song_attr) row_data.append(self._format_cell(config_attr, value)) return row_data def on_mount(self) -> None: dt: DataTable = self.query_one(DataTable) dt.focus() dt.cursor_type = "row" # Add columns based on config for config_attr, header, _ in COLUMN_MAPPING: if getattr(self.columns, config_attr): dt.add_column(header) # Add rows for existing songs for i, song in enumerate(self.songs): dt.add_row(*self._row_data_for_song(song), key=str(i)) def add_song(self, song: SubSonic.Song) -> None: self.songs.append(song) dt: DataTable = self.query_one(DataTable) dt.add_row(*self._row_data_for_song(song), key=str(len(self.songs) - 1)) @on(Input.Submitted) def search_submit(self, event: Input.Submitted) -> None: search_value = event.value.lower() dt: DataTable = self.query_one(DataTable) dt.clear() for i, song in enumerate(self.songs): if ( re.search(search_value, song.title.lower()) or re.search(search_value, song.artist.lower()) or re.search(search_value, song.album.lower()) ): dt.add_row(*self._row_data_for_song(song), key=str(i)) @on(DataTable.RowHighlighted) def row_highlighted(self, event: DataTable.RowHighlighted) -> None: if not self.songs: return self.selected_song_id = self.songs[int(event.row_key.value)].id def action_play(self) -> None: song_index: int = self._find_selected_song_index() self.app.play_song(self.songs[song_index]) self.app.song_queue = self.songs[song_index + 1:] def action_queue(self) -> None: song_index: int = self._find_selected_song_index() self.app.queue_song(self.songs[song_index]) def action_shuffle(self) -> None: q: list[SubSonic.Song] = self.songs.copy() random.shuffle(q) self.app.play_song(q.pop(0)) self.app.song_queue = q def action_add_to_playlist(self) -> None: def process_modal_selection(playlist_id: str) -> None: self.app.conn.updatePlaylist(playlist_id, songIdToAdd=[self.selected_song_id]) self.app.notify("Added song to playlist!") self.app.push_screen(PlaylistSelectionModal(), process_modal_selection) def action_star_song(self) -> None: song = self.songs[self._find_selected_song_index()] is_song_starred: bool = self.app.navidrome.isSongStarred(song.id) if not is_song_starred: asyncio.create_task(asyncio.to_thread(self.app.conn.star, self.selected_song_id)) self.app.notify(f'Starred song "{song.title}"') else: asyncio.create_task(asyncio.to_thread(self.app.conn.unstar, self.selected_song_id)) self.app.notify(f'Unstarred song "{song.title}"') def action_share_song(self) -> None: share_link: str = self.app.conn.createShare(self.selected_song_id) self.app.push_screen(ShareLinkModal(share_link)) def action_rate_song(self) -> None: song: SubSonic.Song = self.songs[self._find_selected_song_index()] def process_modal_selection(rating: int | None) -> None: if rating == None: return self.app.conn.setRating(self.selected_song_id, rating) if rating != 0: self.app.notify(f'Rated "{song.title}" a {rating}.') else: self.app.notify(f'Removed rating for "{song.title}".') self.app.push_screen(RatingModal(f'"{song.title}" By {song.artist}'), process_modal_selection) # Just makes the filter appear def action_filter(self) -> None: search_bar: Input = self.query_one(Input) if not self.filter_active: search_bar.display = 'block' else: search_bar.display = 'none' self.filter_active = not self.filter_active search_bar.focus() # Returns index of selected song def _find_selected_song_index(self) -> int: for i, song in enumerate(self.songs): if song.id == self.selected_song_id: return i