1030 lines
36 KiB
Python
1030 lines
36 KiB
Python
import configparser
|
|
import logging
|
|
import asyncio
|
|
import random
|
|
import math
|
|
import os
|
|
import re
|
|
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)
|
|
|
|
# Read config file
|
|
config = configparser.ConfigParser()
|
|
config.read('config.ini')
|
|
|
|
CACHE = config.getboolean('STREAMING', 'Cache')
|
|
BITRATE = config.get('STREAMING', 'Bitrate')
|
|
BITRATE = re.findall(r'[0-9]+', BITRATE)[0]
|
|
BITRATE = int(BITRATE) if BITRATE else BITRATE
|
|
|
|
os.makedirs(".cache/art/", exist_ok=True)
|
|
os.makedirs(".cache/audio/", exist_ok=True)
|
|
os.makedirs(".cache/art/artists/", exist_ok=True)
|
|
|
|
logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT)
|
|
|
|
import SubSonic
|
|
|
|
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
|
|
import pygame
|
|
|
|
from rich.text import Text
|
|
|
|
from textual import on, events, work
|
|
from textual.binding import Binding
|
|
from textual.reactive import reactive
|
|
from textual.app import App, ComposeResult, RenderResult
|
|
from textual.widgets import Footer, Header, OptionList, DataTable, ProgressBar, Label, Input, Placeholder, Static, Rule
|
|
from textual.widget import Widget
|
|
from textual.containers import Horizontal, Vertical, Container
|
|
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, ImageFile
|
|
|
|
ImageFile.LOAD_TRUNCATED_IMAGES = True
|
|
|
|
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'),
|
|
('m', 'shuffle_mix()', 'Shuffle Play')
|
|
]
|
|
|
|
selected_song_id: str | None = None
|
|
songs: list[SubSonic.Song]
|
|
|
|
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")
|
|
|
|
def action_shuffle_mix(self) -> None:
|
|
q = self.songs.copy()
|
|
random.shuffle(q)
|
|
app.play_song(q.pop(0))
|
|
app.song_queue = q
|
|
app.update_queue_table()
|
|
|
|
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 AlbumPanel(Widget, can_focus=True, can_focus_children=True):
|
|
|
|
album: SubSonic.Album
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield CoverArtPixels(classes="albumArt")
|
|
yield Label(classes="albumName")
|
|
|
|
def set_album(self, album: SubSonic.Album) -> None:
|
|
self.album = album
|
|
|
|
coverArt = self.query_one(CoverArtPixels)
|
|
coverArt.get_cover_art(self.album.id)
|
|
|
|
label = self.query_one(Label)
|
|
label.update(self.album.album)
|
|
|
|
class ArtistAlbums2(Screen):
|
|
BINDINGS = [
|
|
('escape', 'die()', "Back")
|
|
]
|
|
|
|
artist: SubSonic.Artist
|
|
albums: list[SubSonic.Album]
|
|
|
|
def __init__(self, artist: SubSonic.Artist):
|
|
self.artist = artist
|
|
super().__init__()
|
|
|
|
def compose(self) -> ComposeResult:
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
yield Label(classes='artistNameHeader')
|
|
yield Rule()
|
|
with Horizontal():
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
with Horizontal():
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield AlbumPanel(classes="albumPanel")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self.albums = conn.getArtistAlbums(self.artist.id)
|
|
|
|
artist_header = self.query_one(Label)
|
|
artist_header.update(self.artist.name)
|
|
|
|
albums_displayed = 0
|
|
album_panels = self.query(AlbumPanel)
|
|
for album_panel in album_panels:
|
|
album_panel.set_album(self.albums[albums_displayed])
|
|
albums_displayed += 1
|
|
|
|
if albums_displayed >= len(self.albums) - 1:
|
|
return
|
|
|
|
def action_die(self) -> None:
|
|
app.pop_screen()
|
|
app.uninstall_screen("artistAlbums")
|
|
|
|
class Artists(Screen):
|
|
|
|
artists: list[SubSonic.Artist]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon='')
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
with Vertical(id="artistContainer"):
|
|
yield Input(placeholder="Search")
|
|
yield OptionList(id="Artists")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
app.sub_title = 'Artists'
|
|
self.artists = conn.getArtists()
|
|
artistList = self.query_one(OptionList)
|
|
|
|
for artist in self.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')
|
|
|
|
@on(Input.Submitted)
|
|
def search_submit(self, event: Input.Submitted) -> None:
|
|
search_value = event.value.lower()
|
|
|
|
artist_list = self.query_one(OptionList)
|
|
|
|
artist_list.clear_options()
|
|
for artist in self.artists:
|
|
if search_value in artist.name.lower():
|
|
artist_list.add_option(Option(artist.name, artist.id))
|
|
|
|
class ArtistLabel(Label, can_focus=True):
|
|
|
|
def render(self) -> RenderResult:
|
|
if self.has_focus:
|
|
return Text(str(self.content), style="white on blue")
|
|
else:
|
|
return self.content
|
|
|
|
class ArtistArt(Static):
|
|
|
|
art = reactive[Image.Image | None](None)
|
|
|
|
def render(self) -> RenderResult:
|
|
if self.art:
|
|
width, height = self.size
|
|
|
|
height *= 2
|
|
|
|
if width > height:
|
|
width = height
|
|
elif height > width:
|
|
height = width
|
|
|
|
try:
|
|
return Pixels.from_image(self.art, resize=(width, height))
|
|
except Exception as e:
|
|
logging.error(f'Couldn\'t load image {e}')
|
|
|
|
return ''
|
|
|
|
@work(exclusive=True, thread=True)
|
|
def get_artist_image(self, artist_id: str) -> None:
|
|
artist_id = 'ar-' + artist_id
|
|
|
|
image = None
|
|
cache_path = f".cache/art/artists/{artist_id}.png"
|
|
|
|
if CACHE and os.path.exists(cache_path):
|
|
try:
|
|
self.art = Image.open(cache_path)
|
|
except Exception as e:
|
|
logging.error(f"Failed to load cached artist art {cache_path}: {e}")
|
|
|
|
if image is None:
|
|
try:
|
|
artist_art_data = conn.getCoverArtRaw(artist_id, size=300)
|
|
image = Image.open(io.BytesIO(artist_art_data))
|
|
if CACHE:
|
|
try:
|
|
image.save(cache_path, 'PNG')
|
|
except Exception as e:
|
|
logging.error(f"Failed to save artist art to cache {cache_path}: {e}")
|
|
self.art = image
|
|
except Exception as e:
|
|
logging.error(f"Failed to get artist art: {e}")
|
|
self.art = None
|
|
|
|
class ArtistPanel(Widget, can_focus=True, can_focus_children=True):
|
|
|
|
artist: SubSonic.Artist
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield ArtistArt(classes="artistArt")
|
|
yield ArtistLabel(classes="artistName")
|
|
|
|
def set_artist(self, artist: SubSonic.Artist) -> None:
|
|
self.artist = artist
|
|
|
|
art: ArtistArt = self.query_one(ArtistArt)
|
|
art.get_artist_image(self.artist.id)
|
|
|
|
label = self.query_one(Label)
|
|
label.update(self.artist.name)
|
|
|
|
@on(events.Click)
|
|
def clicked(self, event: events.Click) -> None:
|
|
logging.debug(f'ArtistPanel {self.artist.name} was clicked.')
|
|
app.install_screen(ArtistAlbums2(self.artist), 'artistAlbums')
|
|
app.push_screen('artistAlbums')
|
|
|
|
class Artists2(Screen):
|
|
|
|
BINDINGS = [
|
|
Binding('d', "page_forward()", "Next Page"),
|
|
Binding('a', "page_backward()", "Prev. Page")
|
|
]
|
|
|
|
artists: list[SubSonic.Artist]
|
|
artists_displayed: int = 0
|
|
page: int = 1
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon='')
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("* F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
with Horizontal():
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
with Horizontal():
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
with Horizontal():
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield ArtistPanel(classes="artistPanel")
|
|
yield Label(id="page-counter")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
app.sub_title = 'Artists2'
|
|
self.artists = conn.getArtists()
|
|
|
|
self.artists_displayed = 0
|
|
|
|
self.update_pager()
|
|
|
|
artist_panels = self.query(ArtistPanel)
|
|
for artist_panel in artist_panels:
|
|
if self.artists_displayed >= len(self.artists) - 1:
|
|
return
|
|
|
|
artist_panel.set_artist(self.artists[self.artists_displayed])
|
|
self.artists_displayed += 1
|
|
|
|
def update_pager(self) -> None:
|
|
pager = self.query_one("#page-counter", Label)
|
|
pager.update(f"{self.page} ... {math.ceil(len(self.artists)/15)}")
|
|
|
|
def action_page_forward(self) -> None:
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
self.page += 1
|
|
self.update_pager()
|
|
|
|
for artist_panel in artist_panels:
|
|
if self.artists_displayed >= len(self.artists) - 1:
|
|
return
|
|
|
|
artist_panel.set_artist(self.artists[self.artists_displayed])
|
|
self.artists_displayed += 1
|
|
|
|
def action_page_backward(self) -> None:
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
self.page -= 1
|
|
self.update_pager()
|
|
|
|
self.artists_displayed -= 30
|
|
for artist_panel in artist_panels:
|
|
artist_panel.set_artist(self.artists[self.artists_displayed])
|
|
self.artists_displayed += 1
|
|
|
|
class Albums(Screen):
|
|
|
|
albums: list[SubSonic.Album]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon='')
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("* F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
with Vertical(id="albumContainer"):
|
|
yield Input(placeholder="Search", id="albumSearchBar")
|
|
yield DataTable(id="albumList")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
app.sub_title = "Albums"
|
|
self.albums = conn.getAllAlbums()
|
|
albumTable = self.query_one(DataTable)
|
|
|
|
albumTable.cursor_type = 'row'
|
|
albumTable.add_columns("Title", "Artist", "Songs", "Plays", "Year", "Time")
|
|
|
|
for album in self.albums:
|
|
albumTable.add_row(album.title, album.artist, 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')
|
|
|
|
@on(Input.Submitted)
|
|
def search_submit(self, event: Input.Submitted) -> None:
|
|
search_value = event.value.lower()
|
|
|
|
album_table = self.query_one(DataTable)
|
|
|
|
album_table.clear()
|
|
for album in self.albums:
|
|
if search_value in album.title.lower() or search_value in album.artist.lower():
|
|
album_table.add_row(album.title, album.artist, album.songCount, album.playCount, album.year, album.duration, key=album.id)
|
|
|
|
class Playlists(Screen):
|
|
|
|
playlists: list[SubSonic.Playlist] | None = None
|
|
selected_playlist_id: str = None
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon="")
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("* F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
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'),
|
|
('m', 'shuffle_mix()', 'Shuffle Play')
|
|
]
|
|
|
|
selected_song_id: str | None = None
|
|
songs: list[SubSonic.Song]
|
|
|
|
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 i, song in enumerate(self.songs):
|
|
songTable.add_row(song.title, song.album, song.artist, song.playCount, song.duration, song.bitRate, key=str(i))
|
|
|
|
# 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 = self.songs[int(event.row_key.value)].id
|
|
|
|
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:
|
|
song_id: int = self._find_selected_song()
|
|
app.queue_song(self.songs[song_id])
|
|
|
|
# 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
|
|
|
|
def action_shuffle_mix(self) -> None:
|
|
q = self.songs.copy()
|
|
random.shuffle(q)
|
|
app.play_song(q.pop(0))
|
|
app.song_queue = q
|
|
app.update_queue_table()
|
|
|
|
class Favourites(Screen):
|
|
BINDINGS = [
|
|
('p', 'play()', 'Play Song'),
|
|
('q', 'queue()', 'Queue Song'),
|
|
('m', 'shuffle_mix()', 'Shuffle Play')
|
|
]
|
|
|
|
songs: list[SubSonic.Song]
|
|
selected_song_id: str | None = None
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon="")
|
|
with Horizontal(id="headerbs"):
|
|
yield Label(Text("F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("* F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
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:
|
|
song_index = self._find_selected_song()
|
|
app.queue_song(self.songs[song_index])
|
|
|
|
# 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
|
|
|
|
def action_shuffle_mix(self) -> None:
|
|
q = self.songs.copy()
|
|
random.shuffle(q)
|
|
app.play_song(q.pop(0))
|
|
app.song_queue = q
|
|
app.update_queue_table()
|
|
|
|
class CoverArtPixels(Container):
|
|
|
|
cover = reactive[Image.Image | None](None)
|
|
|
|
def render(self) -> RenderResult:
|
|
if self.cover:
|
|
width, height = self.size
|
|
|
|
height *= 2
|
|
|
|
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:
|
|
image = None
|
|
cache_path = f".cache/art/{cover_art_id}.png"
|
|
|
|
if CACHE and os.path.exists(cache_path):
|
|
try:
|
|
self.cover = Image.open(cache_path)
|
|
except Exception as e:
|
|
logging.error(f"Failed to load cached cover art {cache_path}: {e}")
|
|
|
|
if image is None:
|
|
try:
|
|
cover_art_data = conn.getCoverArtRaw(cover_art_id, size=300)
|
|
image = Image.open(io.BytesIO(cover_art_data))
|
|
if CACHE:
|
|
try:
|
|
image.save(cache_path, "PNG")
|
|
except Exception as e:
|
|
logging.error(f"Failed to save cover art to cache {cache_path}: {e}")
|
|
self.cover = image
|
|
except Exception as e:
|
|
logging.error(f"Failed to get cover art: {e}")
|
|
self.cover = None
|
|
|
|
class QueueTable(Widget):
|
|
|
|
BINDINGS = [
|
|
("p", "play_queue_song()", "Play"),
|
|
("f", "star_song()", "Star Song")
|
|
]
|
|
|
|
song_ids: list[str] = []
|
|
selected_row_key: int = 0
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield DataTable(id="songsqueue", cursor_type='row')
|
|
|
|
def on_mount(self) -> None:
|
|
table = self.query_one(DataTable)
|
|
table.add_columns("Title", "Album", "Artist", "Plays", "Duration")
|
|
self.update_queue()
|
|
|
|
def update_queue(self) -> None:
|
|
table = self.query_one(DataTable)
|
|
table.clear()
|
|
queue: list[SubSonic.Song] = app.user_queue.copy()
|
|
queue.extend(app.song_queue)
|
|
|
|
self.song_ids = []
|
|
for i, song in enumerate(queue):
|
|
table.add_row(song.title, song.album, song.artist, str(song.playCount), str(song.duration), key=str(i))
|
|
self.song_ids.append(song.id)
|
|
|
|
@on(DataTable.RowHighlighted)
|
|
def set_selected_row_key(self, e: DataTable.RowHighlighted) -> None:
|
|
if e.row_key.value:
|
|
self.selected_row_key = int(e.row_key.value)
|
|
|
|
def action_play_queue_song(self) -> None:
|
|
# Reconstruct the queue to get the selected song object
|
|
queue: list[SubSonic.Song] = app.user_queue.copy()
|
|
queue.extend(app.song_queue)
|
|
|
|
if self.selected_row_key >= len(queue):
|
|
return # Avoid index out of bounds
|
|
|
|
# Add currently playing song to history
|
|
if app.currently_playing:
|
|
app.play_history.append(app.currently_playing)
|
|
|
|
# Add all songs before the selected one to history
|
|
songs_to_skip = queue[:self.selected_row_key]
|
|
app.play_history.extend(songs_to_skip)
|
|
|
|
song_to_play = queue[self.selected_row_key]
|
|
|
|
# The new queue is everything after the selected song
|
|
new_queue = queue[self.selected_row_key + 1:]
|
|
|
|
# Play the song
|
|
app.play_song(song_to_play)
|
|
|
|
# The user queue is cleared, and the song queue is the rest of the songs
|
|
app.user_queue = []
|
|
app.song_queue = new_queue
|
|
|
|
# Update the visual queue table
|
|
self.update_queue()
|
|
|
|
async def action_star_song(self) -> None:
|
|
queue: list[SubSonic.Song] = app.user_queue.copy()
|
|
queue.extend(app.song_queue)
|
|
|
|
song = queue[self.selected_row_key]
|
|
|
|
asyncio.create_task(asyncio.to_thread(conn.star, song.id))
|
|
|
|
# Fine to be a static renderable because it's not big enough to need a scroll
|
|
class NowPlayingInfo(Container):
|
|
song = reactive[SubSonic.Song | None](None, layout=True)
|
|
|
|
# Place is from pygame.mixer.music.get_pos and is in ms
|
|
place: reactive[int] = reactive(0)
|
|
|
|
def render(self) -> RenderResult:
|
|
if self.song:
|
|
if self.song.title and self.song.artist:
|
|
|
|
# Convert seconds to MM:SS
|
|
duration_min, duration_sec = divmod(self.song.duration, 60)
|
|
duration_formatted = f"{duration_min:02d}:{duration_sec:02d}"
|
|
|
|
place_in_seconds = self.place // 1000
|
|
place_min, place_sec = divmod(place_in_seconds, 60)
|
|
place_formatted = f"{place_min:02d}:{place_sec:02d}"
|
|
|
|
s = f"Song: {self.song.title}\nAlbum: {self.song.album}\nArtist: {self.song.artist}\nProgress: {place_formatted} / {duration_formatted}\n"
|
|
|
|
return s
|
|
return "Nothing Playing"
|
|
|
|
class NowPlaying(Widget):
|
|
def compose(self) -> ComposeResult:
|
|
yield NowPlayingInfo(id="textinfo")
|
|
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)
|
|
|
|
def set_place(self, place: int) -> None:
|
|
info = self.query_one(NowPlayingInfo)
|
|
info.place = place
|
|
|
|
# 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(id="headerbs"):
|
|
yield Label(Text("* F1 ", style="bold orange").append_text(Text("Player", style="white")), classes="headerb")
|
|
yield Label(Text("F2 ", style="bold orange").append_text(Text("Artists", style="white")), classes="headerb")
|
|
yield Label(Text("F3 ", style="bold orange").append_text(Text("Albums", style="white")), classes="headerb")
|
|
yield Label(Text("F4 ", style="bold orange").append_text(Text("Favourites", style="white")), classes="headerb")
|
|
yield Label(Text("F5 ", style="bold orange").append_text(Text("Playlists", style="white")), classes="headerb")
|
|
with Horizontal():
|
|
with Vertical(classes="info", id='info'):
|
|
yield NowPlaying(id="now-playing")
|
|
# yield ProgressBar()
|
|
with Vertical(classes="column"):
|
|
yield Label("Queue")
|
|
yield QueueTable(id="queue")
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
app.sub_title = "Overview"
|
|
|
|
@on(events.ScreenResume)
|
|
def screen_resumed(self, event: events.ScreenResume) -> None:
|
|
app.sub_title = "Overview"
|
|
queue_table = self.query_one(QueueTable)
|
|
queue_table.update_queue()
|
|
app.update_now_playing()
|
|
|
|
class TextualSubpyPlayer(App):
|
|
BINDINGS = [
|
|
("r", "rewind()", "⏪"),
|
|
("space", "play_pause_song()", "⏯"),
|
|
("s", "skip()", "⏭"),
|
|
("d", "volume_down()", "🕩"),
|
|
("u", "volume_up()", "🕪"),
|
|
("F", "star_c_song()", "Star Current Song"),
|
|
Binding('f1', "switch_mode('main')", 'Overview', show=False),
|
|
Binding('f2', "switch_mode('artists')", "Artists", show=False),
|
|
Binding('f3', "switch_mode('albums')", "Albums", show=False),
|
|
Binding('f4', "switch_mode('favourites')", "Favourites", show=False),
|
|
Binding('f5', "switch_mode('playlists')", "Playlists", show=False)
|
|
]
|
|
|
|
MODES = {
|
|
"main" : MainView,
|
|
"artists" : Artists2,
|
|
"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')
|
|
|
|
@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}.")
|
|
|
|
audio_data = None
|
|
audio_cache_path = f'.cache/audio/{song.id}.ogg'
|
|
|
|
if CACHE and os.path.exists(audio_cache_path):
|
|
with open(audio_cache_path, 'rb') as file:
|
|
audio_data = io.BytesIO(file.read())
|
|
|
|
# If file was found in cache record scrobble
|
|
asyncio.create_task(asyncio.to_thread(conn.scrobble, song.id, submission=False))
|
|
else:
|
|
# Run the blocking stream call in a separate thread
|
|
if BITRATE:
|
|
r = await asyncio.to_thread(conn.stream, song.id, maxBitRate=BITRATE)
|
|
else:
|
|
r = await asyncio.to_thread(conn.stream, song.id)
|
|
audio_content = r.content
|
|
|
|
if CACHE:
|
|
# Also run the file writing in a thread
|
|
await asyncio.to_thread(self._cache_audio, audio_cache_path, audio_content)
|
|
|
|
audio_data = io.BytesIO(audio_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.progess_timer = self.set_interval(0.3, self.update_progress, pause=False)
|
|
|
|
self.update_queue_table()
|
|
self.update_now_playing()
|
|
|
|
# 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 _cache_audio(self, path: str, content: bytes) -> None:
|
|
"""Helper method to write audio to cache."""
|
|
with open(path, "wb") as file:
|
|
file.write(content)
|
|
|
|
def check_done_playing(self) -> None:
|
|
if not pygame.mixer.music.get_busy() and not self.is_paused and self.currently_playing:
|
|
self.play_history.append(self.currently_playing)
|
|
|
|
next_song: SubSonic.Song | None = None
|
|
if self.user_queue:
|
|
next_song = self.user_queue.pop(0)
|
|
elif self.song_queue:
|
|
next_song = self.song_queue.pop(0)
|
|
|
|
if next_song:
|
|
self.play_song(next_song)
|
|
|
|
if self.check_playing_timer:
|
|
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)
|
|
|
|
next_song: SubSonic.Song | None = None
|
|
if self.user_queue:
|
|
next_song = self.user_queue.pop(0)
|
|
elif self.song_queue:
|
|
next_song = self.song_queue.pop(0)
|
|
|
|
if next_song:
|
|
self.play_song(next_song)
|
|
|
|
def action_rewind(self) -> None:
|
|
if not self.play_history:
|
|
return
|
|
|
|
if self.check_playing_timer:
|
|
self.check_playing_timer.stop()
|
|
|
|
if self.currently_playing:
|
|
self.song_queue.insert(0, self.currently_playing)
|
|
|
|
song_to_play = self.play_history.pop()
|
|
self.play_song(song_to_play)
|
|
self.update_queue_table()
|
|
|
|
def update_progress(self) -> None:
|
|
if isinstance(self.screen, MainView) and self.currently_playing:
|
|
now_playing = self.screen.query_one(NowPlaying)
|
|
now_playing.set_place(pygame.mixer.music.get_pos())
|
|
|
|
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 action_volume_down(self) -> None:
|
|
pygame.mixer.music.set_volume(max(pygame.mixer.music.get_volume() - 0.05, 0.0))
|
|
|
|
def action_volume_up(self) -> None:
|
|
pygame.mixer.music.set_volume(min(pygame.mixer.music.get_volume() + 0.05, 1.0))
|
|
|
|
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_now_playing(self) -> None:
|
|
if isinstance(self.screen, MainView) and self.currently_playing:
|
|
logging.debug("Updating now playing widget")
|
|
now_playing = self.screen.query_one(NowPlaying)
|
|
now_playing.update_song(self.currently_playing)
|
|
|
|
def update_queue_table(self) -> None:
|
|
if isinstance(self.screen, MainView):
|
|
logging.debug("Updating queue through app")
|
|
queue_table = self.screen.query_one(QueueTable)
|
|
|
|
queue_table.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()
|
|
|
|
async def action_star_c_song(self) -> None:
|
|
if self.currently_playing:
|
|
asyncio.create_task(asyncio.to_thread(conn.star, self.currently_playing.id))
|
|
|
|
def on_key(self, event: events.Key) -> None:
|
|
logging.debug(event.key)
|
|
|
|
if __name__ == '__main__':
|
|
app = TextualSubpyPlayer()
|
|
app.run()
|