539 lines
17 KiB
Python
539 lines
17 KiB
Python
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 rich.table import Table
|
|
|
|
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 = '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
|
|
|
|
class CoverArtPixels(Container):
|
|
|
|
cover_art_id: str | None = None
|
|
|
|
def set_cover_art(self, cover_art_id: str) -> None:
|
|
self.cover_art_id = cover_art_id
|
|
|
|
def render(self) -> RenderResult:
|
|
|
|
if not self.cover_art_id:
|
|
return ""
|
|
|
|
# Make the size square
|
|
width, height = self.size
|
|
|
|
if width > height:
|
|
width = height
|
|
elif height > width:
|
|
height = width
|
|
|
|
image = Image.open(io.BytesIO(conn.getCoverArtRaw(self.cover_art_id)))
|
|
return Pixels().from_image(image, resize=(width, height))
|
|
|
|
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
|
|
|
|
class PlayingInfo(Widget):
|
|
|
|
def render(self) -> RenderResult:
|
|
table = Table()
|
|
|
|
table.add_column("")
|
|
table.add_column("")
|
|
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
|
|
|
|
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()
|
|
# with Vertical(classes="column"):
|
|
# yield Label("Queue")
|
|
yield VerticalScroll(QueueTable())
|
|
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)
|
|
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)
|
|
|
|
self.query_one(CoverArtPixels).set_cover_art(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 = [
|
|
("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)
|
|
|
|
# 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.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()
|