738 lines
24 KiB
Python
738 lines
24 KiB
Python
import logging
|
|
import random
|
|
import os
|
|
import io
|
|
import asyncio
|
|
|
|
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)
|
|
|
|
CACHE = True
|
|
|
|
os.makedirs(".cache/art/", exist_ok=True)
|
|
os.makedirs(".cache/audio/", 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.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, Input
|
|
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
|
|
|
|
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 Artists(Screen):
|
|
|
|
artists: list[SubSonic.Artist]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon='')
|
|
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 Albums(Screen):
|
|
|
|
albums: list[SubSonic.Album]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(icon='')
|
|
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="")
|
|
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="")
|
|
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
|
|
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:
|
|
image = None
|
|
cache_path = f".cache/art/{cover_art_id}.png"
|
|
|
|
if CACHE and os.path.exists(cache_path):
|
|
try:
|
|
image = Image.open(cache_path)
|
|
self.cover = image
|
|
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
|
|
return
|
|
|
|
class QueueTable(Widget):
|
|
|
|
BINDINGS = [
|
|
("p", "play_queue_song()", "Play")
|
|
]
|
|
|
|
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()
|
|
|
|
# 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():
|
|
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()", "🕪"),
|
|
("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')
|
|
|
|
@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
|
|
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()
|
|
|
|
def on_key(self, event: events.Key) -> None:
|
|
logging.debug(event.key)
|
|
|
|
if __name__ == '__main__':
|
|
app = TextualSubpyPlayer()
|
|
app.run()
|