1294 lines
45 KiB
Python
1294 lines
45 KiB
Python
import multiprocessing
|
|
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'
|
|
|
|
# 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
|
|
IMAGE_MODE = config.getboolean('GENERAL', 'GUI')
|
|
|
|
import SubSonic
|
|
|
|
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
|
|
import pygame
|
|
import pynput
|
|
|
|
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, Button
|
|
from textual.widget import Widget
|
|
from textual.containers import Horizontal, Vertical, Container, Center
|
|
from textual.screen import Screen, ModalScreen
|
|
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:]
|
|
|
|
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
|
|
|
|
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)
|
|
|
|
@on(events.Click)
|
|
def clicked(self, event: events.Click) -> None:
|
|
app.install_screen(AlbumSongs(self.album.id), 'albumSongs')
|
|
app.push_screen('albumSongs')
|
|
|
|
def clear(self) -> None:
|
|
self.album = None
|
|
|
|
coverArt = self.query_one(CoverArtPixels)
|
|
coverArt.cover = None
|
|
|
|
label = self.query_one(Label)
|
|
label.update('')
|
|
|
|
class ArtistAlbumsGUI(Screen):
|
|
BINDINGS = [
|
|
('escape', 'die()', "Back"),
|
|
Binding('a', 'page_backward()', 'Prev. Page'),
|
|
Binding('d', 'page_forward()', 'Next Page')
|
|
]
|
|
|
|
artist: SubSonic.Artist
|
|
albums: list[SubSonic.Album]
|
|
page: int = 1
|
|
pages: int = 1
|
|
albums_displayed: int
|
|
|
|
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 Label(id='page-counter')
|
|
yield Footer()
|
|
|
|
def on_mount(self) -> None:
|
|
self.albums = conn.getArtistAlbums(self.artist.id)
|
|
album_panels = self.query(AlbumPanel)
|
|
|
|
artist_header = self.query_one(Label)
|
|
artist_header.update(self.artist.name)
|
|
|
|
self.pages = math.ceil(len(self.albums) / len(album_panels))
|
|
self.update_pager()
|
|
|
|
self.albums_displayed = 0
|
|
for album_panel in album_panels:
|
|
album_panel.set_album(self.albums[self.albums_displayed])
|
|
self.albums_displayed += 1
|
|
|
|
if self.albums_displayed >= len(self.albums):
|
|
return
|
|
|
|
def update_pager(self) -> None:
|
|
pager = self.query_one('#page-counter', Label)
|
|
pager.update(f'{self.page} ... {self.pages}')
|
|
|
|
def _clear_panels(self) -> None:
|
|
album_panels = self.query(AlbumPanel)
|
|
|
|
for panel in album_panels:
|
|
panel.clear()
|
|
|
|
def action_die(self) -> None:
|
|
app.pop_screen()
|
|
app.uninstall_screen("artistAlbums")
|
|
|
|
def action_page_forward(self) -> None:
|
|
album_panels = self.query(AlbumPanel)
|
|
|
|
if self.page + 1 > self.pages:
|
|
return
|
|
|
|
self.page += 1
|
|
self.update_pager()
|
|
self._clear_panels()
|
|
|
|
for album_panel in album_panels:
|
|
if self.albums_displayed >= len(self.albums):
|
|
return
|
|
|
|
album_panel.set_album(self.albums[self.albums_displayed])
|
|
self.albums_displayed += 1
|
|
|
|
def action_page_backward(self) -> None:
|
|
album_panels = self.query(AlbumPanel)
|
|
|
|
if self.page - 1 <= 0:
|
|
return
|
|
|
|
# count how many panels actually are displaying an album
|
|
real_displayed = 0
|
|
for album_panel in album_panels:
|
|
if album_panel.album:
|
|
real_displayed += 1
|
|
|
|
self.page -= 1
|
|
self.update_pager()
|
|
self._clear_panels()
|
|
|
|
self.albums_displayed -= len(album_panels) + real_displayed
|
|
for album_panel in album_panels:
|
|
if self.albums_displayed >= len(self.albums):
|
|
return
|
|
|
|
album_panel.set_album(self.albums[self.albums_displayed])
|
|
self.albums_displayed += 1
|
|
|
|
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 | None
|
|
|
|
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:
|
|
if not self.artist:
|
|
return
|
|
|
|
logging.debug(f'ArtistPanel {self.artist.name} was clicked.')
|
|
app.install_screen(ArtistAlbumsGUI(self.artist), 'artistAlbums')
|
|
app.push_screen('artistAlbums')
|
|
|
|
def clear(self) -> None:
|
|
self.artist = None
|
|
|
|
art: ArtistArt = self.query_one(ArtistArt)
|
|
art.art = None
|
|
|
|
label = self.query_one(Label)
|
|
label.update('')
|
|
|
|
class ArtistsGUI(Screen):
|
|
|
|
BINDINGS = [
|
|
Binding('a', "page_backward()", "Prev. Page"),
|
|
Binding('d', "page_forward()", "Next Page")
|
|
]
|
|
|
|
artists: list[SubSonic.Artist]
|
|
artists_displayed: int = 0
|
|
page: int = 1
|
|
pages: 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")
|
|
yield Input(placeholder="Search", id="artistSearchBar")
|
|
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 = 'Artists'
|
|
self.artists = conn.getArtists()
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
self.artists_displayed = 0
|
|
|
|
self.pages = math.ceil(len(self.artists)/len(artist_panels))
|
|
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 update_pager(self) -> None:
|
|
pager = self.query_one("#page-counter", Label)
|
|
pager.update(f"{self.page} ... {self.pages}")
|
|
|
|
def action_page_forward(self) -> None:
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
if self.page + 1 > self.pages:
|
|
return
|
|
|
|
self.page += 1
|
|
self.update_pager()
|
|
self._clear_panels()
|
|
|
|
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)
|
|
|
|
if self.page - 1 <= 0:
|
|
return
|
|
|
|
# Count artist panels with info in them
|
|
real_displayed = 0
|
|
for artist_panel in artist_panels:
|
|
if artist_panel.artist:
|
|
real_displayed += 1
|
|
|
|
self.page -= 1
|
|
self.update_pager()
|
|
self._clear_panels()
|
|
|
|
self.artists_displayed -= len(artist_panels) + real_displayed
|
|
for artist_panel in artist_panels:
|
|
artist_panel.set_artist(self.artists[self.artists_displayed])
|
|
self.artists_displayed += 1
|
|
|
|
def _clear_panels(self) -> None:
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
for artist_panel in artist_panels:
|
|
artist_panel.clear()
|
|
|
|
@on(Input.Submitted)
|
|
def search_submit(self, event: Input.Submitted) -> None:
|
|
search_value = event.value.lower()
|
|
logging.debug(search_value)
|
|
|
|
artist_panels = self.query(ArtistPanel)
|
|
|
|
self.artists_displayed = 0
|
|
|
|
filtered_artists = []
|
|
for artist in self.artists:
|
|
if search_value in artist.name.lower():
|
|
filtered_artists.append(artist)
|
|
|
|
self.page = 1
|
|
self.pages = math.ceil(len(filtered_artists) / len(artist_panels))
|
|
self.update_pager()
|
|
self._clear_panels()
|
|
|
|
for artist_panel in artist_panels:
|
|
if self.artists_displayed >= len(filtered_artists):
|
|
return
|
|
|
|
artist_panel.set_artist(filtered_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:]
|
|
|
|
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
|
|
|
|
class SongsScreen(Screen):
|
|
BINDINGS = [
|
|
|
|
]
|
|
|
|
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")
|
|
|
|
with Horizontal(id='songsScreen'):
|
|
yield Placeholder(id='songModeSelectorList')
|
|
yield Placeholder(id='songDatatable')
|
|
|
|
yield Footer()
|
|
|
|
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:]
|
|
|
|
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
|
|
|
|
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))
|
|
|
|
# TODO : https://textual.textualize.io/widgets/data_table/#labeled-rows
|
|
class QueueScreen(ModalScreen):
|
|
|
|
BINDINGS = [
|
|
('escape', 'back()', 'Go Back')
|
|
]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
with Container(id='queueModal'):
|
|
yield QueueTable(id='queue')
|
|
yield Footer()
|
|
|
|
def action_back(self) -> None:
|
|
app.pop_screen()
|
|
|
|
class CurrentPlace(Static):
|
|
place: reactive[int] = reactive(0)
|
|
|
|
def render(self) -> RenderResult:
|
|
place_in_seconds = self.place // 1000
|
|
place_min, place_sec = divmod(place_in_seconds, 60)
|
|
|
|
place_min = max(0, place_min)
|
|
place_sec = max(0, place_sec)
|
|
|
|
place_formatted = f"{place_min:02d}:{place_sec:02d}"
|
|
|
|
return place_formatted
|
|
|
|
class TotalDuration(Static):
|
|
duration: reactive[int] = reactive(0)
|
|
|
|
def render(self) -> RenderResult:
|
|
duration_min, duration_sec = divmod(self.duration, 60)
|
|
duration_formatted = f"{duration_min:02d}:{duration_sec:02d}"
|
|
|
|
return duration_formatted
|
|
|
|
class Player(Screen):
|
|
CSS_PATH = 'main.tcss'
|
|
BINDINGS = [
|
|
('q', 'view_queue()', 'View Queue')
|
|
]
|
|
|
|
def on_mount(self) -> None:
|
|
app.sub_title = "Player"
|
|
|
|
def update_song(self, song: SubSonic.Song) -> None:
|
|
if song.album and song.title and song.artist:
|
|
|
|
track_title_label: Label = self.query_one('#trackTitle', Label)
|
|
track_artist_album_label: Label = self.query_one('#trackAlbumArtist', Label)
|
|
|
|
track_title_label.update(song.title)
|
|
track_artist_album_label.update(f'{song.album} by {song.artist}')
|
|
|
|
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 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 Center(id='playerTrackTitleBox'):
|
|
yield Label(id='trackTitle', classes='playerTrackinfo')
|
|
with Center():
|
|
yield Label(id='trackAlbumArtist', classes='playerTrackinfo')
|
|
with Container(id='playerArtc'):
|
|
yield CoverArtPixels(classes='playerArt')
|
|
|
|
with Center():
|
|
with Horizontal(id='progressBarWrapper'):
|
|
yield CurrentPlace(classes='playerProgressBarTime')
|
|
yield ProgressBar(classes='playerProgressBar')
|
|
yield TotalDuration(classes='playerProgressBarTotalTime')
|
|
|
|
with Horizontal(id='playerButtons'):
|
|
yield Button('|<', id='prev_track', classes='playerButton')
|
|
yield Button('|>/||', id='play_pause_track', classes='playerButton')
|
|
yield Button('>|', id='next_track', classes='playerButton')
|
|
|
|
yield Footer()
|
|
|
|
@on(events.ScreenResume)
|
|
def screen_resumed(self, event: events.ScreenResume) -> None:
|
|
app.sub_title = "Player"
|
|
app.update_now_playing()
|
|
|
|
@on(Button.Pressed, "#prev_track")
|
|
def prev_track_pressed(self) -> None:
|
|
app.action_rewind()
|
|
|
|
@on(Button.Pressed, "#play_pause_track")
|
|
def play_pause_track_pressed(self) -> None:
|
|
app.action_play_pause_song()
|
|
|
|
@on(Button.Pressed, "#next_track")
|
|
def next_track_pressed(self) -> None:
|
|
app.action_skip()
|
|
|
|
def action_view_queue(self) -> None:
|
|
# app.install_screen(QueueScreen, 'queue')
|
|
app.push_screen(QueueScreen())
|
|
|
|
# Startup keyboard listener for media keys
|
|
def listen_media_keys(queue: multiprocessing.Queue):
|
|
|
|
media_log_file = 'media_keys.log'
|
|
logger = logging.getLogger('media_keys')
|
|
logger.setLevel(LOGGING_LEVEL)
|
|
formatter = logging.Formatter('[%(asctime)s] %(levelname)s in %(module)s: %(message)s')
|
|
handler = logging.FileHandler(media_log_file)
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
|
|
logger.info('media_key logger initialized')
|
|
|
|
try:
|
|
def on_press(key: pynput.keyboard.Key | pynput.keyboard.KeyCode | None) -> None:
|
|
logger.debug(f'Process Key: {key}')
|
|
match key:
|
|
case pynput.keyboard.Key.media_play_pause:
|
|
logger.info('Media play/pause key pressed')
|
|
queue.put('play_pause')
|
|
case pynput.keyboard.Key.media_next:
|
|
logger.info('Media next key pressed')
|
|
queue.put('next')
|
|
case pynput.keyboard.Key.media_previous:
|
|
logger.info('Media rewind pressed')
|
|
queue.put('prev')
|
|
case pynput.keyboard.Key.media_volume_down:
|
|
logger.info('Media volume down')
|
|
queue.put('vdown')
|
|
case pynput.keyboard.Key.media_volume_up:
|
|
logger.info('Media volume up')
|
|
queue.put('udown')
|
|
with pynput.keyboard.Listener(on_press=on_press) as listener:
|
|
logger.info('Media key listener started')
|
|
listener.join()
|
|
except Exception as e:
|
|
pass
|
|
logger.error(f'Media key listener failed: {e}')
|
|
|
|
class SubPy(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" : Player,
|
|
"artists" : ArtistsGUI,
|
|
"albums" : Albums,
|
|
"favourites" : SongsScreen,
|
|
"playlists" : Playlists
|
|
}
|
|
|
|
currently_playing: SubSonic.Song | None = reactive(SubSonic.Song())
|
|
is_playing: bool = False
|
|
song_queue: list[SubSonic.Song] = reactive([])
|
|
user_queue: list[SubSonic.Song] = reactive([])
|
|
play_history: list[SubSonic.Song] = []
|
|
is_paused: bool = False
|
|
place: int = 0
|
|
|
|
music_command_queue: multiprocessing.Queue
|
|
|
|
check_playing_timer: Timer | None = None
|
|
check_music_commands: Timer
|
|
|
|
def on_mount(self) -> None:
|
|
self.title = "SubPy"
|
|
self.switch_mode('main')
|
|
|
|
self.check_music_commands = self.set_interval(2/3, self._music_player_loop, pause=False)
|
|
|
|
@work(exclusive=True)
|
|
async def play_song(self, song: SubSonic.Song) -> None:
|
|
self.currently_playing = song
|
|
self.is_paused = False
|
|
self.is_playing = True
|
|
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_now_playing()
|
|
|
|
def _music_player_loop(self) -> None:
|
|
if self.music_command_queue.empty():
|
|
return
|
|
|
|
command = self.music_command_queue.get_nowait()
|
|
|
|
match command:
|
|
case "play_pause":
|
|
self.action_play_pause_song()
|
|
case "next":
|
|
self.action_skip()
|
|
case "prev":
|
|
self.action_rewind()
|
|
case "vdown":
|
|
self.action_volume_down()
|
|
case "udown":
|
|
self.action_volume_up()
|
|
|
|
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)
|
|
|
|
self.is_played = False
|
|
|
|
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)
|
|
|
|
def update_progress(self) -> None:
|
|
if isinstance(self.screen, Player) and self.currently_playing:
|
|
current_place_label = self.screen.query_one(CurrentPlace)
|
|
place = pygame.mixer.music.get_pos()
|
|
current_place_label.place = place
|
|
|
|
duration_label = self.screen.query_one(TotalDuration)
|
|
duration_label.duration = self.currently_playing.duration
|
|
|
|
progressbar: ProgressBar = self.screen.query_one(ProgressBar)
|
|
|
|
progress: int = round(place/1000)
|
|
|
|
progressbar.update(
|
|
total = self.currently_playing.duration,
|
|
progress=progress
|
|
)
|
|
|
|
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 or not self.is_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, Player) and self.currently_playing:
|
|
logging.debug("Updating CoverArtPixels widget")
|
|
self.screen.update_song(self.currently_playing)
|
|
|
|
def update_user_queue(self, song: SubSonic.Song) -> None:
|
|
self.user_queue.append(song)
|
|
|
|
def update_song_queue(self, song: SubSonic.Song) -> None:
|
|
self.song_queue.append(song)
|
|
|
|
async def action_star_c_song(self) -> None:
|
|
if self.currently_playing:
|
|
await asyncio.create_task(asyncio.to_thread(conn.star, self.currently_playing.id))
|
|
self.notify(f'Favourite song "{self.currently_playing.title}".')
|
|
|
|
def on_key(self, event: events.Key) -> None:
|
|
logging.debug(event.key)
|
|
|
|
if __name__ == '__main__':
|
|
os.makedirs(".cache/art/", exist_ok=True)
|
|
os.makedirs(".cache/audio/", exist_ok=True)
|
|
os.makedirs(".cache/art/artists/", exist_ok=True)
|
|
|
|
# Clear Log File
|
|
if os.path.exists(LOG_FILE):
|
|
os.remove(LOG_FILE)
|
|
|
|
logging.basicConfig(filename=LOG_FILE, level=LOGGING_LEVEL, format=LOG_FORMAT)
|
|
|
|
q = multiprocessing.Queue(-1)
|
|
p = multiprocessing.Process(target=listen_media_keys, args=(q,))
|
|
p.start()
|
|
|
|
app = SubPy()
|
|
app.music_command_queue = q
|
|
|
|
app.run()
|
|
p.terminate()
|