1475 lines
55 KiB
Python
1475 lines
55 KiB
Python
"""The Textual application: layout, key bindings and orchestration.
|
|
|
|
The app is deliberately catalogue-agnostic: it holds one
|
|
:class:`~beefy_player.providers.base.RadioProvider` and asks it for stations,
|
|
song history and stream URLs. SomaFM is one such provider; see
|
|
:mod:`beefy_player.providers`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import queue
|
|
import time
|
|
from typing import ClassVar
|
|
|
|
import httpx
|
|
from textual import work
|
|
from textual.app import App, ComposeResult
|
|
from textual.binding import Binding
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.css.query import NoMatches
|
|
from textual.theme import Theme
|
|
from textual.widgets import Input
|
|
|
|
from . import artwork, bandcamp, config
|
|
from .mediasession import create_media_session, media_actions_queue
|
|
from .models import Broadcast, Station, parse_last_playing
|
|
from .palettes import PALETTE_ORDER
|
|
from .player import Player
|
|
from .providers import get_provider
|
|
from .providers.base import is_direct_stream, make_client
|
|
from .spectrum import SpectrumEngine
|
|
from .widgets.art import ArtPane
|
|
from .widgets.header import HeaderBar
|
|
from .widgets.help import HelpScreen
|
|
from .widgets.providers import ProviderScreen
|
|
from .widgets.schedule import SchedulePanel
|
|
from .widgets.search import SearchInput
|
|
from .widgets.spectrum import SpectrumPanel
|
|
from .widgets.stage import StagePanel
|
|
from .widgets.station_list import StationList
|
|
from .widgets.status import StatusBar
|
|
|
|
VOLUME_STEP = 5
|
|
|
|
#: How close the cursor may get to the end of a paginated list before the next
|
|
#: directory page is fetched.
|
|
LOAD_MORE_MARGIN = 20
|
|
|
|
# A theme made only of terminal defaults / ANSI palette entries. Textual's
|
|
# built-in themes paint an RGB background and map ANSI names to their own RGB
|
|
# palette, which breaks transparency and ignores the user's colours.
|
|
TERMINAL_THEME = Theme(
|
|
name="terminal",
|
|
primary="ansi_cyan",
|
|
secondary="ansi_blue",
|
|
accent="ansi_magenta",
|
|
warning="ansi_yellow",
|
|
error="ansi_red",
|
|
success="ansi_green",
|
|
# Leave foreground/background unset: Textual's ANSI theme generator then
|
|
# resolves them to "ansi_default" (SGR 49) instead of RGB black.
|
|
dark=True,
|
|
ansi=True,
|
|
text_alpha=1.0,
|
|
variables={
|
|
"ansi-background": "ansi_default",
|
|
"ansi-foreground": "ansi_default",
|
|
},
|
|
)
|
|
|
|
|
|
class BeefyPlayerApp(App[None]):
|
|
CSS_PATH = "theme.tcss"
|
|
ENABLE_COMMAND_PALETTE = False
|
|
TITLE = "Beefy Player"
|
|
SUB_TITLE = "terminal internet radio"
|
|
|
|
BINDINGS: ClassVar[list[Binding]] = [
|
|
Binding("up,k", "cursor_up", "up", show=False),
|
|
Binding("down,j", "cursor_down", "down", show=False),
|
|
Binding("g", "cursor_top", "top", show=False),
|
|
Binding("shift+g", "cursor_bottom", "bottom", show=False),
|
|
Binding("pageup", "page_up", "page up", show=False),
|
|
Binding("pagedown", "page_down", "page down", show=False),
|
|
Binding("enter", "tune", "tune in"),
|
|
Binding("space", "toggle_play", "play/pause"),
|
|
Binding("s", "stop", "stop", show=False),
|
|
Binding("right,bracketright", "next_station", "next", show=False),
|
|
Binding("left,bracketleft", "prev_station", "previous", show=False),
|
|
Binding("plus", "volume_up", "vol +", show=False),
|
|
Binding("equals_sign", "volume_up", "vol +", show=False),
|
|
Binding("minus", "volume_down", "vol -", show=False),
|
|
Binding("underscore", "volume_down", "vol -", show=False),
|
|
Binding("m", "mute", "mute", show=False),
|
|
Binding("f", "favorite", "favourite"),
|
|
Binding("slash", "search", "filter"),
|
|
Binding("escape", "escape", "close", show=False),
|
|
Binding("v", "visualizer", "visualiser"),
|
|
Binding("c", "palette", "palette"),
|
|
Binding("r", "refresh", "refresh"),
|
|
Binding("p", "provider", "catalogue"),
|
|
Binding("question_mark", "help", "help"),
|
|
Binding("q", "quit", "quit"),
|
|
]
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
provider: str | None = None,
|
|
autoplay: bool | None = None,
|
|
station_id: str | None = None,
|
|
volume: int | None = None,
|
|
media: bool = True,
|
|
mpris: bool | None = None,
|
|
offline: bool = False,
|
|
audio: bool = True,
|
|
spectrum_source: str = "auto",
|
|
) -> None:
|
|
# ansi_color=True stops Textual from converting ANSI palette entries
|
|
# to RGB; the terminal then chooses the actual colours.
|
|
super().__init__(ansi_color=True)
|
|
self.register_theme(TERMINAL_THEME)
|
|
self.theme = TERMINAL_THEME.name
|
|
self.config = config.Config.load()
|
|
#: The station catalogue in use (see :mod:`beefy_player.providers`).
|
|
self.provider = get_provider(provider or self.config.provider)
|
|
# One-off command line overrides are deliberately *not* written back
|
|
# into the config file (only an explicit --volume is remembered).
|
|
self.autoplay = self.config.autoplay if autoplay is None else autoplay
|
|
# ``mpris`` is the historical keyword; desktop media integration is
|
|
# MPRIS on Linux and Now Playing (via mpv) on macOS.
|
|
if mpris is not None:
|
|
media = mpris
|
|
self.media_enabled = self.config.mpris and media
|
|
if volume is not None:
|
|
self.config.volume = max(0, min(100, int(volume)))
|
|
|
|
self.start_station_id = station_id or self.config.last_station_for(
|
|
self.provider.id
|
|
)
|
|
self.offline = offline
|
|
self.audio_enabled = audio
|
|
|
|
self.stations: list[Station] = []
|
|
self.view: list[Station] = []
|
|
#: The current catalogue's broadcast schedule, when it has one.
|
|
self.schedule: list[Broadcast] = []
|
|
self.schedule_loaded = False
|
|
self.index = 0
|
|
self.query = ""
|
|
#: Server-side directory results for ``query``; ``None`` means the
|
|
#: view is filtered locally from :attr:`stations`.
|
|
self.search_results: list[Station] | None = None
|
|
self._search_results_query = ""
|
|
#: A server-side search is in flight; the list shows a searching hint.
|
|
self.search_pending = False
|
|
#: Stations reached through a search or a later directory page. They
|
|
#: are kept when a refresh only fetches the first page again.
|
|
self._discovered_station_ids: set[str] = set()
|
|
#: Directory paging: next offset, whether there are more pages, and
|
|
#: whether a page request is already in flight.
|
|
self._directory_offset = 0
|
|
self._directory_exhausted = True
|
|
self._loading_more = False
|
|
self.current: Station | None = None
|
|
self.playing_state = "stopped"
|
|
self.volume = self.config.volume
|
|
self.visualizer = self.config.visualizer
|
|
#: Colour palette name (see :mod:`beefy_player.palettes`).
|
|
self.palette = self.config.palette
|
|
self.tick_count = 0
|
|
self.notice = ""
|
|
|
|
self._notice_until = 0.0
|
|
self._volume_before_mute = self.volume or 70
|
|
self._last_history_fetch = 0.0
|
|
self._history_refresh_seconds = 25.0
|
|
self._schedule_refresh_seconds = 300.0
|
|
self._last_schedule_fetch = 0.0
|
|
self._stream_url = ""
|
|
self.artwork_url = ""
|
|
self.artwork_bytes: bytes | None = None
|
|
self._art_track_key = ""
|
|
self._bandcamp_key = ""
|
|
self._bandcamp_urls: dict[str, str] = {}
|
|
self._http: httpx.AsyncClient | None = None
|
|
self._closed = False
|
|
self._tick_timer = None
|
|
|
|
self.player = Player(media_keys=self.media_enabled)
|
|
self.spectrum = SpectrumEngine(source_mode=spectrum_source)
|
|
self.media_actions: queue.Queue[str] = media_actions_queue()
|
|
self.media = create_media_session(
|
|
self.media_actions,
|
|
enabled=self.media_enabled,
|
|
# No audio engine means nothing to publish from.
|
|
backend=Player.detect_backend() if audio else None,
|
|
)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# layout
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield HeaderBar()
|
|
with Horizontal(id="main"):
|
|
with Vertical(id="left"):
|
|
yield SearchInput()
|
|
yield StationList()
|
|
yield SchedulePanel()
|
|
yield StagePanel()
|
|
yield SpectrumPanel(self.spectrum)
|
|
yield StatusBar()
|
|
|
|
def on_mount(self) -> None:
|
|
config.ensure_dirs()
|
|
|
|
if self.audio_enabled:
|
|
self.player.start()
|
|
self.player.set_volume(self.volume)
|
|
self.spectrum.start()
|
|
self.media.start()
|
|
|
|
self._http = make_client(self.provider.user_agent)
|
|
self._apply_catalogue_chrome()
|
|
|
|
stations = self.provider.load_cached_stations()
|
|
cached = bool(stations)
|
|
if not stations:
|
|
stations = self.provider.fallback_stations()
|
|
self._set_stations(stations)
|
|
self._reset_directory_paging(stations, has_more=cached)
|
|
self._select_initial_station()
|
|
|
|
if self.current is not None:
|
|
cached = self.provider.load_cached_history(self.current.id)
|
|
if cached:
|
|
self.current.history = cached
|
|
self.current.now_playing = cached[0]
|
|
self._update_media()
|
|
|
|
self._tick_timer = self.set_interval(0.05, self._tick)
|
|
|
|
if not self.offline:
|
|
self._refresh_stations()
|
|
if self.current is not None:
|
|
self._fetch_history(self.current.id)
|
|
if self.provider.has_schedule:
|
|
# Offline still shows the cached schedule — the worker decides.
|
|
self._refresh_schedule()
|
|
|
|
self.set_focus(None)
|
|
|
|
if self.autoplay and self.current is not None and self.audio_enabled:
|
|
self._tune(self.current)
|
|
elif not self.audio_enabled:
|
|
self._set_notice("headless mode — audio disabled", 6)
|
|
else:
|
|
self._set_notice("press enter to tune in", 5)
|
|
|
|
def on_unmount(self) -> None:
|
|
self._cleanup()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# state helpers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _set_notice(self, message: str, seconds: float = 4.0) -> None:
|
|
self.notice = message
|
|
self._notice_until = time.monotonic() + seconds
|
|
|
|
def _save_config(self) -> None:
|
|
self.config.volume = self.volume
|
|
if self.current is not None:
|
|
self.config.last_station = self.current.id
|
|
self.config.last_stations[self.provider.id] = self.current.id
|
|
self.config.visualizer = self.visualizer
|
|
self.config.save()
|
|
|
|
def _find_station(self, station_id: str) -> Station | None:
|
|
"""Find a station by id: on air, in the catalogue, or in the search."""
|
|
if self.current is not None and self.current.id == station_id:
|
|
return self.current
|
|
for station in self.stations:
|
|
if station.id == station_id:
|
|
return station
|
|
for station in self.search_results or ():
|
|
if station.id == station_id:
|
|
return station
|
|
return None
|
|
|
|
def _set_stations(self, stations: list[Station]) -> None:
|
|
if not stations:
|
|
return
|
|
stations = list(stations)
|
|
known = {station.id for station in stations}
|
|
for station in self.stations:
|
|
if (
|
|
station.id in self._discovered_station_ids
|
|
and station.id not in known
|
|
):
|
|
# Keep stations found through search or a later page when a
|
|
# refresh only knows the first page.
|
|
stations.append(station)
|
|
known.add(station.id)
|
|
existing = {station.id: station for station in self.stations}
|
|
if self.current is not None:
|
|
existing[self.current.id] = self.current
|
|
for station in stations:
|
|
previous = existing.get(station.id)
|
|
if previous is not None:
|
|
station.history = previous.history
|
|
station.now_playing = previous.now_playing
|
|
station.favorite = previous.favorite
|
|
station.favorite = self.config.is_favorite(station.id)
|
|
if station.now_playing is None and station.last_playing:
|
|
station.now_playing = parse_last_playing(station.last_playing)
|
|
|
|
self.stations = stations
|
|
wanted = self.current.id if self.current is not None else self.start_station_id
|
|
self.current = next((s for s in stations if s.id == wanted), None)
|
|
if self.current is None:
|
|
self.current = stations[0]
|
|
self._rebuild_view(keep_id=wanted)
|
|
self.query_one(StationList).refresh()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self._update_artwork()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
def _reset_directory_paging(
|
|
self, stations: list[Station], *, has_more: bool = True
|
|
) -> None:
|
|
self._loading_more = False
|
|
if not self.provider.paginated or not has_more:
|
|
self._directory_offset = 0
|
|
self._directory_exhausted = True
|
|
return
|
|
# The cached list is in catalogue order, so continue after it.
|
|
self._directory_offset = len(stations)
|
|
self._directory_exhausted = False
|
|
|
|
def _append_stations(self, stations: list[Station]) -> int:
|
|
"""Add a page to the catalogue, preserving existing station state."""
|
|
known = {station.id for station in self.stations}
|
|
added = 0
|
|
for station in stations:
|
|
if station.id in known:
|
|
continue
|
|
station.favorite = self.config.is_favorite(station.id)
|
|
if station.now_playing is None and station.last_playing:
|
|
station.now_playing = parse_last_playing(station.last_playing)
|
|
self.stations.append(station)
|
|
known.add(station.id)
|
|
added += 1
|
|
return added
|
|
|
|
def _select_initial_station(self) -> None:
|
|
if not self.view:
|
|
return
|
|
target = self.start_station_id
|
|
for index, station in enumerate(self.view):
|
|
if station.id == target:
|
|
self.index = index
|
|
self.current = station
|
|
break
|
|
else:
|
|
self.index = 0
|
|
self.current = self.view[0]
|
|
self.config.last_station = self.current.id
|
|
self.query_one(StationList).refresh()
|
|
|
|
def _select_station_id(self, station_id: str) -> None:
|
|
self._rebuild_view(keep_id=station_id)
|
|
for index, station in enumerate(self.view):
|
|
if station.id == station_id:
|
|
self.index = index
|
|
break
|
|
|
|
def _rebuild_view(self, keep_id: str | None = None) -> None:
|
|
query = self.query.strip()
|
|
terms = query.casefold().split()
|
|
|
|
def matches(station: Station) -> bool:
|
|
if not terms:
|
|
return True
|
|
haystack = (
|
|
f"{station.title} {station.genre} {station.description} "
|
|
f"{station.dj} {station.id}"
|
|
).casefold()
|
|
return all(term in haystack for term in terms)
|
|
|
|
if keep_id is None and self.view and 0 <= self.index < len(self.view):
|
|
keep_id = self.view[self.index].id
|
|
|
|
if (
|
|
terms
|
|
and self.search_results is not None
|
|
and self._search_results_query == query
|
|
):
|
|
# The directory matched fields that are not all part of a Station's
|
|
# local search text; trust its answer instead of filtering again.
|
|
view = list(self.search_results)
|
|
else:
|
|
view = [station for station in self.stations if matches(station)]
|
|
if self.provider.preserve_order:
|
|
# Stable sort: keep the catalogue's popularity order within the
|
|
# favourites and everything else.
|
|
view.sort(key=lambda station: 0 if station.favorite else 1)
|
|
else:
|
|
view.sort(
|
|
key=lambda station: (
|
|
0 if station.favorite else 1,
|
|
station.title.casefold(),
|
|
)
|
|
)
|
|
self.view = view
|
|
|
|
if keep_id is not None:
|
|
for index, station in enumerate(view):
|
|
if station.id == keep_id:
|
|
self.index = index
|
|
break
|
|
else:
|
|
self.index = min(self.index, max(0, len(view) - 1))
|
|
else:
|
|
self.index = min(self.index, max(0, len(view) - 1))
|
|
|
|
if self.current is None and view:
|
|
self.current = view[self.index]
|
|
|
|
def _set_search_results(self, query: str, stations: list[Station]) -> None:
|
|
"""Show a server-side directory search in place of the loaded list."""
|
|
known = {station.id: station for station in self.stations}
|
|
if self.current is not None:
|
|
known[self.current.id] = self.current
|
|
results: list[Station] = []
|
|
for station in stations:
|
|
previous = known.get(station.id)
|
|
if previous is not None:
|
|
# Reuse the catalogue's object so favourite/history changes
|
|
# made in the search results show up in the regular list too.
|
|
station = previous
|
|
else:
|
|
station.favorite = self.config.is_favorite(station.id)
|
|
if station.now_playing is None and station.last_playing:
|
|
station.now_playing = parse_last_playing(station.last_playing)
|
|
results.append(station)
|
|
|
|
self.search_results = results
|
|
self._search_results_query = query
|
|
self.search_pending = False
|
|
self._rebuild_view(
|
|
keep_id=self.current.id if self.current is not None else None
|
|
)
|
|
if self.current is None or not any(
|
|
station.id == self.current.id for station in self.view
|
|
):
|
|
self.index = 0
|
|
with contextlib.suppress(NoMatches):
|
|
pane = self.query_one(StationList)
|
|
pane.set_filter(self.query)
|
|
pane.refresh()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# the catalogue's chrome
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _apply_catalogue_chrome(self) -> None:
|
|
"""Point the chrome at the catalogue: hero height and the left pane."""
|
|
provider = self.provider
|
|
hero_rows = len(getattr(provider, "hero", ()) or ())
|
|
# Give the hero its rows, but never more than a quarter of the screen:
|
|
# on a short terminal the spectrum and the stage matter more.
|
|
screen_height = getattr(self, "size", None)
|
|
cap = max(4, (screen_height.height // 4) if screen_height else 10)
|
|
with contextlib.suppress(NoMatches):
|
|
header = self.query_one(HeaderBar)
|
|
header.styles.height = min(hero_rows + 1, cap) if hero_rows else 4
|
|
header.refresh()
|
|
self.query_one(StationList).display = not provider.has_schedule
|
|
self.query_one(SchedulePanel).display = provider.has_schedule
|
|
if provider.has_schedule:
|
|
self._reset_filter()
|
|
|
|
def _reset_filter(self) -> None:
|
|
"""Forget the station filter and put the search box away."""
|
|
self.query = ""
|
|
self.search_results = None
|
|
self._search_results_query = ""
|
|
self.search_pending = False
|
|
search = self.query_one(SearchInput)
|
|
search.value = ""
|
|
search.remove_class("visible")
|
|
search.disabled = True
|
|
self.query_one(StationList).set_filter("")
|
|
|
|
def on_resize(self, event) -> None:
|
|
self._apply_catalogue_chrome()
|
|
|
|
def _refresh_pane(self) -> None:
|
|
with contextlib.suppress(NoMatches):
|
|
if self.provider.has_schedule:
|
|
self.query_one(SchedulePanel).refresh()
|
|
else:
|
|
self.query_one(StationList).refresh()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# playback
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _tune(self, station: Station) -> None:
|
|
self.current = station
|
|
self.start_station_id = station.id
|
|
if not any(existing.id == station.id for existing in self.stations):
|
|
# A directory search can reach a station outside the loaded pages;
|
|
# keep it in the catalogue so clearing the filter still shows it.
|
|
self.stations.append(station)
|
|
# Remember it so a refresh that only fetches the first page keeps it.
|
|
self._discovered_station_ids.add(station.id)
|
|
self.playing_state = "loading"
|
|
self._set_notice(f"tuning in to {station.title}…", 5)
|
|
self._save_config()
|
|
self._update_media()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
if not self.audio_enabled:
|
|
self.playing_state = "stopped"
|
|
self._set_notice("audio disabled (--no-audio)", 5)
|
|
return
|
|
|
|
# A tune gets a clean analyser: forget any previous capture failure and
|
|
# clear the old station's bars before the new stream starts.
|
|
self.spectrum.reset()
|
|
self._resolve_and_play(station.id)
|
|
self._fetch_history(station.id)
|
|
|
|
@work(exclusive=True, group="stream", exit_on_error=False)
|
|
async def _resolve_and_play(self, station_id: str) -> None:
|
|
station = self._find_station(station_id)
|
|
if station is None or self.current is None or self.current.id != station_id:
|
|
return
|
|
|
|
if self.offline or self._http is None:
|
|
urls = self.provider.fallback_urls(station)
|
|
else:
|
|
try:
|
|
urls = await self.provider.stream_candidates(self._http, station)
|
|
except Exception as exc:
|
|
self.playing_state = "error"
|
|
self._set_notice(f"could not resolve stream: {exc}", 8)
|
|
self._update_media()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
return
|
|
|
|
if self.current is None or self.current.id != station_id:
|
|
return
|
|
if not urls:
|
|
self.playing_state = "error"
|
|
self._set_notice("no stream mirrors available", 8)
|
|
self._update_media()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
return
|
|
|
|
direct = next((url for url in urls if is_direct_stream(url)), None)
|
|
self._stream_url = direct or urls[0]
|
|
if direct is not None:
|
|
self.spectrum.set_source(direct)
|
|
self.player.set_volume(self.volume)
|
|
self.player.play(urls)
|
|
|
|
@work(exclusive=True, group="history", exit_on_error=False)
|
|
async def _fetch_history(self, station_id: str) -> None:
|
|
self._last_history_fetch = time.monotonic()
|
|
station = self._find_station(station_id)
|
|
if station is None or self._http is None:
|
|
return
|
|
try:
|
|
songs = await self.provider.fetch_history(self._http, station_id)
|
|
except Exception:
|
|
songs = self.provider.load_cached_history(station_id)
|
|
if not songs:
|
|
return
|
|
|
|
previous = station.now_playing
|
|
station.history = songs
|
|
station.now_playing = songs[0]
|
|
if previous is not None and previous.key() == songs[0].key():
|
|
# History refreshes replace the Song object even when the track has
|
|
# not changed. Keep the asynchronously discovered Bandcamp link.
|
|
songs[0].bandcamp_url = previous.bandcamp_url
|
|
changed = previous is None or previous.key() != songs[0].key()
|
|
if self.current is not None and self.current.id == station_id:
|
|
if changed:
|
|
self._set_notice(
|
|
f"♪ {songs[0].display_artist} — {songs[0].display_title}", 5
|
|
)
|
|
self._update_media()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
def _maybe_refresh_history(self) -> None:
|
|
if self.offline or self.current is None:
|
|
return
|
|
if self.playing_state not in ("playing", "buffering", "loading"):
|
|
return
|
|
if time.monotonic() - self._last_history_fetch >= self._history_refresh_seconds:
|
|
self._fetch_history(self.current.id)
|
|
|
|
def _update_media(self) -> None:
|
|
station = self.current
|
|
if station is None:
|
|
return
|
|
song = station.now_playing
|
|
self.title = (
|
|
f"{self.provider.name} — {song.display_artist} · {song.display_title}"
|
|
if song is not None
|
|
else self.provider.name
|
|
)
|
|
if song is not None:
|
|
self.media.set_track(
|
|
station=station.title,
|
|
title=song.display_title,
|
|
artist=song.display_artist,
|
|
album=song.album,
|
|
art_url=song.art_url or station.image,
|
|
stream_url=self._stream_url,
|
|
track_key=f"{station.id}|{song.key()}",
|
|
)
|
|
else:
|
|
self.media.set_track(
|
|
station=station.title,
|
|
title=station.last_playing or station.title,
|
|
artist=station.dj or self.provider.name,
|
|
album=station.description,
|
|
art_url=station.image,
|
|
stream_url=self._stream_url,
|
|
track_key=station.id,
|
|
)
|
|
self.media.set_status(self._media_status())
|
|
self.media.set_volume(self.volume / 100.0)
|
|
self._update_bandcamp()
|
|
self._update_artwork()
|
|
|
|
def _update_bandcamp(self) -> None:
|
|
"""Find a Bandcamp track link once per now-playing track."""
|
|
station = self.current
|
|
if station is None:
|
|
return
|
|
song = station.now_playing
|
|
if song is None or not (song.artist.strip() or song.title.strip()):
|
|
return
|
|
if not getattr(self.provider, "bandcamp", True):
|
|
# Not a Bandcamp catalogue: never look the track up, and do not
|
|
# leave a stale link behind from the previous station.
|
|
song.bandcamp_url = ""
|
|
return
|
|
|
|
track_key = f"{station.id}|{song.key()}"
|
|
cached_url = self._bandcamp_urls.get(track_key)
|
|
if cached_url:
|
|
self._bandcamp_key = track_key
|
|
if song.bandcamp_url != cached_url:
|
|
song.bandcamp_url = cached_url
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(StagePanel).refresh_stage()
|
|
return
|
|
|
|
if self.offline or self._http is None:
|
|
return
|
|
if track_key == self._bandcamp_key:
|
|
return
|
|
self._bandcamp_key = track_key
|
|
self._fetch_bandcamp(
|
|
station.id, song.key(), song.artist, song.album, song.title
|
|
)
|
|
|
|
@work(exclusive=True, group="bandcamp", exit_on_error=False)
|
|
async def _fetch_bandcamp(
|
|
self,
|
|
station_id: str,
|
|
song_key: str,
|
|
artist: str,
|
|
album: str,
|
|
title: str,
|
|
) -> None:
|
|
if self._http is None or self.offline:
|
|
return
|
|
url = await bandcamp.find_track(self._http, artist, album, title)
|
|
if not url:
|
|
return
|
|
|
|
self._bandcamp_urls[f"{station_id}|{song_key}"] = url
|
|
station = self._find_station(station_id)
|
|
if station is None or self.current is None or self.current.id != station_id:
|
|
return
|
|
song = station.now_playing
|
|
if song is None or song.key() != song_key:
|
|
return
|
|
song.bandcamp_url = url
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(StagePanel).refresh_stage()
|
|
|
|
def _refresh_art(self) -> None:
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(StagePanel).refresh_art()
|
|
|
|
def _update_artwork(self) -> None:
|
|
"""Pick the album art / station logo and fetch it when needed."""
|
|
station = self.current
|
|
if station is None:
|
|
changed = bool(self.artwork_url or self.artwork_bytes)
|
|
self.artwork_url = ""
|
|
self.artwork_bytes = None
|
|
self._art_track_key = ""
|
|
if changed:
|
|
self._refresh_art()
|
|
return
|
|
|
|
song = station.now_playing
|
|
track_key = f"{station.id}|{song.key() if song else ''}"
|
|
track_changed = track_key != self._art_track_key
|
|
self._art_track_key = track_key
|
|
|
|
url = (
|
|
song.art_url
|
|
if song is not None and song.art_url
|
|
else station.image
|
|
) or ""
|
|
|
|
if not artwork.supported() or not url:
|
|
changed = bool(self.artwork_url or self.artwork_bytes)
|
|
self.artwork_url = ""
|
|
self.artwork_bytes = None
|
|
if track_changed or changed:
|
|
self._refresh_art()
|
|
return
|
|
|
|
if url != self.artwork_url:
|
|
self.artwork_url = url
|
|
self.artwork_bytes = artwork.load_cached(url)
|
|
self._refresh_art()
|
|
if self.artwork_bytes is None and not self.offline:
|
|
self._fetch_artwork(url)
|
|
elif track_changed:
|
|
# The image can stay, but the ANSI fallback changes per track.
|
|
self._refresh_art()
|
|
|
|
@work(exclusive=True, group="artwork", exit_on_error=False)
|
|
async def _fetch_artwork(self, url: str) -> None:
|
|
if self._http is None or not artwork.supported():
|
|
return
|
|
data = await artwork.fetch(self._http, url)
|
|
if self.artwork_url != url:
|
|
return
|
|
self.artwork_bytes = data
|
|
if data is None:
|
|
self._set_notice("artwork unavailable — using mosaic", 5)
|
|
self._refresh_art()
|
|
|
|
def _media_status(self) -> str:
|
|
if self.playing_state == "paused":
|
|
return "Paused"
|
|
if self.playing_state in ("playing", "loading", "buffering"):
|
|
return "Playing"
|
|
return "Stopped"
|
|
|
|
def _set_volume(self, value: int) -> None:
|
|
value = max(0, min(100, int(value)))
|
|
if value == self.volume:
|
|
return
|
|
self.volume = value
|
|
self.config.volume = value
|
|
if value > 0:
|
|
self._volume_before_mute = value
|
|
self.player.set_volume(value)
|
|
self.media.set_volume(value / 100.0)
|
|
self._save_config()
|
|
self.query_one(StatusBar).refresh()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# event loops
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _tick(self) -> None:
|
|
if self._closed:
|
|
return
|
|
try:
|
|
self._tick_body()
|
|
except NoMatches:
|
|
# A timer can fire in the tiny window while the screen is being
|
|
# unmounted; nothing to update any more.
|
|
return
|
|
|
|
def _tick_body(self) -> None:
|
|
self.tick_count += 1
|
|
self._drain_player()
|
|
self._drain_media()
|
|
|
|
if self.notice and time.monotonic() >= self._notice_until:
|
|
self.notice = ""
|
|
self.query_one(StatusBar).refresh()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
frame = self.spectrum.snapshot()
|
|
self.query_one(SpectrumPanel).sync(self.visualizer, frame.label)
|
|
|
|
notice = self.spectrum.take_notice()
|
|
if notice:
|
|
self._set_notice(notice, 10)
|
|
|
|
if self.tick_count % 4 == 0:
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self.query_one(HeaderBar).refresh()
|
|
if self.tick_count % 12 == 0:
|
|
self.query_one(StatusBar).refresh()
|
|
self._refresh_pane()
|
|
self._maybe_refresh_history()
|
|
self._maybe_refresh_schedule()
|
|
|
|
def _drain_player(self) -> None:
|
|
while True:
|
|
try:
|
|
event = self.player.events.get_nowait()
|
|
except queue.Empty:
|
|
return
|
|
|
|
if event.kind == "volume":
|
|
if event.volume is not None and event.volume != self.volume:
|
|
self.volume = event.volume
|
|
self.config.volume = event.volume
|
|
self.media.set_volume(self.volume / 100.0)
|
|
self._save_config()
|
|
continue
|
|
if event.kind == "notice":
|
|
if event.detail:
|
|
self._set_notice(event.detail, 6)
|
|
continue
|
|
if event.kind == "media":
|
|
# Media keys / Control Center on macOS, routed via mpv.
|
|
if event.detail:
|
|
self.media_actions.put(event.detail)
|
|
continue
|
|
|
|
self.playing_state = event.state or self.playing_state
|
|
if event.url:
|
|
self._stream_url = event.url
|
|
|
|
if event.state in ("playing", "loading", "buffering"):
|
|
self.media.set_status("Playing")
|
|
if event.url:
|
|
self.spectrum.set_source(event.url)
|
|
elif event.state == "paused":
|
|
self.media.set_status("Paused")
|
|
self.spectrum.set_source(None)
|
|
elif event.state in ("stopped", "error"):
|
|
self.media.set_status("Stopped")
|
|
self.spectrum.set_source(None)
|
|
|
|
if event.detail:
|
|
self._set_notice(event.detail, 6)
|
|
self.query_one(StagePanel).refresh_stage()
|
|
|
|
def _drain_media(self) -> None:
|
|
while True:
|
|
try:
|
|
action = self.media_actions.get_nowait()
|
|
except queue.Empty:
|
|
return
|
|
if action == "play":
|
|
if self.playing_state == "paused":
|
|
self.player.resume()
|
|
elif self.current is not None and self.playing_state in ("stopped", "error"):
|
|
self._tune(self.current)
|
|
elif self.current is None:
|
|
self.action_tune()
|
|
elif action == "pause":
|
|
self.player.pause()
|
|
elif action == "toggle":
|
|
self.action_toggle_play()
|
|
elif action == "stop":
|
|
self.action_stop()
|
|
elif action == "next":
|
|
self.action_next_station()
|
|
elif action == "prev":
|
|
self.action_prev_station()
|
|
elif action == "quit":
|
|
self.action_quit()
|
|
elif action.startswith("volume:"):
|
|
try:
|
|
value = int(action.split(":", 1)[1])
|
|
except ValueError:
|
|
continue
|
|
self._set_volume(value)
|
|
|
|
def _maybe_refresh_schedule(self) -> None:
|
|
if not self.provider.has_schedule or self.offline:
|
|
return
|
|
if (
|
|
time.monotonic() - self._last_schedule_fetch
|
|
>= self._schedule_refresh_seconds
|
|
):
|
|
self._refresh_schedule()
|
|
|
|
@work(exclusive=True, group="schedule", exit_on_error=False)
|
|
async def _refresh_schedule(self) -> None:
|
|
if not self.provider.has_schedule:
|
|
return
|
|
self._last_schedule_fetch = time.monotonic()
|
|
broadcasts: list[Broadcast] = []
|
|
if self.offline or self._http is None:
|
|
broadcasts = self.provider.load_cached_schedule()
|
|
else:
|
|
try:
|
|
broadcasts = await self.provider.fetch_schedule(self._http)
|
|
except Exception:
|
|
broadcasts = self.provider.load_cached_schedule()
|
|
self.schedule = broadcasts
|
|
self.schedule_loaded = True
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(SchedulePanel).refresh()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# workers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@work(exclusive=True, group="stations-page", exit_on_error=False)
|
|
async def _load_more_stations(self) -> None:
|
|
try:
|
|
if (
|
|
not self.provider.paginated
|
|
or self._directory_exhausted
|
|
or self.offline
|
|
or self._http is None
|
|
):
|
|
return
|
|
page = await self.provider.fetch_station_page(
|
|
self._http, self._directory_offset
|
|
)
|
|
except Exception:
|
|
self._set_notice("could not load more stations", 4)
|
|
return
|
|
finally:
|
|
self._loading_more = False
|
|
|
|
self._directory_offset = page.next_offset or 0
|
|
self._directory_exhausted = page.next_offset is None
|
|
if not self._append_stations(page.stations):
|
|
return
|
|
|
|
keep_id = (
|
|
self.view[self.index].id
|
|
if self.view and 0 <= self.index < len(self.view)
|
|
else None
|
|
)
|
|
self._rebuild_view(keep_id=keep_id)
|
|
with contextlib.suppress(NoMatches):
|
|
pane = self.query_one(StationList)
|
|
pane.set_filter(self.query)
|
|
pane.refresh()
|
|
|
|
def _maybe_load_more_stations(self) -> None:
|
|
"""Queue the next directory page when the cursor nears the end."""
|
|
if (
|
|
not self.provider.paginated
|
|
or self._loading_more
|
|
or self._directory_exhausted
|
|
or self.offline
|
|
or not self.view
|
|
or self.query.strip()
|
|
):
|
|
return
|
|
if self.index < len(self.view) - LOAD_MORE_MARGIN:
|
|
return
|
|
self._loading_more = True
|
|
self._set_notice("loading more stations…", 3)
|
|
self._load_more_stations()
|
|
|
|
@work(exclusive=True, group="search", exit_on_error=False)
|
|
async def _search_directory(self, query: str) -> None:
|
|
"""Debounced server-side search, for catalogues that support it."""
|
|
text = query.strip()
|
|
if (
|
|
not text
|
|
or not self.provider.server_search
|
|
or self.offline
|
|
or self._http is None
|
|
):
|
|
self.search_pending = False
|
|
return
|
|
# A new keystroke starts a new exclusive worker and cancels this sleep.
|
|
await asyncio.sleep(0.25)
|
|
if text != self.query.strip():
|
|
return
|
|
try:
|
|
stations = await self.provider.search_stations(self._http, text)
|
|
except Exception:
|
|
if text == self.query.strip():
|
|
self.search_pending = False
|
|
self._set_notice(
|
|
"directory search failed — filtering the loaded stations", 4
|
|
)
|
|
self._refresh_pane()
|
|
return
|
|
if text != self.query.strip():
|
|
return
|
|
self._set_search_results(text, stations)
|
|
|
|
async def _load_favorite_stations(self) -> None:
|
|
"""Fetch favourites that are not on the currently loaded page."""
|
|
if self.offline or self._http is None or not self.provider.paginated:
|
|
return
|
|
known = {station.id for station in self.stations}
|
|
missing = [
|
|
station_id
|
|
for station_id in self.config.favorites
|
|
if station_id not in known
|
|
]
|
|
if not missing:
|
|
return
|
|
try:
|
|
stations = await self.provider.fetch_stations_by_ids(
|
|
self._http, missing
|
|
)
|
|
except Exception:
|
|
return
|
|
|
|
added = 0
|
|
for station in stations:
|
|
if any(existing.id == station.id for existing in self.stations):
|
|
continue
|
|
if not self.config.is_favorite(station.id):
|
|
continue
|
|
previous = next(
|
|
(
|
|
searched
|
|
for searched in (self.search_results or ())
|
|
if searched.id == station.id
|
|
),
|
|
None,
|
|
)
|
|
if previous is not None:
|
|
# A search result and a favourite lookup can race; keep one
|
|
# object so favourite/history changes are shared.
|
|
station = previous
|
|
station.favorite = True
|
|
self.stations.append(station)
|
|
self._discovered_station_ids.add(station.id)
|
|
added += 1
|
|
if not added:
|
|
return
|
|
|
|
keep_id = (
|
|
self.view[self.index].id
|
|
if self.view and 0 <= self.index < len(self.view)
|
|
else None
|
|
)
|
|
self._rebuild_view(keep_id=keep_id)
|
|
with contextlib.suppress(NoMatches):
|
|
pane = self.query_one(StationList)
|
|
pane.set_filter(self.query)
|
|
pane.refresh()
|
|
|
|
@work(exclusive=True, group="stations", exit_on_error=False)
|
|
async def _refresh_stations(self) -> None:
|
|
if self.offline or self._http is None:
|
|
return
|
|
try:
|
|
if self.provider.paginated:
|
|
page = await self.provider.fetch_station_page(self._http, 0)
|
|
stations = page.stations
|
|
next_offset = page.next_offset
|
|
else:
|
|
stations = await self.provider.fetch_stations(self._http)
|
|
next_offset = None
|
|
if not stations:
|
|
raise RuntimeError("the catalogue returned no stations")
|
|
except Exception:
|
|
if not self.stations:
|
|
self._set_stations(self.provider.fallback_stations())
|
|
self._set_notice("offline — using the built-in station list", 8)
|
|
else:
|
|
self._set_notice("station refresh failed — using the cached list", 6)
|
|
# The full directory may be unreachable while a favourite lookup
|
|
# still works; show those rather than hiding them until a restart.
|
|
await self._load_favorite_stations()
|
|
return
|
|
before = self.current.id if self.current is not None else None
|
|
self._set_stations(stations)
|
|
if before is not None:
|
|
self._select_station_id(before)
|
|
self._directory_offset = next_offset or 0
|
|
self._directory_exhausted = next_offset is None
|
|
self._loading_more = False
|
|
self._set_notice(
|
|
"1 station" if len(stations) == 1 else f"{len(stations)} stations", 3
|
|
)
|
|
self.query_one(StationList).refresh()
|
|
await self._load_favorite_stations()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# actions
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def action_cursor_up(self) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = max(0, self.index - 1)
|
|
self.query_one(StationList).refresh()
|
|
|
|
def action_cursor_down(self) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = min(len(self.view) - 1, self.index + 1)
|
|
self.query_one(StationList).refresh()
|
|
self._maybe_load_more_stations()
|
|
|
|
def _page(self, delta: int) -> None:
|
|
if not self.view:
|
|
return
|
|
page = max(1, self.query_one(StationList).size.height - 1)
|
|
self.index = max(0, min(len(self.view) - 1, self.index + delta * page))
|
|
self.query_one(StationList).refresh()
|
|
self._maybe_load_more_stations()
|
|
|
|
def action_page_up(self) -> None:
|
|
self._page(-1)
|
|
|
|
def action_page_down(self) -> None:
|
|
self._page(1)
|
|
|
|
def action_cursor_top(self) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = 0
|
|
self.query_one(StationList).refresh()
|
|
|
|
def action_cursor_bottom(self) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = len(self.view) - 1
|
|
self.query_one(StationList).refresh()
|
|
self._maybe_load_more_stations()
|
|
|
|
def action_tune(self) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = max(0, min(self.index, len(self.view) - 1))
|
|
self._tune(self.view[self.index])
|
|
|
|
def action_toggle_play(self) -> None:
|
|
if self.current is None:
|
|
self.action_tune()
|
|
return
|
|
if self.playing_state in ("stopped", "error"):
|
|
self._tune(self.current)
|
|
elif not self.audio_enabled:
|
|
self._set_notice("audio disabled (--no-audio)", 4)
|
|
else:
|
|
self.player.toggle_pause()
|
|
|
|
def action_stop(self) -> None:
|
|
self.player.stop()
|
|
self.spectrum.set_source(None)
|
|
self.playing_state = "stopped"
|
|
self.media.set_status("Stopped")
|
|
self._set_notice("stopped", 2)
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self.query_one(HeaderBar).refresh()
|
|
|
|
def _step_station(self, delta: int) -> None:
|
|
if not self.view:
|
|
return
|
|
self.index = (self.index + delta) % len(self.view)
|
|
self.query_one(StationList).refresh()
|
|
self._tune(self.view[self.index])
|
|
|
|
def action_next_station(self) -> None:
|
|
self._step_station(1)
|
|
|
|
def action_prev_station(self) -> None:
|
|
self._step_station(-1)
|
|
|
|
def action_volume_up(self) -> None:
|
|
self._set_volume(self.volume + VOLUME_STEP)
|
|
|
|
def action_volume_down(self) -> None:
|
|
self._set_volume(self.volume - VOLUME_STEP)
|
|
|
|
def action_mute(self) -> None:
|
|
if self.volume > 0:
|
|
self._volume_before_mute = self.volume
|
|
self._set_volume(0)
|
|
else:
|
|
self._set_volume(self._volume_before_mute or 70)
|
|
|
|
def action_favorite(self) -> None:
|
|
station = self.view[self.index] if self.view else self.current
|
|
if station is None:
|
|
return
|
|
added = self.config.toggle_favorite(station.id)
|
|
station.favorite = added
|
|
known = False
|
|
for existing in self.stations:
|
|
if existing.id == station.id:
|
|
# A search copy and the catalogue copy can briefly coexist;
|
|
# keep both in step so clearing the filter cannot resurrect
|
|
# the old favourite state.
|
|
existing.favorite = added
|
|
known = True
|
|
break
|
|
if added:
|
|
# A favourite reached through a directory search must join the
|
|
# catalogue, or clearing the filter would hide it again.
|
|
self._discovered_station_ids.add(station.id)
|
|
if not known:
|
|
self.stations.append(station)
|
|
self._save_config()
|
|
self._rebuild_view(keep_id=station.id)
|
|
self._set_notice(("★ added " if added else "☆ removed ") + station.title, 3)
|
|
self.query_one(StationList).refresh()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
|
|
def action_search(self) -> None:
|
|
if self.provider.has_schedule:
|
|
self._set_notice("this catalogue shows a schedule, not a station list", 3)
|
|
return
|
|
search = self.query_one(SearchInput)
|
|
search.disabled = False
|
|
search.add_class("visible")
|
|
search.focus()
|
|
with contextlib.suppress(Exception):
|
|
search.cursor_position = len(search.value)
|
|
|
|
def action_escape(self) -> None:
|
|
search = self.query_one(SearchInput)
|
|
if self.query and not search.has_class("visible"):
|
|
# The filter is already sticky: a second escape clears it.
|
|
self._close_search(clear=True)
|
|
else:
|
|
# Put the search box away but keep the filter on the station list.
|
|
self._close_search(clear=False)
|
|
|
|
def _close_search(self, clear: bool = False) -> None:
|
|
search = self.query_one(SearchInput)
|
|
search.remove_class("visible")
|
|
if clear:
|
|
self.query = ""
|
|
self.search_results = None
|
|
self._search_results_query = ""
|
|
self.search_pending = False
|
|
search.value = ""
|
|
search.disabled = True
|
|
self.set_focus(None)
|
|
self._rebuild_view()
|
|
self.query_one(StationList).set_filter(self.query)
|
|
self.query_one(StationList).refresh()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self._maybe_load_more_stations()
|
|
|
|
def action_visualizer(self) -> None:
|
|
modes = list(config.VISUALIZERS)
|
|
if self.visualizer in modes:
|
|
self.visualizer = modes[(modes.index(self.visualizer) + 1) % len(modes)]
|
|
else:
|
|
self.visualizer = "both"
|
|
self.config.visualizer = self.visualizer
|
|
self._save_config()
|
|
self._set_notice(f"visualiser: {self.visualizer}", 2)
|
|
self.query_one(SpectrumPanel).refresh()
|
|
|
|
def action_palette(self) -> None:
|
|
order = list(PALETTE_ORDER)
|
|
index = order.index(self.palette) if self.palette in order else 0
|
|
self.palette = order[(index + 1) % len(order)]
|
|
self.config.palette = self.palette
|
|
self._save_config()
|
|
self._set_notice(f"palette: {self.palette}", 2)
|
|
# Frames, spectrum, mosaic and wordmarks all read the palette live, so
|
|
# a repaint of every widget is all it takes. (``self.query`` is the
|
|
# station filter here, hence ``self.screen.query``.)
|
|
for widget in self.screen.query("*"):
|
|
with contextlib.suppress(Exception):
|
|
widget.refresh()
|
|
|
|
def action_refresh(self) -> None:
|
|
self._last_history_fetch = 0.0
|
|
self._bandcamp_key = ""
|
|
self._bandcamp_urls.clear()
|
|
with contextlib.suppress(Exception):
|
|
self.workers.cancel_group(self, "stations-page")
|
|
self._loading_more = False
|
|
self._set_notice("refreshing…", 2)
|
|
if self.offline:
|
|
self._set_notice("offline mode — nothing to refresh", 3)
|
|
return
|
|
self.artwork_url = ""
|
|
self._refresh_stations()
|
|
self._refresh_schedule()
|
|
if self.current is not None:
|
|
self._update_artwork()
|
|
self._fetch_history(self.current.id)
|
|
|
|
def action_provider(self) -> None:
|
|
self._suspend_artwork()
|
|
self.push_screen(ProviderScreen(self.provider.id), self._provider_chosen)
|
|
|
|
def _provider_chosen(self, provider_id: str | None) -> None:
|
|
self._resume_artwork()
|
|
if provider_id:
|
|
self._switch_catalogue(provider_id)
|
|
|
|
def _suspend_artwork(self) -> None:
|
|
"""Hide the inline artwork: terminal images would sit over a menu."""
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(ArtPane).suspend()
|
|
|
|
def _resume_artwork(self) -> None:
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(ArtPane).resume()
|
|
|
|
def _cancel_workers(self) -> None:
|
|
for group in (
|
|
"stream",
|
|
"history",
|
|
"search",
|
|
"stations",
|
|
"stations-page",
|
|
"schedule",
|
|
"bandcamp",
|
|
"artwork",
|
|
):
|
|
with contextlib.suppress(Exception):
|
|
self.workers.cancel_group(self, group)
|
|
|
|
def _switch_catalogue(self, provider_id: str) -> None:
|
|
"""Move to another station catalogue without restarting the app."""
|
|
provider = get_provider(provider_id)
|
|
if provider is self.provider:
|
|
return
|
|
|
|
self._cancel_workers()
|
|
with contextlib.suppress(Exception):
|
|
self.player.stop()
|
|
self.spectrum.set_source(None)
|
|
self.playing_state = "stopped"
|
|
self.media.set_status("Stopped")
|
|
|
|
old_client = self._http
|
|
self.provider = provider
|
|
self._http = make_client(provider.user_agent)
|
|
if old_client is not None:
|
|
with contextlib.suppress(RuntimeError):
|
|
asyncio.get_running_loop().create_task(old_client.aclose())
|
|
|
|
self.config.provider = provider.id
|
|
self.config.save()
|
|
|
|
self.stations = []
|
|
self.view = []
|
|
self.index = 0
|
|
self.search_results = None
|
|
self._search_results_query = ""
|
|
self.search_pending = False
|
|
self._discovered_station_ids.clear()
|
|
self.current = None
|
|
self.schedule = []
|
|
self.schedule_loaded = False
|
|
self._last_schedule_fetch = 0.0
|
|
self._last_history_fetch = 0.0
|
|
self._bandcamp_key = ""
|
|
self._bandcamp_urls.clear()
|
|
self._stream_url = ""
|
|
self._art_track_key = ""
|
|
self.artwork_url = ""
|
|
self.artwork_bytes = None
|
|
self.start_station_id = self.config.last_station_for(provider.id)
|
|
|
|
self._apply_catalogue_chrome()
|
|
with contextlib.suppress(NoMatches):
|
|
self._reset_filter()
|
|
stations = provider.load_cached_stations()
|
|
cached = bool(stations)
|
|
if not stations:
|
|
stations = provider.fallback_stations()
|
|
self._set_stations(stations)
|
|
self._reset_directory_paging(stations, has_more=cached)
|
|
self._select_initial_station()
|
|
self._update_media()
|
|
self._update_artwork()
|
|
self._refresh_art()
|
|
with contextlib.suppress(NoMatches):
|
|
self.query_one(HeaderBar).refresh()
|
|
self.query_one(StagePanel).refresh_stage()
|
|
self.query_one(StatusBar).refresh()
|
|
|
|
if self.audio_enabled and len(stations) == 1 and self.current is not None:
|
|
# A catalogue with one station has nothing to choose between:
|
|
# switching to it is the same as tuning it.
|
|
self._tune(self.current)
|
|
else:
|
|
self._set_notice(f"catalogue: {provider.name}", 3)
|
|
|
|
if not self.offline:
|
|
self._refresh_stations()
|
|
if self.current is not None:
|
|
self._fetch_history(self.current.id)
|
|
if provider.has_schedule:
|
|
self._refresh_schedule()
|
|
|
|
def action_help(self) -> None:
|
|
self._suspend_artwork()
|
|
self.push_screen(HelpScreen(), lambda _result: self._resume_artwork())
|
|
|
|
def action_quit(self) -> None:
|
|
self._cleanup()
|
|
self.exit()
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# input widget events
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def on_input_changed(self, event: Input.Changed) -> None:
|
|
if event.input.id != "search":
|
|
return
|
|
self.query = event.value
|
|
self.search_results = None
|
|
self._search_results_query = ""
|
|
self.search_pending = (
|
|
self.provider.server_search
|
|
and not self.offline
|
|
and bool(event.value.strip())
|
|
)
|
|
self._rebuild_view()
|
|
self.query_one(StationList).set_filter(self.query)
|
|
self.query_one(StationList).refresh()
|
|
if self.provider.server_search:
|
|
self._search_directory(event.value)
|
|
self._maybe_load_more_stations()
|
|
|
|
def on_input_submitted(self, event: Input.Submitted) -> None:
|
|
if event.input.id != "search":
|
|
return
|
|
station = self.view[self.index] if self.view else None
|
|
# Enter tunes, and leaves the filter in place exactly like the first
|
|
# escape does; a second escape returns to the regular station list.
|
|
self._close_search(clear=False)
|
|
if station is not None:
|
|
self._tune(station)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# shutdown
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _cleanup(self) -> None:
|
|
if self._closed:
|
|
return
|
|
self._closed = True
|
|
if self._tick_timer is not None:
|
|
with contextlib.suppress(Exception):
|
|
self._tick_timer.stop()
|
|
self._save_config()
|
|
with contextlib.suppress(Exception):
|
|
self.media.stop()
|
|
with contextlib.suppress(Exception):
|
|
self.spectrum.stop()
|
|
with contextlib.suppress(Exception):
|
|
self.player.shutdown()
|
|
if self._http is not None:
|
|
with contextlib.suppress(RuntimeError):
|
|
asyncio.get_running_loop().create_task(self._http.aclose())
|