went back to working on the first verison, fixed queue updating on main screen + weird footer bug
This commit is contained in:
JISAUAY committed
1 parent
4c1f5bfe82
commit
a07ee3bd52
6 files changed
+1018
-993
No files matched your search
+540
@@ -0,0 +1,540 @@
|
||||
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
|
||||
if os.path.exists(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 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:
|
||||
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:]
|
||||
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:
|
||||
try:
|
||||
cover_art_data = conn.getCoverArtRaw(cover_art_id)
|
||||
image = Image.open(io.BytesIO(cover_art_data))
|
||||
self.cover = image
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to get cover art: {e}")
|
||||
self.cover = None
|
||||
|
||||
class QueueTable(Widget):
|
||||
|
||||
def render(self) -> RenderResult:
|
||||
table = Table()
|
||||
table.add_column("Title")
|
||||
table.add_column("Album")
|
||||
table.add_column("Artist")
|
||||
table.add_column("Plays")
|
||||
table.add_column("Duration")
|
||||
table.add_column("Bitrate")
|
||||
|
||||
queue: list[SubSonic.Song] = app.user_queue
|
||||
queue.extend(app.song_queue)
|
||||
|
||||
for song in queue:
|
||||
table.add_row(song.title, song.album, song.artist, str(song.playCount), str(song.duration), str(song.bitRate))
|
||||
|
||||
return table
|
||||
|
||||
# Fine to be a static renderable because it's not big enough to need a scroll
|
||||
class PlayingInfo(Widget):
|
||||
|
||||
def render(self) -> RenderResult:
|
||||
table = Table()
|
||||
|
||||
table.add_column("")
|
||||
table.add_column("", min_width=10)
|
||||
table.show_header = False
|
||||
|
||||
if app.currently_playing:
|
||||
song = app.currently_playing
|
||||
table.add_row("Title", song.title)
|
||||
table.add_row("Artist", song.artist)
|
||||
table.add_row("Album", song.album)
|
||||
table.add_row("Plays", str(song.playCount) if song.playCount else "0")
|
||||
table.add_row("Duration", str(song.duration))
|
||||
table.add_row("Track Number", str(song.trackNumber))
|
||||
table.add_row("Bit Rate", str(song.bitRate))
|
||||
|
||||
return table
|
||||
|
||||
# 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 Label("Playing")
|
||||
yield PlayingInfo()
|
||||
yield ProgressBar()
|
||||
yield CoverArtPixels(id="coverart")
|
||||
with Vertical(classes="column"):
|
||||
yield Label("Queue")
|
||||
yield DataTable(id="queue", cursor_type='row')
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
app.sub_title = "Overview"
|
||||
queueTable: DataTable = self.query_one(DataTable)
|
||||
queueTable.add_columns("Title", "Album", "Artist", "Plays", "Duration", "Bitrate")
|
||||
|
||||
@on(events.ScreenResume)
|
||||
def screen_resumed(self, event: events.ScreenResume) -> None:
|
||||
app.sub_title = "Overview"
|
||||
self.update_queue()
|
||||
|
||||
def update_queue(self) -> None:
|
||||
table = self.query_one(DataTable)
|
||||
table.clear()
|
||||
|
||||
queue: list[SubSonic.Song] = []
|
||||
queue.extend(app.user_queue)
|
||||
queue.extend(app.song_queue)
|
||||
|
||||
for song in queue:
|
||||
table.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate)
|
||||
|
||||
|
||||
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}.")
|
||||
|
||||
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)
|
||||
|
||||
self.update_queue_table()
|
||||
|
||||
# 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:
|
||||
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 self.check_playing_timer:
|
||||
self.check_playing_timer.stop()
|
||||
|
||||
self.play_song(self.play_history.pop())
|
||||
|
||||
# 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.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_queue_table(self) -> None:
|
||||
if isinstance(self.screen, MainView):
|
||||
logging.debug("Updating queue through app")
|
||||
self.screen.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()
|
||||
Reference in new issue
Block a user