139 lines
4.5 KiB
Python
139 lines
4.5 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
|
|
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 NowPlaying(Widget):
|
|
song = reactive[SubSonic.Song | None](None, layout=True)
|
|
cover = reactive[Image.Image | None](None)
|
|
|
|
def render(self) -> RenderResult:
|
|
if not self.song:
|
|
return "Nothing playing"
|
|
|
|
if self.cover:
|
|
return Pixels.from_image(self.cover)
|
|
else:
|
|
return f"Playing: {self.song.title} by {self.song.artist}"
|
|
|
|
@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
|
|
|
|
def watch_song(self, song: SubSonic.Song | None) -> None:
|
|
if song and song.albumId:
|
|
album = conn.getAlbumSongs(song.albumId)
|
|
if album:
|
|
cover_id = album[0].coverArtId
|
|
if cover_id:
|
|
self.get_cover_art(cover_id)
|
|
|
|
|
|
class MusicApp(App):
|
|
def compose(self) -> ComposeResult:
|
|
yield Header()
|
|
yield Horizontal(
|
|
VerticalScroll(OptionList(id="playlists"), id="sidebar"),
|
|
Vertical(
|
|
NowPlaying(id="now-playing"),
|
|
DataTable(id="songs"),
|
|
id="main-panel"
|
|
)
|
|
)
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self.run_worker(self.get_playlists(), exclusive=True)
|
|
table = self.query_one(DataTable)
|
|
table.add_columns("Track", "Title", "Artist", "Album")
|
|
|
|
async def get_playlists(self) -> None:
|
|
try:
|
|
playlists = conn.getPlaylists()
|
|
option_list = self.query_one("#playlists", OptionList)
|
|
for playlist in playlists:
|
|
option_list.add_option(Option(playlist.name, id=playlist.id))
|
|
except Exception as e:
|
|
logging.error(f"Failed to get playlists: {e}")
|
|
|
|
@on(OptionList.OptionSelected)
|
|
async def on_playlist_selected(self, event: OptionList.OptionSelected) -> None:
|
|
self.run_worker(self.get_playlist_songs(event.option.id), exclusive=True)
|
|
|
|
async def get_playlist_songs(self, playlist_id: str) -> None:
|
|
try:
|
|
songs = conn.getPlaylist(playlist_id)
|
|
table = self.query_one(DataTable)
|
|
table.clear()
|
|
logging.debug([song.title for song in songs])
|
|
for song in songs:
|
|
table.add_row(song.trackNumber, song.title, song.artist, song.album, key=song.id)
|
|
except Exception as e:
|
|
logging.error(f"Failed to get playlist songs: {e}")
|
|
|
|
@on(DataTable.RowSelected)
|
|
def on_song_selected(self, event: DataTable.RowSelected) -> None:
|
|
song_id = event.row_key.value
|
|
if song_id:
|
|
self.play_song(song_id)
|
|
|
|
@work(exclusive=True, thread=True)
|
|
def play_song(self, song_id: str) -> None:
|
|
try:
|
|
song = conn.getSong(song_id)
|
|
now_playing = self.query_one(NowPlaying)
|
|
self.call_from_thread(setattr, now_playing, "song", song)
|
|
|
|
stream_data = conn.stream(song_id)
|
|
pygame.mixer.music.load(io.BytesIO(stream_data.content))
|
|
pygame.mixer.music.play()
|
|
except Exception as e:
|
|
logging.error(f"Failed to play song {song_id}: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
app = MusicApp()
|
|
app.run()
|
|
|