diff --git a/modules/controlcenter/main.py b/modules/controlcenter/main.py index 1e61a46..8837d02 100644 --- a/modules/controlcenter/main.py +++ b/modules/controlcenter/main.py @@ -16,8 +16,8 @@ from services.brightness import Brightness from utils.exml import exml from utils.roam import audio_service, modus_service from widgets.wayland import WaylandWindow as Window -from modules.controlcenter.player import PlayerContainer -# from modules.controlcenter.player1 import PlayerContainer +from modules.controlcenter.player import PlayerBoxStack +from services.mpris import MprisPlayerManager brightness_service = Brightness.get_initial() @@ -106,7 +106,7 @@ class ModusControlCenter(Window): self.music_widget = Box( name="music-widget", - children=[PlayerContainer()], + children=[PlayerBoxStack(MprisPlayerManager())], ) self.wifi_man = WifiConnections(self) diff --git a/modules/controlcenter/player.py b/modules/controlcenter/player.py index 0a1c6d8..83db33a 100644 --- a/modules/controlcenter/player.py +++ b/modules/controlcenter/player.py @@ -1,560 +1,722 @@ +import os import re +import tempfile +import urllib.parse +import urllib.request +from functools import partial +from typing import List -from gi.repository import Playerctl, GLib -from fabric.core.service import Signal +from fabric.utils import ( + bulk_connect, + cooldown, +) from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox -from fabric.widgets.label import Label -from fabric.widgets.stack import Stack -from fabric.widgets.eventbox import EventBox from fabric.widgets.image import Image -from fabric.widgets.svg import Svg +from fabric.widgets.label import Label +from fabric.widgets.overlay import Overlay +from fabric.widgets.scale import Scale +from fabric.widgets.stack import Stack +from gi.repository import GLib, GObject from loguru import logger +from services.mpris import MprisPlayer, MprisPlayerManager +from utils.animator import Animator import config.data as data -from fabric.utils import get_relative_path -from services.mpris import PlayerManager, PlayerService + +CACHE_DIR = f"{data.CACHE_DIR}/media" -class Player(Box): - def __init__(self, player, **kwargs): - super().__init__(style_classes="player-box", **kwargs) - - self._player = PlayerService(player=player) - self._current_thumbnail = None # Cache for current thumbnail - self._update_queue = [] - self._is_updating = False - self._update_timeout_id = None - - self.duration = 0.0 - - self._player.connect("meta-change", self.on_metadata) - self._player.connect("track-position", self.on_update_track_position) - - player.connect("metadata", self.on_direct_metadata_change) - - self.player_name = Image( - name=player.props.player_name, - style_classes="player-icon", - icon_name=self._get_player_icon_name(player.props.player_name), - size=20, - ) - - self.song = Label( - name="song", - label="song", - justification="left", - h_align="start", - ellipsization="end", - max_chars_width=22, - ) - self.artist = Label( - name="artist", - label="artist", - justification="left", - h_align="start", - ) - self.music = Box( - name="music", - h_expand=True, - v_expand=True, - children=[self.song], - ) - - self.play_pause_button = Button( - name="play-pause-button", - child=Svg( - svg_file=get_relative_path("../../config/assets/icons/media/play.svg"), - size=20, - ), - tooltip_text="Play/Pause", - on_clicked=lambda b, *_: self.handle_play_pause(player), - ) - - self.shuffle_button = Button( - name="shuffle-button", - child=Svg( - svg_file=get_relative_path( - "../../config/assets/icons/media/shuffle.svg" - ), - size=20, - ), - on_clicked=lambda b, *_: self.handle_shuffle(b, player), - ) - - self.album_cover = Box(style_classes="album-image") - self.album_cover.set_style( - f"background-image:url('{data.HOME_DIR}/.current.wall')" - ) - - self.children = [ - self.album_cover, - Box( - name="source", - v_align="end", - children=self.player_name, - ), - Box( - orientation="v", - v_expand=True, - h_expand=True, - style="padding-left:10px;", - children=[ - CenterBox( - name="details", - center_children=self.music, - ), - Box( - name="controls", - spacing=5, - h_expand=True, - children=[ - CenterBox( - h_expand=True, - v_expand=True, - center_children=[ - Button( - name="prev-button", - child=Svg( - svg_file=get_relative_path( - "../../config/assets/icons/media/previous.svg" - ), - size=20, - ), - on_clicked=lambda b, *_: self.handle_prev( - player - ), - ), - self.play_pause_button, - Button( - name="next-button", - child=Svg( - svg_file=get_relative_path( - "../../config/assets/icons/media/next.svg" - ), - size=20, - ), - on_clicked=lambda b, *_: self.handle_next( - player - ), - ), - self.shuffle_button, - ], - ) - ], - ), - ], - ), - ] - - self.on_metadata(self._player, metadata=player.props.metadata, player=player) - - def _get_player_icon_name(self, player_name): - symbolic_players = [ - "spotify", - "firefox", - "chromium", - "chrome", - "brave", - "vivaldi", - ] - - if player_name.lower() in symbolic_players: - icon_name = f"{player_name.lower()}-symbolic" - return icon_name - else: - return player_name.lower() - - def _update_play_pause_icon(self, is_playing): - if is_playing: - self.play_pause_button.get_child().set_from_file( - get_relative_path("../../config/assets/icons/media/pause.svg") - ) - else: - self.play_pause_button.get_child().set_from_file( - get_relative_path("../../config/assets/icons/media/play.svg") - ) - - def _update_shuffle_icon(self, is_shuffled): - if is_shuffled: - self.shuffle_button.get_child().set_from_file( - get_relative_path("../../config/assets/icons/media/shuffle.svg") - ) - self.shuffle_button.get_child().set_name("shuffle") - else: - self.shuffle_button.get_child().set_from_file( - get_relative_path("../../config/assets/icons/media/no-shuffle.svg") - ) - self.shuffle_button.get_child().set_name("disable-shuffle") - - def on_direct_metadata_change(self, player, metadata): - try: - logger.debug(f"Direct metadata change for {player.props.player_name}") - self.on_metadata(self._player, metadata, player) - except Exception as e: - logger.warning(f"Failed to handle direct metadata change: {e}") - - def on_update_track_position(self, sender, pos, dur): - if dur == 0: - return - self.duration = dur - - def on_seek(self, sender, ratio): - pos = ratio * self.duration # duration in seconds - self._player.set_position(int(pos)) - - def on_metadata(self, sender, metadata, player): - def _update_metadata(): - keys = metadata.keys() - if "xesam:artist" in keys and "xesam:title" in keys: - song_title = metadata["xesam:title"] - self.song.set_label(song_title) - - artist_list = metadata["xesam:artist"] - artist_name = artist_list[0] if artist_list else "Unknown Artist" - self.artist.set_label(artist_name) - - self._handle_thumbnail(metadata, player) - - # Update play/pause state - if ( - player.props.playback_status.value_name - == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" - ): - self._update_play_pause_icon(True) - else: - self._update_play_pause_icon(False) - - # Update shuffle state - if self._is_shuffle_supported(player): - if player.props.shuffle == True: - self._update_shuffle_icon(True) - else: - self._update_shuffle_icon(False) - else: - self.shuffle_button.set_visible(False) - - self._queue_update(_update_metadata) - - def cleanup(self): - """Clean up resources to prevent memory leaks.""" - if self._update_timeout_id: - GLib.source_remove(self._update_timeout_id) - self._update_timeout_id = None - self._update_queue.clear() - self._is_updating = False - - def _is_shuffle_supported(self, player): - try: - _ = player.props.shuffle - return True - except Exception: - return False - - def _queue_update(self, update_func, *args, **kwargs): - """Add an update function to the queue and process if not already running.""" - self._update_queue.append((update_func, args, kwargs)) - if not self._is_updating: - self._process_queue() - - def _process_queue(self): - """Process the update queue sequentially.""" - if not self._update_queue or self._is_updating: +def cleanup_old_cache_files(): + """Clean up old artwork cache files (older than 1 day).""" + try: + if not os.path.exists(CACHE_DIR): return - self._is_updating = True - update_func, args, kwargs = self._update_queue.pop(0) + import time - try: - update_func(*args, **kwargs) - except Exception as e: - logger.warning(f"Error in queue update: {e}") - finally: - # Schedule next update after a short delay - if self._update_timeout_id: - GLib.source_remove(self._update_timeout_id) - self._update_timeout_id = GLib.timeout_add(100, self._finish_update) + current_time = time.time() + one_day_ago = current_time - (24 * 60 * 60) # 24 hours - def _finish_update(self): - """Finish current update and process next item in queue.""" - self._is_updating = False - self._update_timeout_id = None - - if self._update_queue: - self._process_queue() - return False - - def _handle_thumbnail(self, metadata, player): - """Queue thumbnail update to prevent overlapping operations.""" - - def _update_thumbnail(): - keys = metadata.keys() - player_name = player.props.player_name - - new_thumbnail = None - - # Check for MPRIS art URL first - if "mpris:artUrl" in keys: - art_url = metadata["mpris:artUrl"] - if art_url and art_url.strip(): - new_thumbnail = art_url - logger.debug(f"Found MPRIS art for {player_name}: {art_url}") - - # Check for browser players and YouTube URLs - elif ( - player_name in ["firefox", "chromium", "vivaldi", "brave"] - and "xesam:url" in keys - ): - webpage_url = metadata["xesam:url"] - if webpage_url and "youtube.com" in webpage_url: - # Simple YouTube thumbnail extraction - import re - - match = re.search(r"(?:v=|/)([a-zA-Z0-9_-]{11})", webpage_url) - if match: - video_id = match.group(1) - new_thumbnail = ( - f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg" - ) - logger.debug( - f"Found YouTube thumbnail for {player_name}: { - new_thumbnail - }" - ) - - # Only update if thumbnail has changed - if new_thumbnail and new_thumbnail != self._current_thumbnail: - self._current_thumbnail = new_thumbnail - self.album_cover.set_style(f"background-image:url('{new_thumbnail}')") - logger.debug(f"Updated thumbnail for {player_name}") - elif not new_thumbnail and self._current_thumbnail: - # Reset to default if no thumbnail found - self._current_thumbnail = None - self.album_cover.set_style( - f"background-image:url('{data.HOME_DIR}/.current.wall')" - ) - logger.debug(f"Reset to default wallpaper for {player_name}") - - self._queue_update(_update_thumbnail) - - def handle_next(self, player): - def _do_next(): + for filename in os.listdir(CACHE_DIR): + filepath = os.path.join(CACHE_DIR, filename) try: - self._player._player.next() - except Exception as e: - logger.warning(f"Failed to go to next track: {e}") - - self._queue_update(_do_next) - - def handle_prev(self, player): - def _do_prev(): - try: - self._player._player.previous() - except Exception as e: - logger.warning(f"Failed to go to previous track: {e}") - - self._queue_update(_do_prev) - - def handle_play_pause(self, player): - def _do_play_pause(): - is_playing = ( - self._player._player.props.playback_status.value_name - == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" - ) - - def _set_play_ui(): - self._update_play_pause_icon(True) - - def _set_pause_ui(): - self._update_play_pause_icon(False) - - if is_playing: - _set_pause_ui() - else: - _set_play_ui() - - try: - self._player._player.play_pause() - except Exception as e: - if is_playing: - _set_pause_ui() - else: - _set_play_ui() - logger.warning("Failed to toggle playback:", e) - - self._queue_update(_do_play_pause) - - def handle_shuffle(self, shuffle_button, player): - def _do_shuffle(): - print("shuffle", player.props.shuffle) - - if not self._is_shuffle_supported(player): - print(f"Shuffle not supported for {player.props.player_name}") - return - - try: - if player.props.shuffle == False: - player.set_shuffle(True) - print("setting to true", player.props.player_name) - else: - player.set_shuffle(False) - except Exception as e: - print(f"Failed to toggle shuffle for {player.props.player_name}: {e}") - pass - - self._queue_update(_do_shuffle) + if os.path.isfile(filepath): + file_mtime = os.path.getmtime(filepath) + if file_mtime < one_day_ago: + os.unlink(filepath) + except Exception: + pass # Ignore individual file errors + except Exception: + pass # Ignore all errors in cleanup -class Placeholder(Box): - def __init__(self, **kwargs): - super().__init__(style_classes="player-box", **kwargs) +class PlayerBoxStack(Box): + """A widget that displays the current player information.""" - self.album_cover = Box(style_classes="album-image") - self.album_cover.set_style( - f"background-image:url('{data.HOME_DIR}/.current.wall')" - ) - self.player_name = Image( - name="player-icon", - icon_name="media-player-48", - size=20, - ) - - self.children = [ - self.album_cover, - Box( - name="source", - v_align="end", - children=self.player_name, - ), - Box( - orientation="v", - v_expand=True, - h_expand=True, - style="padding-left:10px;", - v_align="center", - children=[ - CenterBox( - name="details", - center_children=Label(label="Nothing Playing"), - ) - ], - ), - ] - - -class PlayerContainer(Box): - @Signal - def active_player_changed(self, player: Playerctl.Player) -> None: ... - - def __init__(self, **kwargs): - super().__init__(name="player-container", orientation="v", **kwargs) - - self.manager = PlayerManager() - self.manager.connect("new-player", self.new_player) - self.manager.connect("player-vanish", self.on_player_vanish) - self.placeholder = Placeholder() - self.stack = Stack( - name="player-container", - transition_type="crossfade", - transition_duration=100, - children=[self.placeholder], - ) + def __init__(self, mpris_manager: MprisPlayerManager, **kwargs): + # Clean up old cache files on startup + cleanup_old_cache_files() + # The player stack self.player_stack = Stack( + # transition_type="slide-left-right", + # transition_duration=500, name="player-stack", - transition_type="crossfade", - transition_duration=100, - children=[], + ) + self.current_stack_pos = 0 + + # List to store player buttons + self.player_buttons: list[Button] = [] + + # Create a "No media playing" placeholder + self.no_media_box = self._create_no_media_box() + + super().__init__(orientation="v", name="media", children=[self.player_stack]) + + # Show the no media box initially + self.player_stack.children = [self.no_media_box] + self.set_visible(True) + + self.mpris_manager = mpris_manager + + bulk_connect( + self.mpris_manager, + { + "player-appeared": self.on_new_player, + "player-vanished": self.on_lost_player, + }, ) - self.player_switch_container = CenterBox( - name="player-switch-container", - orientation="v", - center_children=[], - ) - self.event_box = EventBox( - child=Box(children=[self.stack, self.player_switch_container]), - ) - self.children = self.event_box - self.player = [] - self.player_objects = {} # Map player names to player objects - self.active_player = None # Track the currently active player - self.manager.init_all_players() - - def new_player(self, manager, player): - new_player = Player(player=player) - new_player.set_name(player.props.player_name) - self.player.append(new_player) - # Store the player object - self.player_objects[player.props.player_name] = player - self.stack.add_named(new_player, player.props.player_name) - if len(self.player) == 1: - self.stack.remove(self.placeholder) - - self.player_switch_container.add_center( - Button( - name=player.props.player_name, - style_classes="player-button", - on_clicked=lambda b: self.switch_player(player.props.player_name, b), + for player in self.mpris_manager.players: # type: ignore + logger.info( + f"[PLAYER MANAGER] player found: {player.get_property('player-name')}", ) + self.on_new_player(self.mpris_manager, player) + + def _create_no_media_box(self): + """Create a placeholder box for when no media is playing.""" + fallback_cover_path = f"{data.HOME_DIR}/.current.wall" + + # Album cover with fallback image + album_cover = Box(style_classes="album-image") + album_cover.set_style(f"background-image:url('{fallback_cover_path}')") + + image_stack = Box(h_align="start", v_align="center", name="player-image-stack") + image_stack.children = [album_cover] + + # Track info showing "No media playing" + track_title = Label( + label="No media playing", + name="player-title", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", ) - # If this is the first player, make it active - if not self.active_player: - self.active_player = player - self.active_player_changed(player) + track_artist = Label( + label="", + name="player-artist", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", + visible=False, # Hide artist and album when no media + ) - self.update_player_list() + track_album = Label( + label="", + name="player-album", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", + visible=False, # Hide artist and album when no media + ) - def switch_player(self, player_name, button): - self.stack.set_visible_child_name(player_name) + track_info = Box( + name="track-info", + spacing=5, + orientation="v", + v_align="start", + h_align="start", + children=[track_title, track_artist, track_album], + ) - for btn in self.player_switch_container.center_children: - btn.remove_style_class("active") - button.add_style_class("active") + # No control buttons for no media state - just an empty box + controls_box = Box( + name="player-controls", + visible=False, # Hide controls when no media + ) - # Get the player object and emit signal - if player_name in self.player_objects: - self.active_player = self.player_objects[player_name] - self.active_player_changed(self.active_player) + player_info_box = Box( + name="player-info-box", + v_align="center", + h_align="start", + orientation="v", + children=[track_info, controls_box], + ) - def on_player_vanish(self, manager, player): + inner_box = Box( + name="inner-player-box", + v_align="center", + h_align="start", + ) + + outer_box = Box( + name="outer-player-box", + h_align="start", + ) + + overlay_box = Overlay( + child=outer_box, + overlays=[ + inner_box, + player_info_box, + image_stack, + ], + ) + + no_media_box = Box( + h_align="center", + name="player-box", + h_expand=True, + children=[overlay_box], + ) + + return no_media_box + + def on_player_clicked(self, type): + # unset active from prev active button + self.player_buttons[self.current_stack_pos].remove_style_class("active") + if type == "next": + self.current_stack_pos = ( + self.current_stack_pos + 1 + if self.current_stack_pos != len(self.player_stack.get_children()) - 1 + else 0 + ) + elif type == "prev": + self.current_stack_pos = ( + self.current_stack_pos - 1 + if self.current_stack_pos != 0 + else len(self.player_stack.get_children()) - 1 + ) + # set new active button + self.player_buttons[self.current_stack_pos].add_style_class("active") + self.player_stack.set_visible_child( + self.player_stack.get_children()[self.current_stack_pos], + ) + + def on_player_clicked_by_index(self, index): + """Switch to player at given index""" + if 0 <= index < len(self.player_buttons): + # unset active from prev active button + self.player_buttons[self.current_stack_pos].remove_style_class("active") + # set new position + self.current_stack_pos = index + # set new active button + self.player_buttons[self.current_stack_pos].add_style_class("active") + self.player_stack.set_visible_child( + self.player_stack.get_children()[self.current_stack_pos], + ) + # Update all player boxes with new button state + self._update_all_player_buttons() + + def on_new_player(self, mpris_manager, player): player_name = player.props.player_name - for player_instance in self.player: - if player_instance.get_name() == player_name: - player_instance.cleanup() # Clean up resources - self.stack.remove(player_instance) - self.player.remove(player_instance) - for btn in self.player_switch_container.center_children: - if btn.get_name() == player_name: - self.player_switch_container.remove_center(btn) - break + # if player_name in self.config.get("ignore", []): + # return - # Clean up player objects dictionary - if player_name in self.player_objects: - del self.player_objects[player_name] + # Remove the no media box if it's the only child + if ( + len(self.player_stack.get_children()) == 1 + and self.player_stack.get_children()[0] == self.no_media_box + ): + self.player_stack.children = [] + self.current_stack_pos = 0 - # If the active player vanished, switch to another one - if self.active_player and self.active_player.props.player_name == player_name: - self.active_player = None - # Try to find another player to make active - if self.player_objects: - first_player_name = next(iter(self.player_objects)) - self.active_player = self.player_objects[first_player_name] - self.active_player_changed(self.active_player) + self.set_visible(True) - self.update_player_list() + new_player_box = PlayerBox(player=MprisPlayer(player), player_stack=self) + self.player_stack.children = [ + *self.player_stack.children, + new_player_box, + ] - if len(self.player) == 0: - self.stack.add_named(self.placeholder, "placeholder") + self.make_new_player_button(self.player_stack.get_children()[-1]) + logger.info( + f"[PLAYER MANAGER] adding new player: {player.get_property('player-name')}", + ) + if self.player_buttons: + self.player_buttons[self.current_stack_pos].set_style_classes(["active"]) - def update_player_list(self): - curr = self.stack.get_visible_child() - if curr: - for btn in self.player_switch_container.center_children: - if btn.get_name() == curr.get_name(): - btn.add_style_class("active") - else: - btn.remove_style_class("active") + # Update all player boxes with current button state + self._update_all_player_buttons() + + def on_lost_player(self, mpris_manager, player_name): + # the playerBox is automatically removed from mprisbox children on being removed + logger.info(f"[PLAYER_MANAGER] Player Removed {player_name}") + players: List[PlayerBox] = self.player_stack.get_children() + + # Check if this was the last player + if len(players) == 1 and player_name == players[0].player.player_name: + # Show the no media box instead of hiding + self.player_stack.children = [self.no_media_box] + self.current_stack_pos = 0 + self.player_buttons = [] # Clear player buttons + return + + if ( + players + and self.current_stack_pos < len(players) + and players[self.current_stack_pos].player.player_name == player_name + ): + self.current_stack_pos = max(0, self.current_stack_pos - 1) + if self.player_stack.get_children(): + self.player_stack.set_visible_child( + self.player_stack.get_children()[self.current_stack_pos], + ) + + if self.player_buttons and self.current_stack_pos < len(self.player_buttons): + self.player_buttons[self.current_stack_pos].set_style_classes(["active"]) + + # Update all player boxes with current button state + self._update_all_player_buttons() + + def make_new_player_button(self, player_box): + new_button = Button(name="player-stack-button") + + def on_player_button_click(button: Button): + self.player_buttons[self.current_stack_pos].remove_style_class("active") + self.current_stack_pos = self.player_buttons.index(button) + button.add_style_class("active") + self.player_stack.set_visible_child(player_box) + + new_button.connect( + "clicked", + on_player_button_click, + ) + self.player_buttons.append(new_button) + + # This will automatically destroy our used button + player_box.connect( + "destroy", + lambda *_: [ + new_button.destroy(), # type: ignore + self.player_buttons.pop(self.player_buttons.index(new_button)), + ], + ) + + def _update_all_player_buttons(self): + """Update all player boxes with the current button state""" + players: List[PlayerBox] = self.player_stack.get_children() + logger.info( + f"[PlayerBoxStack] Updating buttons for {len(players)} players, { + len(self.player_buttons) + } buttons" + ) + for player_box in players: + if hasattr(player_box, "update_buttons"): + player_box.update_buttons(self.player_buttons, len(players) > 1) + else: + logger.warning( + f"[PlayerBoxStack] PlayerBox missing update_buttons method" + ) + + +class PlayerBox(Box): + """A widget that displays the current player information.""" + + def __init__(self, player: MprisPlayer, player_stack=None, **kwargs): + super().__init__( + h_align="center", + name="player-box", + **kwargs, + h_expand=True, + ) + # Setup + self.player: MprisPlayer = player + self.player_stack = player_stack + self.fallback_cover_path = f"{data.HOME_DIR}/.current.wall" + + self.image_size = 120 + + self.icon_size = 15 + + # State + self.exit = False + self.skipped = False + + # Memory management + self.temp_artwork_files = [] # Track temp files for cleanup + self.current_download_thread = None # Track current download thread + + self.album_cover = Box(style_classes="album-image") + self.album_cover.set_style( + f"background-image:url('{self.fallback_cover_path}')" + ) + + self.image_stack = Box( + h_align="start", v_align="center", name="player-image-stack" + ) + self.image_stack.children = [*self.image_stack.children, self.album_cover] + + # Track Info + self.track_title = Label( + label="No Title", + name="player-title", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", + ) + + self.track_artist = Label( + label="No Artist", + name="player-artist", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", + visible=True, + ) + + self.track_album = Label( + label="No Album", + name="player-album", + justification="left", + max_chars_width=15, + ellipsization="end", + h_align="start", + visible=True, + ) + + self.player.bind_property( + "title", + self.track_title, + "label", + GObject.BindingFlags.DEFAULT, + lambda _, x: ( + re.sub(r"\r?\n", " ", x) if x != "" and x is not None else "No Title" + ), # type: ignore + ) + self.player.bind_property( + "artist", + self.track_artist, + "label", + GObject.BindingFlags.DEFAULT, + lambda _, x: ( + re.sub(r"\r?\n", " ", x) if x != "" and x is not None else "No Artist" + ), # type: ignore + ) + + self.player.bind_property( + "album", + self.track_album, + "label", + GObject.BindingFlags.DEFAULT, + lambda _, x: ( + re.sub(r"\r?\n", " ", x) if x != "" and x is not None else "No Album" + ), # type: ignore + ) + + self.track_info = Box( + name="track-info", + spacing=5, + orientation="v", + v_align="start", + h_align="start", + children=[ + self.track_title, + self.track_artist, + self.track_album, + ], + ) + + # Buttons + self.button_box = Box( + name="button-box", + h_align="center", + spacing=2, + ) + + # Stack switcher buttons box + self.stack_buttons_box = Box( + name="stack-buttons-box", spacing=4, h_align="center" + ) + self.stack_buttons_box.hide() # Initially hidden + + self.controls_box = CenterBox( + name="player-controls", + center_children=self.button_box, + end_children=self.stack_buttons_box, + ) + + self.skip_next_icon = Image(icon_name="media-skip-forward") + self.skip_prev_icon = Image(icon_name="media-skip-backward") + self.shuffle_icon = Image(icon_name="shuffle") + self.play_pause_icon = Image(icon_name="media-playback-start") + + self.play_pause_button = Button( + name="player-button", + child=self.play_pause_icon, + on_clicked=self.player.play_pause, + ) + + self.player.bind_property("can_pause", self.play_pause_button, "sensitive") + + self.next_button = Button( + style_classes=["player-button"], + child=self.skip_next_icon, + on_clicked=self._on_player_next, + ) + self.player.bind_property("can_go_next", self.next_button, "sensitive") + + self.prev_button = Button( + style_classes=["player-button"], + child=self.skip_prev_icon, + on_clicked=self._on_player_prev, + ) + self.shuffle_button = Button( + style_classes=["player-button"], + child=self.shuffle_icon, + on_clicked=self.player.toggle_shuffle, + ) + self.player.bind_property("can_shuffle", self.shuffle_button, "sensitive") + + self.button_box.children = ( + self.shuffle_button, + self.prev_button, + self.play_pause_button, + self.next_button, + ) + self.player_info_box = Box( + name="player-info-box", + v_align="center", + h_align="start", + orientation="v", + children=[self.track_info, self.controls_box, self.stack_buttons_box], + ) + + self.inner_box = Box( + name="inner-player-box", + v_align="center", + h_align="start", + ) + # resize the inner box + self.outer_box = Box( + name="outer-player-box", + h_align="start", + ) + + self.overlay_box = Overlay( + child=self.outer_box, + overlays=[ + self.inner_box, + self.player_info_box, + self.image_stack, + Box( + children=Image(icon_name=self.player.player_name, icon_size=18), + h_align="end", + v_align="start", + style="margin-top: 5px; margin-right: 10px;", + tooltip_text=self.player.player_name, # type: ignore + ), + ], + ) + + self.children = [*self.children, self.overlay_box] + + bulk_connect( + self.player, + { + "exit": self._on_player_exit, + "notify::playback-status": self._on_playback_change, + "notify::shuffle": self._on_shuffle_update, + "notify::metadata": self._on_metadata, + }, + ) + + def __del__(self): + """Ensure cleanup happens even if player exits unexpectedly.""" + try: + self._cleanup_temp_files() + except Exception: + pass # Ignore errors during cleanup in destructor + + def update_buttons(self, player_buttons, show_buttons): + """Update the stack switcher buttons in this player box""" + logger.info( + f"[PlayerBox] update_buttons called: show_buttons={ + show_buttons + }, num_buttons={len(player_buttons)}" + ) + + # Clear existing buttons + for child in self.stack_buttons_box.get_children(): + child.destroy() + + if show_buttons and len(player_buttons) > 1: + logger.info(f"[PlayerBox] Creating {len(player_buttons)} stack buttons") + # Create simple dot indicators for each player + for i, button in enumerate(player_buttons): + # Create a more visible dot button + dot_button = Button( + name="player-stack-button", + style="min-width: 16px; min-height: 16px; border-radius: 8px; margin: 4px; background-color: #666; border: 1px solid #999;", + ) + + # Set active state based on original button + if button.get_style_context().has_class("active"): + dot_button.add_style_class("active") + dot_button.set_style( + "min-width: 16px; min-height: 16px; border-radius: 8px; margin: 4px; background-color: #fff; border: 1px solid #ccc;" + ) + + # Connect click handler to switch to this player + def make_click_handler(index): + return lambda *_: self.player_stack.on_player_clicked_by_index( + index + ) + + dot_button.connect("clicked", make_click_handler(i)) + self.stack_buttons_box.children = [ + *self.stack_buttons_box.children, + dot_button, + ] + logger.info( + f"[PlayerBox] Added dot button {i}, total children: { + len(self.stack_buttons_box.get_children()) + }" + ) + + self.stack_buttons_box.show_all() + logger.info("[PlayerBox] Stack buttons box shown") + else: + self.stack_buttons_box.hide() + logger.info("[PlayerBox] Stack buttons box hidden") + + def _on_metadata(self, *_): + self._set_image() + + def _cleanup_temp_files(self): + """Clean up temporary artwork files.""" + for temp_file in self.temp_artwork_files: + try: + if os.path.exists(temp_file): + os.unlink(temp_file) + except Exception as e: + logger.warning(f"Failed to cleanup temp file {temp_file}: {e}") + self.temp_artwork_files.clear() + + def _on_player_exit(self, _, value): + self.exit = value + self._cleanup_temp_files() # Clean up temp files before destroying + self.destroy() + + def _on_player_next(self, *_): + self.player.next() + + def _on_player_prev(self, *_): + self.player.previous() + + def _on_shuffle_update(self, *_): + if self.player.shuffle is None: + return + if self.player.shuffle is True: + self.shuffle_icon.style_classes = [] + self.shuffle_icon.add_style_class("shuffle-on") + else: + self.shuffle_icon.style_classes = [] + self.shuffle_icon.add_style_class("shuffle-off") + + def _on_playback_change(self, player, status): + status = player.get_property("playback-status") + + if status == "paused": + self.play_pause_icon.set_from_icon_name("media-playback-start") + + if status == "playing": + self.play_pause_icon.set_from_icon_name("media-playback-pause") + + def _update_image(self, image_path): + if image_path and os.path.isfile(image_path): + self.album_cover.set_style(f"background-image:url('{image_path}')") + else: + self.album_cover.set_style( + f"background-image:url('{self.fallback_cover_path}')" + ) + + def _set_image(self, *_): + art_url = self.player.arturl + + parsed = urllib.parse.urlparse(art_url) + if parsed.scheme == "file": + local_arturl = urllib.parse.unquote(parsed.path) + self._update_image(local_arturl) + elif parsed.scheme in ("http", "https"): + # Cancel any existing download thread to prevent memory buildup + if self.current_download_thread is not None: + # Note: GLib.Thread doesn't have a direct cancel method, + # but we can track and ignore old downloads + pass + self.current_download_thread = GLib.Thread.new( + "download-artwork", self._download_and_set_artwork, art_url + ) + else: + self._update_image(art_url) + + def _download_and_set_artwork(self, arturl): + """ + Download the artwork from the given URL asynchronously and update the cover + using GLib.idle_add to ensure UI updates occur on the main thread. + """ + local_arturl = self.fallback_cover_path + temp_file_path = None + + try: + # Clean up old temp files first (keep only last 2 to avoid excessive cleanup) + if len(self.temp_artwork_files) > 2: + old_files = self.temp_artwork_files[:-2] + for old_file in old_files: + try: + if os.path.exists(old_file): + os.unlink(old_file) + except Exception: + pass + self.temp_artwork_files = self.temp_artwork_files[-2:] + + # Download artwork + parsed = urllib.parse.urlparse(arturl) + suffix = os.path.splitext(parsed.path)[1] or ".png" + + with urllib.request.urlopen(arturl, timeout=10) as response: # Add timeout + data = response.read() + + # Create temp file in cache directory instead of system temp + os.makedirs(CACHE_DIR, exist_ok=True) + with tempfile.NamedTemporaryFile( + delete=False, suffix=suffix, dir=CACHE_DIR + ) as temp_file: + temp_file.write(data) + temp_file_path = temp_file.name + local_arturl = temp_file_path + + # Track temp file for cleanup + if temp_file_path: + self.temp_artwork_files.append(temp_file_path) + + except Exception as e: + logger.warning(f"Failed to download artwork from {arturl}: {e}") + # Clean up failed temp file + if temp_file_path and os.path.exists(temp_file_path): + try: + os.unlink(temp_file_path) + except Exception: + pass + + GLib.idle_add(self._update_image, local_arturl) + return None diff --git a/modules/panel/components/enhanced_system_tray.py b/modules/panel/components/enhanced_system_tray.py index ac7e63b..43d0aba 100644 --- a/modules/panel/components/enhanced_system_tray.py +++ b/modules/panel/components/enhanced_system_tray.py @@ -3,13 +3,23 @@ Enhanced System Tray Icon Handling This module provides enhanced icon loading capabilities for system tray items, including fallback mechanisms for file paths and common icon locations. +It also adds dropdown menu functionality for system tray items with: + +- Dynamic menu state tracking (checkboxes, radio buttons) +- Real-time menu refresh on each right-click +- Proper toggle state visualization (✓ for checked, ● for radio selected) +- Enhanced error handling and debugging +- Support for enabled/disabled menu items """ import os from gi.repository import GdkPixbuf, Gtk +from fabric.widgets.label import Label from fabric.system_tray.widgets import SystemTrayItem +from widgets.dropdown import ModusDropdown, dropdown_divider +from widgets.mousecapture import DropDownMouseCapture # FIX: the tooltip should show application names instead of unknown @@ -148,18 +158,552 @@ def _set_tooltip(self): ) +def dropdown_option(label: str = "", on_click=None, submenu_items=None, has_native_submenu=False, native_menu_item=None): + """Create a dropdown option for system tray items with optional submenu""" + from fabric.widgets.button import Button + from fabric.widgets.centerbox import CenterBox + from fabric.widgets.box import Box + + # If this has native submenu (like NetworkManager WiFi list), create special expandable option + if has_native_submenu and native_menu_item: + from fabric.widgets.revealer import Revealer + + # State for submenu expansion + submenu_expanded = False + + def toggle_native_submenu(_): + nonlocal submenu_expanded + submenu_expanded = not submenu_expanded + submenu_revealer.set_reveal_child(submenu_expanded) + + # Update the arrow indicator + arrow_label = "▼" if submenu_expanded else "▶" + main_button.get_child().get_start_children()[0].set_label(f"{label} {arrow_label}") + + print(f"[DEBUG] Native submenu {'expanded' if submenu_expanded else 'collapsed'}") + + def show_native_menu(_): + """Show the native context menu for this item""" + try: + print(f"[DEBUG] Showing native menu for '{label}'") + + # Hide our dropdown first + from widgets.dropdown import dropdowns + for dropdown in dropdowns: + if dropdown.is_visible() and hasattr(dropdown, "hide_via_mousecapture"): + dropdown.hide_via_mousecapture() + break + + # Try to activate the native menu item + if hasattr(native_menu_item, 'activate'): + native_menu_item.activate() + elif hasattr(native_menu_item, 'handle_event'): + native_menu_item.handle_event("clicked", None, 0) + + except Exception as e: + print(f"[DEBUG] Failed to show native menu for '{label}': {e}") + + # Create main button with arrow indicator + main_button = Button( + child=CenterBox( + start_children=[ + Label(label=f"{label} ▶", h_align="start", name="dropdown-option-label"), + ], + orientation="horizontal", + h_align="fill", + h_expand=True, + v_expand=True, + ), + name="dropdown-option", + h_align="fill", + on_clicked=toggle_native_submenu, + h_expand=True, + v_expand=True, + ) + + # Create submenu with native menu option + native_submenu_items = [ + Button( + child=CenterBox( + start_children=[ + Label(label=" Show Native Menu", h_align="start", name="dropdown-option-label"), + ], + orientation="horizontal", + h_align="fill", + h_expand=True, + v_expand=True, + ), + name="dropdown-option", + h_align="fill", + on_clicked=show_native_menu, + h_expand=True, + v_expand=True, + ), + Button( + child=CenterBox( + start_children=[ + Label(label=" (Hover over tray icon for full menu)", h_align="start", name="dropdown-option-label"), + ], + orientation="horizontal", + h_align="fill", + h_expand=True, + v_expand=True, + ), + name="dropdown-option", + h_align="fill", + sensitive=False, # Disabled info text + h_expand=True, + v_expand=True, + ) + ] + + # Create revealer for submenu + submenu_revealer = Revealer( + child=Box( + children=native_submenu_items, + orientation="vertical", + name="submenu-items", + ), + reveal_child=False, + transition_type="slide-down", + transition_duration=200, + ) + + # Return container with main button and submenu + return Box( + children=[main_button, submenu_revealer], + orientation="vertical", + name="dropdown-option-with-submenu", + ) + + # If this has submenu items, create an expandable option + elif submenu_items: + from fabric.widgets.revealer import Revealer + + # State for submenu expansion + submenu_expanded = False + + def toggle_submenu(_): + nonlocal submenu_expanded + submenu_expanded = not submenu_expanded + submenu_revealer.set_reveal_child(submenu_expanded) + + # Update the arrow indicator + arrow_label = "▼" if submenu_expanded else "▶" + main_button.get_child().get_start_children()[0].set_label(f"{label} {arrow_label}") + + print(f"[DEBUG] Submenu {'expanded' if submenu_expanded else 'collapsed'}") + + # Create main button with arrow indicator + main_button = Button( + child=CenterBox( + start_children=[ + Label(label=f"{label} ▶", h_align="start", name="dropdown-option-label"), + ], + orientation="horizontal", + h_align="fill", + h_expand=True, + v_expand=True, + ), + name="dropdown-option", + h_align="fill", + on_clicked=toggle_submenu, + h_expand=True, + v_expand=True, + ) + + # Create submenu items with indentation + indented_submenu_items = [] + for item in submenu_items: + # Add indentation to submenu items + if hasattr(item, 'get_child') and hasattr(item.get_child(), 'get_start_children'): + original_label = item.get_child().get_start_children()[0] + if hasattr(original_label, 'get_label'): + original_text = original_label.get_label() + original_label.set_label(f" {original_text}") # Add indentation + indented_submenu_items.append(item) + + # Create revealer for submenu + submenu_revealer = Revealer( + child=Box( + children=indented_submenu_items, + orientation="vertical", + name="submenu-items", + ), + reveal_child=False, + transition_type="slide-down", + transition_duration=200, + ) + + # Return container with main button and submenu + return Box( + children=[main_button, submenu_revealer], + orientation="vertical", + name="dropdown-option-with-submenu", + ) + + # Regular option without submenu + def on_click_handler(_): + if on_click: + on_click() + # Hide dropdown after action + from widgets.dropdown import dropdowns + for dropdown in dropdowns: + if dropdown.is_visible() and hasattr(dropdown, "hide_via_mousecapture"): + dropdown.hide_via_mousecapture() + break + + return Button( + child=CenterBox( + start_children=[ + Label(label=label, h_align="start", name="dropdown-option-label"), + ], + orientation="horizontal", + h_align="fill", + h_expand=True, + v_expand=True, + ), + name="dropdown-option", + h_align="fill", + on_clicked=on_click_handler, + h_expand=True, + v_expand=True, + ) + + +# Note: Submenu functionality is now handled inline within dropdown_option +# using Revealer widgets for expandable menu sections + + +def extract_menu_items(menu, tray_item): + """Recursively extract menu items from a DbusmenuGtk3.Menu with state tracking""" + options = [] + + try: + if hasattr(menu, 'get_children'): + children = menu.get_children() + for child in children: + try: + # Get menu item properties + label = "" + if hasattr(child, 'property_get'): + label = child.property_get("label") or "" + elif hasattr(child, 'get_label'): + label = child.get_label() or "" + + # Skip empty labels and separators + if not label or label.strip() == "": + if hasattr(child, 'property_get') and child.property_get("type") == "separator": + options.append(dropdown_divider("")) + continue + + # Check for toggle state and other properties + toggle_type = None + toggle_state = None + enabled = True + visible = True + + if hasattr(child, 'property_get'): + toggle_type = child.property_get("toggle-type") + toggle_state = child.property_get("toggle-state") + enabled = child.property_get("enabled") + visible = child.property_get("visible") + + # Handle different property formats + if enabled is None: + enabled = True + if visible is None: + visible = True + + # Debug output for complex menu items + if toggle_type or toggle_state is not None: + print(f"[DEBUG] Menu item '{label}': toggle_type={toggle_type}, toggle_state={toggle_state}, enabled={enabled}") + + # Skip invisible items + if not visible: + continue + + # Format label based on toggle state + display_label = label.replace("_", "") # Remove mnemonics + + if toggle_type == "checkmark" and toggle_state is not None: + # Add checkmark for checked items + if toggle_state == 1 or toggle_state is True: # Checked + display_label = f"✓ {display_label}" + else: # Unchecked + display_label = f" {display_label}" + elif toggle_type == "radio" and toggle_state is not None: + # Add radio button indicator + if toggle_state == 1 or toggle_state is True: # Selected + display_label = f"● {display_label}" + else: # Not selected + display_label = f"○ {display_label}" + + # Create action for this menu item + def create_menu_action(menu_child): + def action(): + try: + if hasattr(menu_child, 'handle_event'): + menu_child.handle_event("clicked", None, 0) + elif hasattr(menu_child, 'activate'): + menu_child.activate() + else: + print(f"[DEBUG] No activation method found for menu item: {label}") + except Exception as e: + print(f"[DEBUG] Failed to activate menu item '{label}': {e}") + return action + + # Handle submenus + submenu_items = None + has_native_submenu = False + + if hasattr(child, 'get_submenu'): + submenu = child.get_submenu() + if submenu: + submenu_items = extract_menu_items(submenu, tray_item) + if submenu_items: + print(f"[DEBUG] Found submenu for '{label}' with {len(submenu_items)} items") + + # Check for potential native submenus (common patterns) + if not submenu_items and hasattr(child, 'property_get'): + # Check for indicators that this might have a native submenu + item_type = child.property_get("type") or "" + children_display = child.property_get("children-display") or "" + + # Common patterns for items that show native submenus on hover + submenu_indicators = [ + "WiFi Networks" in label, + "Available Networks" in label, + "Connect to" in label, + "Bluetooth" in label and ("Device" in label or "Connect" in label), + "Audio" in label and ("Device" in label or "Output" in label), + "VPN" in label and "Connect" in label, + children_display == "submenu", + item_type == "submenu", + # Additional patterns for common system tray items + label.endswith("...") and ("Network" in label or "Audio" in label or "Bluetooth" in label), + "Select" in label and ("Device" in label or "Network" in label), + ] + + if any(submenu_indicators): + has_native_submenu = True + print(f"[DEBUG] Detected potential native submenu for '{label}'") + + # Create dropdown option with proper state indication and submenu + option = dropdown_option( + display_label, + on_click=create_menu_action(child) if not submenu_items and not has_native_submenu else None, + submenu_items=submenu_items, + has_native_submenu=has_native_submenu, + native_menu_item=child if has_native_submenu else None + ) + + # Disable option if not enabled + if not enabled: + option.set_sensitive(False) + + options.append(option) + + except Exception as e: + print(f"[DEBUG] Error processing menu item: {e}") + continue + + except Exception: + pass + + return options + + +def get_tray_menu_options(tray_item): + """Extract menu options from the system tray item's native menu""" + options = [] + app_name = tray_item._item.title or tray_item._item.identifier or "Unknown App" + + # Create unique functions for this specific tray item to avoid closure issues + def activate_app(): + try: + tray_item._item.activate_for_event(None) + except Exception as e: + print(f"[DEBUG] Failed to activate {app_name}: {e}") + + def show_native_menu(): + try: + tray_item._item.invoke_menu_for_event(None) + except Exception as e: + print(f"[DEBUG] Failed to show native menu for {app_name}: {e}") + + try: + # Try to get the actual menu items from the tray item + menu = None + if hasattr(tray_item._item, 'menu'): + menu = tray_item._item.menu + elif hasattr(tray_item._item, 'get_menu'): + menu = tray_item._item.get_menu() + + menu_extraction_successful = False + + if menu is not None: + print(f"[DEBUG] Extracting menu for {app_name}, menu type: {type(menu)}") + # Extract actual menu items + extracted_options = extract_menu_items(menu, tray_item) + if extracted_options: + print(f"[DEBUG] Successfully extracted {len(extracted_options)} menu items for {app_name}") + # Add dividers after each option (except the last one) + for i, option in enumerate(extracted_options): + options.append(option) + # Add divider after each option except the last one + if i < len(extracted_options) - 1: + options.append(dropdown_divider("")) + + menu_extraction_successful = True + else: + print(f"[DEBUG] No menu items extracted for {app_name}") + + # Only add fallback options if menu extraction was not successful + if not menu_extraction_successful: + if menu is not None: + # Menu exists but extraction failed, show native menu option + print(f"[DEBUG] Menu exists but extraction failed for {app_name}, adding fallback") + options.append(dropdown_option( + f"Show {app_name} Menu", + on_click=show_native_menu + )) + else: + # No menu available, just add basic activation + print(f"[DEBUG] No menu available for {app_name}, adding basic options") + options.append(dropdown_option( + f"Open {app_name}", + on_click=activate_app + )) + options.append(dropdown_divider("")) + options.append(dropdown_option( + "Show Context Menu", + on_click=show_native_menu + )) + + except Exception as e: + print(f"[DEBUG] Exception while extracting menu for {app_name}: {e}") + # Fallback to basic options if anything fails + options = [ + dropdown_option( + f"Open {app_name}", + on_click=activate_app + ), + dropdown_divider(""), + dropdown_option( + "Show Context Menu", + on_click=show_native_menu + ), + ] + + return options + + +def refresh_dropdown_menu(dropdown, tray_item): + """Refresh the dropdown menu with current menu state""" + try: + # Get fresh menu options + fresh_options = get_tray_menu_options(tray_item) + + # Update the dropdown children + if hasattr(dropdown, 'child_window') and hasattr(dropdown.child_window, 'dropdown'): + dropdown.child_window.dropdown.children = fresh_options + + return True + except Exception: + return False + + +def create_system_tray_dropdown(tray_item, parent=None): + """Create a dropdown menu for a system tray item""" + dropdown_children = get_tray_menu_options(tray_item) + + # Create unique ID using both identifier and object id to ensure uniqueness + unique_id = f"systray-{tray_item._item.identifier}-{id(tray_item)}" + + dropdown = ModusDropdown( + parent=parent, + dropdown_id=unique_id, + dropdown_children=dropdown_children, + ) + + dropdown_menu = DropDownMouseCapture( + layer="bottom", + child_window=dropdown + ) + + # Store reference to tray item for refreshing + dropdown_menu._tray_item = tray_item + + # Custom positioning for system tray items + def custom_get_coords_for_widget(widget): + """Custom coordinate calculation for system tray items""" + if not ((toplevel := widget.get_toplevel()) and toplevel.is_toplevel()): + return 0, 0 + + # Get the widget's allocation and position + allocation = widget.get_allocation() + + # Try to get absolute position on screen + try: + # Get widget position relative to toplevel - this should give us the correct coordinates + widget_x, widget_y = widget.translate_coordinates(toplevel, 0, 0) or (0, 0) + + # Position the dropdown below the tray item + # Don't divide by 2 like the original method does - that's causing the positioning issue + return widget_x, widget_y + allocation.height + + except Exception: + # Fallback to basic positioning + return 0, allocation.height + + # Override the get_coords_for_widget method for this specific dropdown window + # Store the original method in case we need it + dropdown._original_get_coords_for_widget = dropdown.get_coords_for_widget + + # Replace with our custom method + def bound_custom_get_coords(widget): + return custom_get_coords_for_widget(widget) + + dropdown.get_coords_for_widget = bound_custom_get_coords + + dropdown_menu.child_window.set_pointing_to(tray_item) + + return dropdown_menu + + + +def patched_on_clicked(self, widget, event): + """Enhanced click handler that adds dropdown functionality""" + # Right click shows our custom dropdown instead of native menu + if event.button == 3: + # Always refresh the dropdown to get current menu state + parent = self.get_toplevel() + self._dropdown_capture = create_system_tray_dropdown(self, parent) + + # Toggle the dropdown (positioning is handled by custom positioning function) + self._dropdown_capture.toggle_mousecapture() + return True + + # For left clicks and other buttons, use original behavior + return original_on_clicked(self, widget, event) + + def apply_enhanced_system_tray(): - # Store original method - global original_do_update_properties + # Store original methods + global original_do_update_properties, original_on_clicked original_do_update_properties = SystemTrayItem.do_update_properties + original_on_clicked = SystemTrayItem.on_clicked # Attach helper methods to SystemTrayItem class SystemTrayItem._try_load_icon_from_path = _try_load_icon_from_path SystemTrayItem._set_tooltip = _set_tooltip - # Replace the do_update_properties method + # Replace the methods SystemTrayItem.do_update_properties = patched_do_update_properties + SystemTrayItem.on_clicked = patched_on_clicked -# Store reference to original method +# Store references to original methods original_do_update_properties = None +original_on_clicked = None diff --git a/services/mpris.py b/services/mpris.py index c5131c2..b52a807 100644 --- a/services/mpris.py +++ b/services/mpris.py @@ -1,249 +1,282 @@ +# Standard library imports +import contextlib + import gi -from gi.repository import GLib, Playerctl -from config.data import ALLOWED_PLAYERS -from fabric import Fabricator -from fabric.core.service import Service, Signal - -gi.require_version("Playerctl", "2.0") +# Fabric imports +from fabric.core.service import Property, Service, Signal +from fabric.utils import bulk_connect +from gi.repository import GLib +from loguru import logger -class PlayerService(Service): - @Signal - def shuffle_toggle(self, player: Playerctl.Player, status: bool) -> None: ... +class PlayerctlImportError(ImportError): + def __init__(self, *args): + super().__init__( + "Playerctl is not installed, please install it first", + *args, + ) + + +try: + gi.require_version("Playerctl", "2.0") + from gi.repository import Playerctl +except ValueError: + raise PlayerctlImportError + + +class MprisPlayer(Service): + """A service to manage a mpris player.""" @Signal - def meta_change(self, metadata: GLib.Variant, player: Playerctl.Player) -> None: ... + def exit(self, value: bool) -> bool: ... @Signal - def pause(self) -> None: ... + def changed(self) -> None: ... - @Signal - def play(self) -> None: ... - - @Signal - def track_position(self, pos: float, dur: float) -> None: ... - - def __init__(self, player: Playerctl.Player, **kwargs): + def __init__( + self, + player: Playerctl.Player, + **kwargs, + ): + self._signal_connectors: dict = {} self._player: Playerctl.Player = player super().__init__(**kwargs) - if player.props.player_name in ALLOWED_PLAYERS: - self._player.connect("playback-status::playing", self.on_play) - self._player.connect("playback-status::paused", self.on_pause) - self._player.connect("shuffle", self.on_shuffle) - self._player.connect("metadata", self.on_metadata) - self._player.connect("seeked", self.on_seeked) + for sn in ["playback-status", "loop-status", "shuffle"]: + self._signal_connectors[sn] = self._player.connect( + sn, + lambda *args, sn=sn: self.notifier(sn, args), + ) - self.status = self._player.props.playback_status - self.pos_fabricator = Fabricator( - interval=1000, # 1s - poll_from=lambda f, *_: self._safe_get_position(), - on_changed=lambda f, *_: self.fabricating(), + self._signal_connectors["exit"] = self._player.connect( + "exit", + self.on_player_exit, ) - # Add state tracking to prevent repeated operations - self._fabricator_running = False - self._cleanup_done = False - self.poll_progress() + self._signal_connectors["metadata"] = self._player.connect( + "metadata", + lambda *_: self.update_status(), + ) + GLib.idle_add(self.update_status_once) - def _safe_get_position(self): - """Safely get player position with error handling""" + def update_status(self): + # schedule each notifier asynchronously. + def notify_property(prop): + if self.get_property(prop) is not None: + self.notifier(prop) + + for prop in [ + "metadata", + "title", + "artist", + "arturl", + "length", + ]: + GLib.idle_add(lambda p=prop: (notify_property(p), False)) + for prop in [ + "can-seek", + "can-pause", + "can-shuffle", + "can-go-next", + "can-go-previous", + ]: + GLib.idle_add(lambda p=prop: (self.notifier(p), False)) + + def update_status_once(self): + # schedule notifier calls for each property + def notify_all(): + for prop in self.list_properties(): # type: ignore + self.notifier(prop.name) + return False + + GLib.idle_add(notify_all, priority=GLib.PRIORITY_DEFAULT_IDLE) + + def notifier(self, name: str, args=None): + def notify_and_emit(): + self.notify(name) + self.emit("changed") + return False + + GLib.idle_add(notify_and_emit, priority=GLib.PRIORITY_DEFAULT_IDLE) + + def on_player_exit(self, player): + for id in list(self._signal_connectors.values()): + with contextlib.suppress(Exception): + self._player.disconnect(id) + del self._signal_connectors + GLib.idle_add(lambda: (self.emit("exit", True), False)) + del self._player + + def toggle_shuffle(self, *_): + if self.can_shuffle: + # schedule the shuffle toggle in the GLib idle loop + GLib.idle_add(lambda: (setattr(self, "shuffle", not self.shuffle), False)) + # else do nothing + + def play_pause(self, *_): + if self.can_pause: + GLib.idle_add(lambda: (self._player.play_pause(), False)) + + def next(self, *_): + if self.can_go_next: + GLib.idle_add(lambda: (self._player.next(), False)) + + def previous(self, *_): + if self.can_go_previous: + GLib.idle_add(lambda: (self._player.previous(), False)) + + # Properties + @Property(str, "readable") + def player_name(self) -> int: + return self._player.get_property("player-name") # type: ignore + + @Property(int, "read-write", default_value=0) + def position(self) -> int: + return self._player.get_property("position") # type: ignore + + @position.setter + def position(self, new_pos: int): + self._player.set_position(new_pos) + + @Property(object, "readable") + def metadata(self) -> dict: + return self._player.get_property("metadata") # type: ignore + + @Property(str or None, "readable") + def arturl(self) -> str | None: + if "mpris:artUrl" in self.metadata.keys(): # type: ignore # noqa: SIM118 + return self.metadata["mpris:artUrl"] # type: ignore + return None + + @Property(str or None, "readable") + def length(self) -> str | None: + if "mpris:length" in self.metadata.keys(): # type: ignore # noqa: SIM118 + return self.metadata["mpris:length"] # type: ignore + return None + + @Property(str, "readable") + def artist(self) -> str: + artist = self._player.get_artist() # type: ignore + if isinstance(artist, (list, tuple)): + return ", ".join(artist) + return artist + + @Property(str, "readable") + def album(self) -> str: + return self._player.get_album() # type: ignore + + @Property(str, "readable") + def title(self): + return self._player.get_title() + + @Property(bool, "read-write", default_value=False) + def shuffle(self) -> bool: + return self._player.get_property("shuffle") # type: ignore + + @shuffle.setter + def shuffle(self, do_shuffle: bool): + self.notifier("shuffle") + return self._player.set_shuffle(do_shuffle) + + @Property(str, "readable") + def playback_status(self) -> str: + return { + Playerctl.PlaybackStatus.PAUSED: "paused", + Playerctl.PlaybackStatus.PLAYING: "playing", + Playerctl.PlaybackStatus.STOPPED: "stopped", + # type: ignore + }.get(self._player.get_property("playback_status"), "unknown") + + @Property(str, "read-write") + def loop_status(self) -> str: + return { + Playerctl.LoopStatus.NONE: "none", + Playerctl.LoopStatus.TRACK: "track", + Playerctl.LoopStatus.PLAYLIST: "playlist", + }.get( + self._player.get_property("loop_status"), "unknown" + ) # type: ignore + + @loop_status.setter + def loop_status(self, status: str): + loop_status = { + "none": Playerctl.LoopStatus.NONE, + "track": Playerctl.LoopStatus.TRACK, + "playlist": Playerctl.LoopStatus.PLAYLIST, + }.get(status) + self._player.set_loop_status(loop_status) if loop_status else None + + @Property(bool, "readable", default_value=False) + def can_go_next(self) -> bool: + return self._player.get_property("can_go_next") # type: ignore + + @Property(bool, "readable", default_value=False) + def can_go_previous(self) -> bool: + return self._player.get_property("can_go_previous") # type: ignore + + @Property(bool, "readable", default_value=False) + def can_seek(self) -> bool: + return self._player.get_property("can_seek") # type: ignore + + @Property(bool, "readable", default_value=False) + def can_pause(self) -> bool: + return self._player.get_property("can_pause") # type: ignore + + @Property(bool, "readable", default_value=False) + def can_shuffle(self) -> bool: try: - if not self._player or self._cleanup_done: - return 0 - return self._player.get_position() + self._player.set_shuffle(self._player.get_property("shuffle")) + return True except Exception: - # Player is no longer available, stop the fabricator - if ( - hasattr(self, "pos_fabricator") - and self.pos_fabricator - and self._fabricator_running - ): - try: - self.pos_fabricator.stop() - self._fabricator_running = False - except Exception: - pass - return 0 - - def cleanup(self): - """Clean up resources when service is no longer needed""" - if self._cleanup_done: - return # Already cleaned up + return False + @Property(bool, "readable", default_value=False) + def can_loop(self) -> bool: try: - if hasattr(self, "pos_fabricator") and self.pos_fabricator: - if self._fabricator_running: - self.pos_fabricator.stop() - self._fabricator_running = False - self.pos_fabricator = None - self._cleanup_done = True - except Exception as e: - print(f"[DEBUG] Error during PlayerService cleanup: {e}") - - def on_seeked(self, player, position): - if ( - not self._cleanup_done - and self.status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" - and not self._fabricator_running - ): - try: - self.pos_fabricator.start() - self._fabricator_running = True - except Exception: - pass # Ignore if already started - - def set_position(self, pos: float): - if not self._cleanup_done and self._fabricator_running: - try: - self.pos_fabricator.stop() - self._fabricator_running = False - except Exception: - pass # Ignore if already stopped - - micro_pos = int(pos * 1_000_000) - try: - self._player.set_position(micro_pos) - except GLib.Error as e: - print(f"Failed to seek: {e}") - - def poll_progress(self): - if self._cleanup_done: - return # Don't operate on cleaned up fabricator - - if self.status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING": - if not self._fabricator_running: - try: - self.pos_fabricator.start() - self._fabricator_running = True - except Exception: - pass # Ignore if already started - else: - if self._fabricator_running: - try: - self.pos_fabricator.stop() - self._fabricator_running = False - except Exception: - pass # Ignore if already stopped - - def fabricating(self): - try: - # Check if cleanup has been done - if self._cleanup_done: - return - - # Validate player object - if not self._player or not hasattr(self._player, "get_position"): - return - - # Get position safely - try: - pos = self._player.get_position() / 1_000_000 # seconds - except Exception: - # Player is no longer available, stop fabricator - if self._fabricator_running: - try: - self.pos_fabricator.stop() - self._fabricator_running = False - except Exception: - pass - return - - # Get duration safely from multiple sources - dur = 0.0 # Default to 0 - - if ( - hasattr(self._player, "props") - and hasattr(self._player.props, "metadata") - and self._player.props.metadata - ): - metadata = self._player.props.metadata - - # Try mpris:length first (most common) - try: - raw_dur = metadata["mpris:length"] - if raw_dur and raw_dur > 0: - dur = float(raw_dur) / 1_000_000 # seconds - except (KeyError, Exception): - pass - - # Fallback to xesam:duration (alternative field) - if dur <= 0: - try: - raw_dur = metadata["xesam:duration"] - if raw_dur and raw_dur > 0: - dur = float(raw_dur) / 1_000_000 # seconds - except (KeyError, Exception): - pass - - # Try to get duration from player directly (some players support this) - if dur <= 0: - try: - if hasattr(self._player, "get_metadata"): - player_metadata = self._player.get_metadata() - if player_metadata: - raw_dur = player_metadata["mpris:length"] - if raw_dur and raw_dur > 0: - dur = float(raw_dur) / 1_000_000 - except (KeyError, Exception): - pass - - # Validate values before emitting signal - # Allow dur=0 for position-only tracking - if pos is not None and dur is not None and pos >= 0 and dur >= 0: - self.track_position(pos, dur) - + self._player.set_shuffle(self._player.get_property("shuffle")) + return True except Exception: - pass - - def on_play(self, player, status): - self.status = player.props.playback_status - self.poll_progress() - self.play() - - def on_pause(self, player, status): - self.poll_progress() - self.pause() - - def on_shuffle(self, player, status): - self.shuffle_toggle(player, status) - - def on_metadata(self, player, metadata): - keys = metadata.keys() - self.meta_change(metadata, player) + return False -class PlayerManager(Service): - @Signal - def new_player(self, player: Playerctl.Player) -> Playerctl.Player: ... +class MprisPlayerManager(Service): + """A service to manage mpris players.""" @Signal - def player_vanish(self, player: Playerctl.Player) -> None: ... + def player_appeared(self, player: Playerctl.Player) -> Playerctl.Player: ... - def __init__(self, **kwargs): + @Signal + def player_vanished(self, player_name: str) -> str: ... + + def __init__( + self, + **kwargs, + ): + self._manager = Playerctl.PlayerManager.new() + bulk_connect( + self._manager, + { + "name-appeared": self.on_name_appeared, + "name-vanished": self.on_name_vanished, + }, + ) + self.add_players() super().__init__(**kwargs) - self._manager = Playerctl.PlayerManager() - self._manager.connect("name-appeared", self._on_name_appeared, self._manager) - self._manager.connect( - "player-vanished", self._on_player_vanished, self._manager - ) - self._players = {} + def on_name_appeared(self, manager, player_name: Playerctl.PlayerName): + logger.info(f"[MprisPlayer] {player_name.name} appeared") + new_player = Playerctl.Player.new_from_name(player_name) + manager.manage_player(new_player) + self.emit("player-appeared", new_player) # type: ignore - def init_all_players(self): - # invoked in the UI - for player_obj in self._manager.props.player_names: - name_str = player_obj.name - if name_str in ALLOWED_PLAYERS: - player = Playerctl.Player.new_from_name(player_obj) - self._manager.manage_player(player) - self.new_player(player) + def on_name_vanished(self, manager, player_name: Playerctl.PlayerName): + logger.info(f"[MprisPlayer] {player_name.name} vanished") + self.emit("player-vanished", player_name.name) # type: ignore - def _on_name_appeared(self, sender, name, manager): - name_str = name.name - if name_str in ALLOWED_PLAYERS: - player = Playerctl.Player.new_from_name(name) - self._manager.manage_player(player) - self.new_player(player) + def add_players(self): + # type: ignore + for player in self._manager.get_property("player-names"): + self._manager.manage_player(Playerctl.Player.new_from_name(player)) # type: ignore - def _on_player_vanished(self, sender, player, manager): - self.player_vanish(player) + @Property(object, "readable") + def players(self): + return self._manager.get_property("players") # type: ignore diff --git a/styles/player.css b/styles/player.css index add6784..4199cb7 100644 --- a/styles/player.css +++ b/styles/player.css @@ -1,128 +1,80 @@ -.player-box { - min-height: 60px; - min-width: 300px; - padding: 10px; - background-color: alpha(#000, 0.3); - box-shadow: inset 0 0 0 1px alpha(#aaa, 0.4); - border-radius: 10px; - background-position: center; - background-size: cover; +#player-stack-button { + border-radius: 5px; + min-width: 7px; + min-height: 7px; + margin: 10px 3px; } -#player-container { - min-height: 60px; - border-radius: 10px; - padding: 2px; - margin-left: 2px; - background-position: center; - background-size: cover; -} - -#player-details { - background-color: rgba(0, 0, 0, 0.5); -} - -.album-image { - min-width: 60px; - background-position: center; - background-size: cover; - border-radius: 10px; -} - -/* #artist { */ -/* color: var(--primary); */ -/* } */ - -#song { - font-size: 15px; - margin-top: 5px; -} - -#source { - margin-left: -20px; - border-radius: 10px; -} - -#play-pause-label { - color: var(--surface); - text-shadow: none; -} - -#pause-button { - /* background-color: whitesmoke; */ - border-radius: 16px; - padding: 10px; -} - -#pause-label, -#play-label { - /* color: var(--shadow); */ - font-size: 20px; - padding: 2px; - border-radius: 9px; -} - -#shuffle-button, -#next-button, -#prev-button, -#play-pause-button { - border: none; - border-radius: 12px; - padding: 0px; - padding-left: 11px; - padding-right: 11px; - min-width: 25px; - min-height: 25px; -} - -#shuffle-button:hover, -#next-button:hover, -#prev-button:hover, -#play-pause-button:hover { +#player-stack-button:hover { + background-color: #646464; box-shadow: inset rgba(255, 255, 255, 0.5) 0 0 10px; } -.player-button { +#player-stack-button.active { border-radius: 5px; - /* background-color: var(--shadow); */ - background-color: white; - opacity: 1; - min-width: 10px; - min-height: 10px; - padding: 0px; - margin: 0px; - border: none; + background-color: rgb(221.25, 221.25, 223.65); +} + +#outer-player-box { + min-height: 100px; + min-width: 363px; + border-radius: 10px; + margin: 0 6px; + background-color: alpha(#000, 0.3); + box-shadow: inset 0 0 0 1px alpha(#aaa, 0.4); +} + +#player-box:disabled highlight, +#player-box:disabled progress { + background-color: rgba(205, 214, 244, 0.6); + background-image: none; +} + +#player-info-box { + margin-left: 130px; +} + +#player-title { + margin-top: 18px; + font-weight: 700; + font-size: 16px; +} + +#player-artist, +#player-album { + font-weight: 500; + font-size: 12px; +} + +#player-controls { margin-top: 5px; + margin-bottom: 15px; } -.player-button.active { - opacity: 1; - /* min-width: 40px; */ - /* border-radius: 5px; */ +.player-icon { + font-size: 16px; } -#player-switch-container { - margin-top: 0px; - margin-left: -18px; - padding-right: 10px; +.player-button { + min-width: 20px; + min-height: 20px; + padding: 1px; + border-radius: 100%; } -#controls { - padding-bottom: 10px; - margin-top: 5px; - margin-bottom: 0px; +.player-button:hover { + box-shadow: inset rgba(255, 255, 255, 0.5) 0 0 10px; } -#player-icon { - font-size: 20px; - color: black; - text-shadow: 0 0 5px white; +#player-box #player-controls #button-box .player-button:disabled { + color: #5f5f5f; } -#shuffle { - font-size: 17px; -} - -#disable-shuffle { - font-size: 27px; +.album-image { + min-width: 100px; + min-height: 90px; + background-position: center; + background-size: cover; + margin-left: 12px; + border-radius: 10px; }