#!/usr/bin/env python3 """ GNOME Radio – objektorientert Python-implementasjon ===================================================== Bygget med GTK4 og PyGObject, i samme ånd som resten av Aamot Innovation sine GNOME-verktøy (GTK4/Python-stilen brukt i GNOME Noark5 Client og GNOME Photo Companion). Arkitektur: Station – dataklasse som representerer én radiostasjon StationLibrary – innlasting, lagring, søk og CRUD i stasjonslisten RadioPlayer – innkapsler GStreamer-avspilling (playbin) StationRow – én rad i listen, med favoritt-/rediger-/slett-meny StationEditDialog – modal dialog for å legge til/redigere en stasjon RadioWindow – GTK4-vinduet (View), lytter på RadioPlayer-signaler RadioApplication – Gtk.Application, kobler alt sammen Funksjoner: - Legg til, rediger og slett stasjoner direkte i grensesnittet - Filtrer på favoritter, i tillegg til fritekstsøk - Live sporinfo (låt/programtittel) fra strømmens metadata-tags - Mute/unmute som husker forrige volumnivå - Toast-varsler for feil, lagring og sletting (Adw.ToastOverlay) - Import/eksport av stasjonsliste som JSON - MPRIS2-integrasjon: medietaster og GNOME Shell sitt now-playing-widget Kjør med: python3 gnome_radio_oop.py Avhengigheter (Debian/Ubuntu): sudo apt install python3-gi gir1.2-gtk-4.0 gir1.2-gstreamer-1.0 \ gstreamer1.0-plugins-good gstreamer1.0-plugins-bad \ gstreamer1.0-plugins-ugly gstreamer1.0-libav """ from __future__ import annotations import json import logging from collections import deque from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from enum import Enum, auto from pathlib import Path from typing import Callable, Optional import gi gi.require_version("Gtk", "4.0") gi.require_version("Adw", "1") gi.require_version("Gst", "1.0") from gi.repository import Gtk, Adw, Gst, GLib, Gio, GObject # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") logger = logging.getLogger("gnome-radio") APP_ID = "com.aamotinnovation.GnomeRadio" CONFIG_DIR = Path.home() / ".config" / "gnome-radio" STATIONS_FILE = CONFIG_DIR / "stations.json" # --------------------------------------------------------------------------- # Domenemodell # --------------------------------------------------------------------------- @dataclass class Station: """Representerer én radiostasjon.""" name: str url: str genre: str = "Ukjent" country: str = "Norge" favorite: bool = False def to_dict(self) -> dict: return asdict(self) @staticmethod def from_dict(data: dict) -> "Station": return Station( name=data.get("name", "Ukjent stasjon"), url=data["url"], genre=data.get("genre", "Ukjent"), country=data.get("country", "Norge"), favorite=data.get("favorite", False), ) class PlaybackState(Enum): STOPPED = auto() BUFFERING = auto() PLAYING = auto() PAUSED = auto() ERROR = auto() # --------------------------------------------------------------------------- # Stasjonsbibliotek # --------------------------------------------------------------------------- class StationLibrary: """Holder styr på alle kjente stasjoner: innlasting, lagring, søk.""" DEFAULT_STATIONS = [ Station("NRK P1", "https://lyd.nrk.no/nrk_radio_p1_ostlandssendingen_mp3_h", genre="Allmennradio"), Station("NRK P2", "https://lyd.nrk.no/nrk_radio_p2_mp3_h", genre="Kultur"), Station("NRK P3", "https://lyd.nrk.no/nrk_radio_p3_mp3_h", genre="Musikk"), Station("Radio Norge", "https://http-live.sr.se/p3-mp3-192", genre="Musikk", country="Norge"), Station("NRK Klassisk", "https://lyd.nrk.no/nrk_radio_klassisk_mp3_h", genre="Klassisk"), ] def __init__(self, storage_path: Path = STATIONS_FILE) -> None: self._storage_path = storage_path self._stations: list[Station] = [] self.load() # -- persistens --------------------------------------------------- def load(self) -> None: if self._storage_path.exists(): try: raw = json.loads(self._storage_path.read_text(encoding="utf-8")) self._stations = [Station.from_dict(s) for s in raw] logger.info("Lastet %d stasjoner fra %s", len(self._stations), self._storage_path) return except (json.JSONDecodeError, KeyError, OSError) as exc: logger.warning("Kunne ikke lese stasjonsfil (%s), bruker standard", exc) self._stations = list(self.DEFAULT_STATIONS) def save(self) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) data = [s.to_dict() for s in self._stations] self._storage_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") # -- CRUD ----------------------------------------------------------- def add(self, station: Station) -> None: self._stations.append(station) self.save() def remove(self, station: Station) -> None: self._stations = [s for s in self._stations if s.url != station.url] self.save() def update(self, original: Station, updated: Station) -> None: """Erstatter `original` med `updated` på samme plass i listen, og bevarer favoritt-status hvis den nye versjonen ikke selv setter den.""" for i, s in enumerate(self._stations): if s.url == original.url: updated.favorite = original.favorite self._stations[i] = updated self.save() return # Fant den ikke (bør normalt ikke skje) – legg til i stedet. self.add(updated) def toggle_favorite(self, station: Station) -> None: station.favorite = not station.favorite self.save() def all(self) -> list[Station]: return list(self._stations) def favorites(self) -> list[Station]: return [s for s in self._stations if s.favorite] def search(self, query: str) -> list[Station]: q = query.strip().lower() if not q: return self.all() return [ s for s in self._stations if q in s.name.lower() or q in s.genre.lower() or q in s.country.lower() ] # -- import/eksport -------------------------------------------------- def export_to_path(self, path: Path) -> int: """Eksporterer hele stasjonslisten som en selvstendig JSON-fil, med et enkelt filformat-stempel slik at importen kan verifisere den.""" payload = { "format": "gnome-radio-stations", "format_version": 1, "exported": datetime.now(timezone.utc).isoformat(), "station_count": len(self._stations), "stations": [s.to_dict() for s in self._stations], } path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") return len(self._stations) def import_from_path(self, path: Path, replace: bool = False) -> int: """Importerer stasjoner fra en fil eksportert med export_to_path() (eller en enkel liste av stasjon-objekter). Slår som standard sammen med eksisterende liste (dedupliserer på URL) med mindre replace=True.""" raw = json.loads(path.read_text(encoding="utf-8")) if isinstance(raw, dict) and "stations" in raw: entries = raw["stations"] elif isinstance(raw, list): entries = raw else: raise ValueError("Ukjent filformat – fant ingen stasjonsliste") imported = [Station.from_dict(s) for s in entries] if replace: self._stations = imported else: existing_urls = {s.url for s in self._stations} for station in imported: if station.url not in existing_urls: self._stations.append(station) existing_urls.add(station.url) self.save() return len(imported) # --------------------------------------------------------------------------- # Avspiller (GStreamer-innkapsling) # --------------------------------------------------------------------------- class RadioPlayer(GObject.Object): """ Innkapsler en GStreamer 'playbin' for strømmeavspilling. Emitterer GObject-signaler slik at UI-laget (RadioWindow) kan reagere på tilstandsendringer uten å kjenne til GStreamer. """ __gsignals__ = { "state-changed": (GObject.SignalFlags.RUN_FIRST, None, (object,)), "station-changed": (GObject.SignalFlags.RUN_FIRST, None, (object,)), "track-changed": (GObject.SignalFlags.RUN_FIRST, None, (str,)), "error": (GObject.SignalFlags.RUN_FIRST, None, (str,)), } def __init__(self) -> None: super().__init__() Gst.init(None) self._playbin = Gst.ElementFactory.make("playbin", "player") if self._playbin is None: raise RuntimeError("Klarte ikke å opprette GStreamer 'playbin'-element") self._current_station: Optional[Station] = None self._state = PlaybackState.STOPPED self._volume = 1.0 self._muted = False self._pre_mute_volume = 1.0 bus = self._playbin.get_bus() bus.add_signal_watch() bus.connect("message::error", self._on_bus_error) bus.connect("message::buffering", self._on_bus_buffering) bus.connect("message::state-changed", self._on_bus_state_changed) bus.connect("message::tag", self._on_bus_tag) # -- offentlig API ---------------------------------------------------- @property def state(self) -> PlaybackState: return self._state @property def current_station(self) -> Optional[Station]: return self._current_station @property def volume(self) -> float: return self._volume @volume.setter def volume(self, value: float) -> None: self._volume = max(0.0, min(1.0, value)) self._muted = self._volume == 0.0 self._playbin.set_property("volume", self._volume) @property def is_muted(self) -> bool: return self._muted def toggle_mute(self) -> None: if self._muted: self._muted = False self.volume = self._pre_mute_volume or 1.0 else: self._pre_mute_volume = self._volume if self._volume > 0 else 1.0 self._muted = True self.volume = 0.0 self._muted = True # volume-setteren nullstiller flagget over, så sett det tilbake def play(self, station: Station) -> None: self._current_station = station self._playbin.set_state(Gst.State.NULL) self._playbin.set_property("uri", station.url) self._playbin.set_property("volume", self._volume) self._playbin.set_state(Gst.State.PLAYING) self._set_state(PlaybackState.BUFFERING) self.emit("station-changed", station) self.emit("track-changed", "") # nullstill forrige sporinfo def stop(self) -> None: self._playbin.set_state(Gst.State.NULL) self._set_state(PlaybackState.STOPPED) self.emit("track-changed", "") def toggle_pause(self) -> None: if self._state == PlaybackState.PLAYING: self._playbin.set_state(Gst.State.PAUSED) self._set_state(PlaybackState.PAUSED) elif self._state in (PlaybackState.PAUSED, PlaybackState.STOPPED): self._playbin.set_state(Gst.State.PLAYING) self._set_state(PlaybackState.PLAYING) # -- interne hjelpere --------------------------------------------------- def _set_state(self, new_state: PlaybackState) -> None: self._state = new_state self.emit("state-changed", new_state) def _on_bus_error(self, _bus, message) -> None: err, debug = message.parse_error() logger.error("GStreamer-feil: %s (%s)", err, debug) self._set_state(PlaybackState.ERROR) self.emit("error", str(err)) self._playbin.set_state(Gst.State.NULL) def _on_bus_buffering(self, _bus, message) -> None: percent = message.parse_buffering() if percent < 100: self._set_state(PlaybackState.BUFFERING) else: self._set_state(PlaybackState.PLAYING) def _on_bus_state_changed(self, _bus, message) -> None: if message.src != self._playbin: return _old, new, _pending = message.parse_state_changed() if new == Gst.State.PLAYING and self._state != PlaybackState.PLAYING: self._set_state(PlaybackState.PLAYING) def _on_bus_tag(self, _bus, message) -> None: """Plukker ut løpende metadata fra strømmen (f.eks. 'artist – tittel').""" tags = message.parse_tag() title = None ok, value = tags.get_string("title") if ok: title = value else: ok, value = tags.get_string("organization-name") if ok: title = value if title: self.emit("track-changed", title) # --------------------------------------------------------------------------- # MPRIS2-integrasjon # --------------------------------------------------------------------------- _MPRIS_INTROSPECTION_XML = """ """ class MPRISService: """Eksponerer RadioPlayer over MPRIS2 (org.mpris.MediaPlayer2) på D-Bus-sesjonsbussen. Dette gjør at GNOME Shell sitt medieelement viser stasjonen som spilles, og at medietaster på tastaturet (play/pause/stop) fungerer – uten at RadioPlayer eller RadioWindow trenger å kjenne til D-Bus. """ BUS_NAME = "org.mpris.MediaPlayer2.gnomeradio" OBJECT_PATH = "/org/mpris/MediaPlayer2" PLAYER_IFACE = "org.mpris.MediaPlayer2.Player" ROOT_IFACE = "org.mpris.MediaPlayer2" def __init__(self, player: RadioPlayer) -> None: self._player = player self._connection: Optional[Gio.DBusConnection] = None self._registration_ids: list[int] = [] self._bus_name_id = Gio.bus_own_name( Gio.BusType.SESSION, self.BUS_NAME, Gio.BusNameOwnerFlags.NONE, self._on_bus_acquired, None, None, ) self._player.connect( "state-changed", lambda *_a: self._notify_player_props(["PlaybackStatus"]) ) self._player.connect( "station-changed", lambda *_a: self._notify_player_props(["Metadata"]) ) self._player.connect( "track-changed", lambda *_a: self._notify_player_props(["Metadata"]) ) # -- oppsett ------------------------------------------------------------ def _on_bus_acquired(self, connection: Gio.DBusConnection, _name: str) -> None: self._connection = connection node_info = Gio.DBusNodeInfo.new_for_xml(_MPRIS_INTROSPECTION_XML) for interface in node_info.interfaces: reg_id = connection.register_object( self.OBJECT_PATH, interface, self._handle_method_call, self._handle_get_property, self._handle_set_property, ) self._registration_ids.append(reg_id) logger.info("MPRIS-tjeneste registrert som %s", self.BUS_NAME) def shutdown(self) -> None: if self._connection is not None: for reg_id in self._registration_ids: self._connection.unregister_object(reg_id) self._registration_ids.clear() if self._bus_name_id: Gio.bus_unown_name(self._bus_name_id) # -- metodekall ----------------------------------------------------- def _handle_method_call( self, _connection, _sender, _object_path, interface_name, method_name, _parameters, invocation, ) -> None: if interface_name == self.PLAYER_IFACE: if method_name == "Play": if self._player.state == PlaybackState.PAUSED: self._player.toggle_pause() elif self._player.current_station is not None: self._player.play(self._player.current_station) elif method_name == "Pause": if self._player.state == PlaybackState.PLAYING: self._player.toggle_pause() elif method_name == "PlayPause": self._player.toggle_pause() elif method_name == "Stop": self._player.stop() elif method_name in ("Next", "Previous"): pass # Ikke støttet ennå – ingen spillelistenavigasjon # org.mpris.MediaPlayer2: Raise/Quit gir ingen effekt i denne appen invocation.return_value(None) # -- egenskaper ----------------------------------------------------- def _handle_get_property( self, _connection, _sender, _object_path, interface_name, property_name ): if interface_name == self.ROOT_IFACE: root_values = { "CanQuit": GLib.Variant("b", False), "CanRaise": GLib.Variant("b", False), "HasTrackList": GLib.Variant("b", False), "Identity": GLib.Variant("s", "GNOME Radio"), "DesktopEntry": GLib.Variant("s", APP_ID), "SupportedUriSchemes": GLib.Variant("as", ["http", "https"]), "SupportedMimeTypes": GLib.Variant("as", []), } return root_values.get(property_name) if interface_name == self.PLAYER_IFACE: if property_name == "PlaybackStatus": status = { PlaybackState.PLAYING: "Playing", PlaybackState.PAUSED: "Paused", }.get(self._player.state, "Stopped") return GLib.Variant("s", status) if property_name == "Metadata": return GLib.Variant("a{sv}", self._build_metadata()) if property_name == "Volume": return GLib.Variant("d", self._player.volume) if property_name == "CanGoNext": return GLib.Variant("b", False) if property_name == "CanGoPrevious": return GLib.Variant("b", False) if property_name == "CanPlay": return GLib.Variant("b", True) if property_name == "CanPause": return GLib.Variant("b", True) if property_name == "CanControl": return GLib.Variant("b", True) return None def _handle_set_property( self, _connection, _sender, _object_path, interface_name, property_name, value ) -> bool: if interface_name == self.PLAYER_IFACE and property_name == "Volume": self._player.volume = value.unpack() return True return False def _build_metadata(self) -> dict: station = self._player.current_station if station is None: return {} track_id = self.OBJECT_PATH + "/current_track" return { "mpris:trackid": GLib.Variant("o", track_id), "xesam:title": GLib.Variant("s", station.name), "xesam:artist": GLib.Variant("as", [station.genre]), "xesam:album": GLib.Variant("s", station.country), } def _notify_player_props(self, prop_names: list[str]) -> None: if self._connection is None: return changed = {} for name in prop_names: value = self._handle_get_property( None, None, None, self.PLAYER_IFACE, name ) if value is not None: changed[name] = value if not changed: return self._connection.emit_signal( None, self.OBJECT_PATH, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant("(sa{sv}as)", (self.PLAYER_IFACE, changed, [])), ) # --------------------------------------------------------------------------- # UI-lag # --------------------------------------------------------------------------- class StationRow(Gtk.ListBoxRow): """Én rad i stasjonslisten. Rent visningsobjekt, ingen forretningslogikk.""" def __init__( self, station: Station, on_favorite_toggle: Callable[[Station], None], on_edit: Callable[[Station], None], on_delete: Callable[[Station], None], ): super().__init__() self.station = station box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8, margin_top=8, margin_bottom=8, margin_start=12, margin_end=12) text_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2, hexpand=True) name_label = Gtk.Label(label=station.name, xalign=0) name_label.add_css_class("title-4") meta_label = Gtk.Label(label=f"{station.genre} · {station.country}", xalign=0) meta_label.add_css_class("dim-label") text_box.append(name_label) text_box.append(meta_label) fav_button = Gtk.ToggleButton() fav_button.set_icon_name( "starred-symbolic" if station.favorite else "non-starred-symbolic" ) fav_button.set_active(station.favorite) fav_button.connect("toggled", lambda _b: on_favorite_toggle(station)) menu_button = Gtk.MenuButton(icon_name="view-more-symbolic") menu_button.add_css_class("flat") popover_menu = Gio.Menu() popover_menu.append("Rediger", "row.edit") popover_menu.append("Slett", "row.delete") menu_button.set_menu_model(popover_menu) actions = Gio.SimpleActionGroup() edit_action = Gio.SimpleAction.new("edit", None) edit_action.connect("activate", lambda *_a: on_edit(station)) delete_action = Gio.SimpleAction.new("delete", None) delete_action.connect("activate", lambda *_a: on_delete(station)) actions.add_action(edit_action) actions.add_action(delete_action) self.insert_action_group("row", actions) box.append(text_box) box.append(fav_button) box.append(menu_button) self.set_child(box) class StationEditDialog(Adw.Window): """Modal dialog for å legge til eller redigere en stasjon. Rent visningsobjekt: kjenner ikke til StationLibrary, bare til en `on_save`-callback som mottar den ferdige Station-instansen. """ def __init__( self, parent: Gtk.Window, on_save: Callable[[Station], None], station: Optional[Station] = None, ) -> None: is_edit = station is not None super().__init__( transient_for=parent, modal=True, default_width=380, default_height=-1, title="Rediger stasjon" if is_edit else "Legg til stasjon", ) self._on_save = on_save self._editing_station = station toolbar_view = Adw.ToolbarView() header = Adw.HeaderBar() toolbar_view.add_top_bar(header) save_button = Gtk.Button(label="Lagre") save_button.add_css_class("suggested-action") save_button.connect("clicked", self._on_save_clicked) header.pack_end(save_button) cancel_button = Gtk.Button(label="Avbryt") cancel_button.connect("clicked", lambda _b: self.close()) header.pack_start(cancel_button) group = Adw.PreferencesGroup( margin_top=16, margin_bottom=16, margin_start=16, margin_end=16 ) self._name_row = Adw.EntryRow(title="Navn") self._url_row = Adw.EntryRow(title="Strøm-URL") self._genre_row = Adw.EntryRow(title="Sjanger") self._country_row = Adw.EntryRow(title="Land") if station is not None: self._name_row.set_text(station.name) self._url_row.set_text(station.url) self._genre_row.set_text(station.genre) self._country_row.set_text(station.country) else: self._country_row.set_text("Norge") for row in (self._name_row, self._url_row, self._genre_row, self._country_row): group.add(row) toolbar_view.set_content(group) self.set_content(toolbar_view) def _on_save_clicked(self, _button: Gtk.Button) -> None: name = self._name_row.get_text().strip() url = self._url_row.get_text().strip() if not name or not url: self._name_row.add_css_class("error" if not name else "") self._url_row.add_css_class("error" if not url else "") return station = Station( name=name, url=url, genre=self._genre_row.get_text().strip() or "Ukjent", country=self._country_row.get_text().strip() or "Norge", favorite=self._editing_station.favorite if self._editing_station else False, ) self._on_save(station) self.close() class RadioWindow(Adw.ApplicationWindow): """Hovedvinduet. Observerer RadioPlayer og delegerer handlinger til det.""" def __init__(self, app: "RadioApplication") -> None: super().__init__(application=app, title="GNOME Radio") self.set_default_size(420, 640) self._library = app.library self._player = app.player self._player.connect("state-changed", self._on_state_changed) self._player.connect("station-changed", self._on_station_changed) self._player.connect("track-changed", self._on_track_changed) self._player.connect("error", self._on_player_error) self._showing_favorites_only = False self._current_search = "" self._build_ui() self._populate_station_list(self._library.all()) # -- oppbygging av UI --------------------------------------------------- def _build_ui(self) -> None: toolbar_view = Adw.ToolbarView() header = Adw.HeaderBar() toolbar_view.add_top_bar(header) search_entry = Gtk.SearchEntry(placeholder_text="Søk etter stasjon, sjanger …") search_entry.connect("search-changed", self._on_search_changed) header.set_title_widget(search_entry) self._favorites_toggle = Gtk.ToggleButton(icon_name="starred-symbolic") self._favorites_toggle.set_tooltip_text("Vis kun favoritter") self._favorites_toggle.connect("toggled", self._on_favorites_filter_toggled) header.pack_start(self._favorites_toggle) add_button = Gtk.Button(icon_name="list-add-symbolic") add_button.set_tooltip_text("Legg til stasjon") add_button.connect("clicked", self._on_add_station_clicked) header.pack_end(add_button) primary_menu = Gio.Menu() primary_menu.append("Importer stasjoner …", "win.import-stations") primary_menu.append("Eksporter stasjoner …", "win.export-stations") primary_menu_button = Gtk.MenuButton(icon_name="open-menu-symbolic") primary_menu_button.set_menu_model(primary_menu) primary_menu_button.set_tooltip_text("Meny") header.pack_end(primary_menu_button) import_action = Gio.SimpleAction.new("import-stations", None) import_action.connect("activate", self._on_import_stations) self.add_action(import_action) export_action = Gio.SimpleAction.new("export-stations", None) export_action.connect("activate", self._on_export_stations) self.add_action(export_action) root = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) self._station_list = Gtk.ListBox() self._station_list.add_css_class("boxed-list") self._station_list.set_selection_mode(Gtk.SelectionMode.NONE) self._station_list.connect("row-activated", self._on_row_activated) scroller = Gtk.ScrolledWindow(vexpand=True) scroller.set_child(self._station_list) scroller.set_margin_top(8) scroller.set_margin_start(8) scroller.set_margin_end(8) root.append(scroller) root.append(self._build_playback_bar()) toolbar_view.set_content(root) self._toast_overlay = Adw.ToastOverlay() self._toast_overlay.set_child(toolbar_view) self.set_content(self._toast_overlay) def _build_playback_bar(self) -> Gtk.Widget: bar = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4, margin_top=8, margin_bottom=12, margin_start=16, margin_end=16) self._now_playing_label = Gtk.Label(label="Ingen stasjon valgt", xalign=0) self._now_playing_label.add_css_class("title-3") self._track_label = Gtk.Label(label="", xalign=0, visible=False) self._track_label.add_css_class("dim-label") self._track_label.set_ellipsize(3) # Pango.EllipsizeMode.END self._status_label = Gtk.Label(label="Stoppet", xalign=0) self._status_label.add_css_class("dim-label") controls = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) self._play_pause_button = Gtk.Button(icon_name="media-playback-start-symbolic") self._play_pause_button.connect("clicked", lambda _b: self._player.toggle_pause()) stop_button = Gtk.Button(icon_name="media-playback-stop-symbolic") stop_button.connect("clicked", lambda _b: self._player.stop()) self._mute_button = Gtk.ToggleButton(icon_name="audio-volume-high-symbolic") self._mute_button.connect("toggled", self._on_mute_toggled) volume_scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 1, 0.05) volume_scale.set_value(self._player.volume) volume_scale.set_hexpand(True) volume_scale.connect("value-changed", lambda s: setattr(self._player, "volume", s.get_value())) self._volume_scale = volume_scale controls.append(self._play_pause_button) controls.append(stop_button) controls.append(self._mute_button) controls.append(volume_scale) bar.append(self._now_playing_label) bar.append(self._track_label) bar.append(self._status_label) bar.append(controls) return bar # -- stasjonslisten ----------------------------------------------------- def _populate_station_list(self, stations: list[Station]) -> None: child = self._station_list.get_first_child() while child is not None: next_child = child.get_next_sibling() self._station_list.remove(child) child = next_child for station in stations: row = StationRow( station, on_favorite_toggle=self._on_favorite_toggle, on_edit=self._on_edit_station, on_delete=self._on_delete_station, ) self._station_list.append(row) def _refresh_station_list(self) -> None: """Slår sammen gjeldende søk og favoritt-filter og tegner listen på nytt.""" stations = self._library.search(self._current_search) if self._showing_favorites_only: stations = [s for s in stations if s.favorite] self._populate_station_list(stations) def _on_search_changed(self, entry: Gtk.SearchEntry) -> None: self._current_search = entry.get_text() self._refresh_station_list() def _on_favorites_filter_toggled(self, button: Gtk.ToggleButton) -> None: self._showing_favorites_only = button.get_active() self._refresh_station_list() def _on_row_activated(self, _listbox, row: StationRow) -> None: self._player.play(row.station) def _on_favorite_toggle(self, station: Station) -> None: self._library.toggle_favorite(station) if self._showing_favorites_only: self._refresh_station_list() def _on_add_station_clicked(self, _button: Gtk.Button) -> None: dialog = StationEditDialog(self, on_save=self._on_station_saved) dialog.present() def _on_edit_station(self, station: Station) -> None: dialog = StationEditDialog(self, on_save=self._on_station_saved, station=station) dialog.present() def _on_station_saved(self, station: Station) -> None: existing_urls = {s.url for s in self._library.all()} is_new_url = station.url not in existing_urls # Enkel heuristikk: hvis dialogen ble åpnet for redigering av en # stasjon som fortsatt finnes med samme URL, bruk update(); ellers add(). matching = next((s for s in self._library.all() if s.url == station.url), None) if matching is not None: self._library.update(matching, station) self._toast_overlay.add_toast(Adw.Toast(title=f"Oppdaterte «{station.name}»")) else: self._library.add(station) self._toast_overlay.add_toast(Adw.Toast(title=f"La til «{station.name}»")) self._refresh_station_list() def _on_delete_station(self, station: Station) -> None: self._library.remove(station) if self._player.current_station is station: self._player.stop() self._refresh_station_list() self._toast_overlay.add_toast(Adw.Toast(title=f"Slettet «{station.name}»")) # -- import/eksport ------------------------------------------------- def _on_export_stations(self, _action=None, _param=None) -> None: dialog = Gtk.FileDialog(initial_name="radiostasjoner.json") dialog.save(self, None, self._on_export_dialog_done) def _on_export_dialog_done(self, dialog: Gtk.FileDialog, result: Gio.AsyncResult) -> None: try: gfile = dialog.save_finish(result) except GLib.Error as exc: logger.info("Eksport avbrutt eller feilet: %s", exc) return path = Path(gfile.get_path()) try: count = self._library.export_to_path(path) except OSError as exc: self._toast_overlay.add_toast(Adw.Toast(title=f"Kunne ikke eksportere: {exc}")) return self._toast_overlay.add_toast( Adw.Toast(title=f"Eksporterte {count} stasjoner til {path.name}") ) def _on_import_stations(self, _action=None, _param=None) -> None: json_filter = Gtk.FileFilter() json_filter.set_name("JSON-filer") json_filter.add_pattern("*.json") filters = Gio.ListStore.new(Gtk.FileFilter) filters.append(json_filter) dialog = Gtk.FileDialog(default_filter=json_filter) dialog.set_filters(filters) dialog.open(self, None, self._on_import_dialog_done) def _on_import_dialog_done(self, dialog: Gtk.FileDialog, result: Gio.AsyncResult) -> None: try: gfile = dialog.open_finish(result) except GLib.Error as exc: logger.info("Import avbrutt eller feilet: %s", exc) return path = Path(gfile.get_path()) try: count = self._library.import_from_path(path, replace=False) except (json.JSONDecodeError, ValueError, KeyError, OSError) as exc: self._toast_overlay.add_toast(Adw.Toast(title=f"Kunne ikke importere: {exc}")) return self._refresh_station_list() self._toast_overlay.add_toast(Adw.Toast(title=f"Importerte {count} stasjoner")) def _on_mute_toggled(self, button: Gtk.ToggleButton) -> None: self._player.toggle_mute() muted = self._player.is_muted button.set_icon_name( "audio-volume-muted-symbolic" if muted else "audio-volume-high-symbolic" ) self._volume_scale.set_sensitive(not muted) # -- reaksjoner på RadioPlayer-signaler --------------------------------- def _on_state_changed(self, _player, state: PlaybackState) -> None: labels = { PlaybackState.STOPPED: "Stoppet", PlaybackState.BUFFERING: "Bufrer …", PlaybackState.PLAYING: "Spiller av", PlaybackState.PAUSED: "Pause", PlaybackState.ERROR: "Feil ved avspilling", } self._status_label.set_label(labels.get(state, "")) icon = "media-playback-pause-symbolic" if state == PlaybackState.PLAYING \ else "media-playback-start-symbolic" self._play_pause_button.set_icon_name(icon) def _on_station_changed(self, _player, station: Station) -> None: self._now_playing_label.set_label(station.name) self._track_label.set_visible(False) self._track_label.set_label("") def _on_track_changed(self, _player, title: str) -> None: if title: self._track_label.set_label(title) self._track_label.set_visible(True) else: self._track_label.set_visible(False) self._track_label.set_label("") def _on_player_error(self, _player, message: str) -> None: logger.error("Avspillingsfeil: %s", message) self._toast_overlay.add_toast( Adw.Toast(title=f"Kunne ikke spille av: {message}", timeout=5) ) # --------------------------------------------------------------------------- # Applikasjon # --------------------------------------------------------------------------- class RadioApplication(Adw.Application): """Gtk.Application-underklasse. Eier de langlevde objektene.""" def __init__(self) -> None: super().__init__(application_id=APP_ID, flags=Gio.ApplicationFlags.DEFAULT_FLAGS) self.library = StationLibrary() self.player = RadioPlayer() self.mpris = MPRISService(self.player) self._window: Optional[RadioWindow] = None def do_activate(self) -> None: # noqa: N802 (GTK-navnekonvensjon) if self._window is None: self._window = RadioWindow(self) self._window.present() def do_shutdown(self) -> None: # noqa: N802 (GTK-navnekonvensjon) self.mpris.shutdown() Adw.Application.do_shutdown(self) def main() -> int: app = RadioApplication() return app.run(None) if __name__ == "__main__": raise SystemExit(main())