subpy/textual-take2/app.py

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, Static
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 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 NowPlayingInfo(Container):
    song = reactive[SubSonic.Song | None](None, layout=True)

    def render(self) -> RenderResult:
        if self.song:
            return f"Playing: {self.song.title} by {self.song.artist}"
        else:
            return "Nothing Playing"

class NowPlaying(Widget):
    def compose(self) -> ComposeResult:
        yield NowPlayingInfo()
        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)



class MusicApp(App):
    CSS_PATH = 'app.tcss'

    BINDINGS = [
        # ("r", "rewind()", "⏪"),
        ("space", "play_pause_song()", "⏯")
        # ("s", "skip()", "⏭")
    ]

    currently_playing: bool = False
    is_paused: bool = False
    place: int = 0

    def compose(self) -> ComposeResult:
        yield Header()
        yield Horizontal(
            VerticalScroll(OptionList(id="playlists"), id="sidebar"),
            Vertical(
                DataTable(id="songs", cursor_type='row'),
                NowPlaying(id="now-playing"),
                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:
                # Doesn't work with duplicated 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)
    async def play_song(self, song_id: str) -> None:
        try:
            song = conn.getSong(song_id)
            now_playing = self.query_one(NowPlaying)
            # self.call_from_thread(NowPlaying.update_song, now_playing, song)
            now_playing.update_song(song)

            stream_data = conn.stream(song_id)
            pygame.mixer.music.load(io.BytesIO(stream_data.content))
            pygame.mixer.music.play()
            self.currently_playing = True
        except Exception as e:
            logging.error(f"Failed to play song {song_id}: {e}")

    def action_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

if __name__ == "__main__":
    app = MusicApp()
    app.run()