sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

✨ feat(player): improve player metadata handling and thumbnail loading with queue system

Browse Source
This commit is contained in:
sankalp committed 2025-08-03 11:49:26 +05:45
1 parent 2a90007987
commit 00ed45eda6
2 files changed
+238 -375
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

modules/controlcenter/player.py
+223 -348
View File
@@ -1,19 +1,19 @@
import urllib.parse
from urllib.parse import urlparse
import re
from gi.repository import GLib
from gi.repository import Playerctl, GLib
from fabric.core.service import Signal
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 loguru import logger
import config.data as data
from fabric.utils import get_relative_path
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.widgets.eventbox import EventBox
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.stack import Stack
from fabric.widgets.svg import Svg
from services.mpris import PlayerManager, PlayerService
@@ -22,13 +22,14 @@ class Player(Box):
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("pause", self.on_pause)
self._player.connect("play", self.on_play)
self._player.connect("meta-change", self.on_metadata)
self._player.connect("shuffle-toggle", self.on_shuffle)
self._player.connect("track-position", self.on_update_track_position)
player.connect("metadata", self.on_direct_metadata_change)
@@ -46,7 +47,7 @@ class Player(Box):
justification="left",
h_align="start",
ellipsization="end",
max_chars_width=15,
max_chars_width=22,
)
self.artist = Label(
name="artist",
@@ -87,6 +88,7 @@ class Player(Box):
f"background-image:url('{data.HOME_DIR}/.current.wall')"
)
self.children = [
self.album_cover,
Box(
@@ -199,16 +201,53 @@ class Player(Box):
return
self.duration = dur
def on_seek(self, sender, ratio):
pos = ratio * self.duration # duration in seconds
print(f"Seeking to {pos:.2f}s")
self._player.set_position(int(pos))
def skip_forward(self, seconds=10):
self._player._player.seek(seconds * 1000000)
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)
def skip_backward(self, seconds=10):
self._player._player.seek(-1 * seconds * 1000000)
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:
@@ -217,300 +256,152 @@ class Player(Box):
except Exception:
return False
def on_metadata(self, sender, metadata, player):
keys = metadata.keys()
if "xesam:artist" in keys and "xesam:title" in keys:
song_title = metadata["xesam:title"]
self.song.set_label(song_title)
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()
artist_list = metadata["xesam:artist"]
artist_name = artist_list[0] if artist_list else "Unknown Artist"
self.artist.set_label(artist_name)
def _process_queue(self):
"""Process the update queue sequentially."""
if not self._update_queue or self._is_updating:
return
self._handle_thumbnail(metadata, player)
self._is_updating = True
update_func, args, kwargs = self._update_queue.pop(0)
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)
if (
player.props.playback_status.value_name
== "PLAYERCTL_PLAYBACK_STATUS_PLAYING"
):
self.on_play(self._player)
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)
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):
keys = metadata.keys()
player_name = player.props.player_name
browser_players = ["firefox", "chromium", "vivaldi", "brave"]
is_browser = player_name in browser_players
thumbnail_url = None
if "mpris:artUrl" in keys:
art_url = metadata["mpris:artUrl"]
if art_url and art_url.strip():
if art_url.startswith("data:image/"):
thumbnail_url = art_url
elif art_url.startswith("http"):
thumbnail_url = art_url
elif art_url.startswith("file://"):
thumbnail_url = art_url
if is_browser and not thumbnail_url and "xesam:url" in keys:
webpage_url = metadata["xesam:url"]
if webpage_url and webpage_url.startswith("http"):
# Try to extract actual media thumbnail from the webpage
thumbnail_url = self._extract_media_thumbnail(
webpage_url, metadata, player_name
)
if is_browser and not thumbnail_url:
thumbnail_url = self._extract_from_browser_metadata(metadata, player_name)
if thumbnail_url:
logger.debug(f"Setting media thumbnail for {player_name}: {thumbnail_url}")
self.album_cover.set_style(f"background-image:url('{thumbnail_url}')")
else:
logger.debug(
f"No media thumbnail found for {player_name}, using default wallpaper"
)
self.album_cover.set_style(
f"background-image:url('{data.HOME_DIR}/.current.wall')"
)
def _extract_media_thumbnail(self, webpage_url, metadata, player_name):
try:
parsed_url = urlparse(webpage_url)
domain = parsed_url.netloc
path = parsed_url.path
logger.debug(
f"Extracting media thumbnail for {player_name} from: {webpage_url}"
)
if "youtube.com" in domain or "youtu.be" in domain:
video_id = self._extract_youtube_video_id(webpage_url)
if video_id:
logger.debug(f"Extracted YouTube video ID: {video_id}")
return f"https://img.youtube.com/vi/{video_id}/maxresdefault.jpg"
elif "vimeo.com" in domain:
video_id = self._extract_vimeo_video_id(webpage_url)
if video_id:
logger.debug(f"Extracted Vimeo video ID: {video_id}")
return f"https://vumbnail.com/{video_id}.jpg"
elif "twitch.tv" in domain:
channel_name = self._extract_twitch_channel(webpage_url)
if channel_name:
logger.debug(f"Extracted Twitch channel: {channel_name}")
return f"https://static-cdn.jtvnw.net/previews-ttv/live_user_{channel_name}-1280x720.jpg"
return self._extract_from_metadata_fields(metadata, player_name)
except Exception as e:
logger.warning(f"Failed to extract media thumbnail: {e}")
return None
def _extract_youtube_video_id(self, url):
try:
if "youtube.com/watch" in url:
query = urlparse(url).query
params = dict(urllib.parse.parse_qsl(query))
return params.get("v")
elif "youtu.be/" in url:
path = urlparse(url).path
return path.strip("/")
elif "youtube.com/embed/" in url:
path = urlparse(url).path
return path.split("/")[-1]
except Exception as e:
logger.warning(f"Failed to extract YouTube video ID: {e}")
return None
def _extract_vimeo_video_id(self, url):
try:
path = urlparse(url).path
video_id = path.strip("/")
if video_id.isdigit():
return video_id
except Exception as e:
logger.warning(f"Failed to extract Vimeo video ID: {e}")
return None
def _extract_twitch_channel(self, url):
try:
path = urlparse(url).path
channel = path.strip("/")
if channel and not channel.startswith("videos"):
return channel
except Exception as e:
logger.warning(f"Failed to extract Twitch channel: {e}")
return None
def _extract_from_browser_metadata(self, metadata, player_name):
try:
"""Queue thumbnail update to prevent overlapping operations."""
def _update_thumbnail():
keys = metadata.keys()
player_name = player.props.player_name
thumbnail_fields = [
"mpris:artUrl",
"xesam:artUrl",
"xesam:albumArt",
"xesam:thumbnail",
"xesam:image",
"xesam:cover",
]
new_thumbnail = None
for field in thumbnail_fields:
if field in keys:
value = metadata[field]
if value and isinstance(value, str) and value.strip():
if value.startswith(("http://", "https://", "data:image/")):
logger.debug(f"Found thumbnail in {field}: {value}")
return value
# 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}")
return None
# 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}")
except Exception as e:
logger.warning(f"Failed to extract from browser metadata: {e}")
return None
# 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}")
def _extract_from_metadata_fields(self, metadata, player_name):
try:
keys = metadata.keys()
self._queue_update(_update_thumbnail)
thumbnail_fields = [
"mpris:artUrl",
"xesam:artUrl",
"xesam:albumArt",
"xesam:thumbnail",
"xesam:image",
"xesam:cover",
"xesam:albumArtUrl",
]
for field in thumbnail_fields:
if field in keys:
value = metadata[field]
if value and isinstance(value, str) and value.strip():
if value.startswith(
("http://", "https://", "data:image/", "file://")
):
logger.debug(f"Found thumbnail in {field}: {value}")
return value
return None
except Exception as e:
logger.warning(f"Failed to extract from metadata fields: {e}")
return None
def on_pause(self, sender):
self._update_play_pause_icon(False)
def on_play(self, sender):
self._update_play_pause_icon(True)
def on_shuffle(self, sender, player, status):
print("callback status", status)
self._update_shuffle_icon(status)
def handle_next(self, player):
self._player._player.next()
self._refresh_metadata_after_delay(player)
if player.props.player_name in ["firefox", "chromium", "vivaldi", "brave"]:
self._refresh_metadata_immediate(player)
def _do_next():
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):
self._player._player.previous()
self._refresh_metadata_after_delay(player)
if player.props.player_name in ["firefox", "chromium", "vivaldi", "brave"]:
self._refresh_metadata_immediate(player)
def _refresh_metadata_immediate(self, player):
try:
current_metadata = player.props.metadata
if current_metadata:
logger.debug(
f"Immediate metadata refresh for {player.props.player_name}"
)
self.on_metadata(self._player, current_metadata, player)
except Exception as e:
logger.warning(f"Failed immediate metadata refresh: {e}")
def _refresh_metadata_after_delay(self, player):
def refresh_metadata():
def _do_prev():
try:
current_metadata = player.props.metadata
if current_metadata:
logger.debug(f"Refreshing metadata for {player.props.player_name}")
self.on_metadata(self._player, current_metadata, player)
else:
logger.debug(
f"No metadata available for {
player.props.player_name
}, retrying..."
)
GLib.timeout_add(1000, refresh_metadata)
self._player._player.previous()
except Exception as e:
logger.warning(f"Failed to refresh metadata: {e}")
logger.warning(f"Failed to go to previous track: {e}")
self._queue_update(_do_prev)
GLib.timeout_add(500, refresh_metadata)
def handle_play_pause(self, player):
is_playing = (
self._player._player.props.playback_status.value_name
== "PLAYERCTL_PLAYBACK_STATUS_PLAYING"
)
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_play_ui():
self._update_play_pause_icon(True)
def _set_pause_ui():
self._update_play_pause_icon(False)
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)
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):
print("shuffle", player.props.shuffle)
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
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
shuffle_button.get_child().set_style("color: var(--outline)")
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)
class Placheholder(Box):
class Placeholder(Box):
def __init__(self, **kwargs):
super().__init__(style_classes="player-box", **kwargs)
@@ -550,14 +441,18 @@ class Placheholder(Box):
class PlayerContainer(Box):
@Signal
def active_player_changed(self, player: Playerctl.Player) -> None: ...
def __init__(self, **kwargs):
super().__init__(name="player-container", **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 = Placheholder()
self.placeholder = Placeholder()
self.stack = Stack(
name="player-container",
transition_type="crossfade",
transition_duration=100,
children=[self.placeholder],
@@ -575,33 +470,23 @@ class PlayerContainer(Box):
orientation="v",
center_children=[],
)
self.tile_icon = Image(
name="disc",
style="font-size:30px; margin:0px; padding:0px;",
icon_name="disc-symbolic",
size=30,
)
self.tile_view = CenterBox(
v_expand=False,
center_children=self.tile_icon,
)
self.event_box = EventBox(
events=["scroll"],
child=Box(children=[self.stack, self.player_switch_container]),
)
self.children = self.event_box
self.players = []
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()
self.event_box.connect("scroll-event", self.on_scroll)
self.stack.connect("notify::visible-child", self.on_visible_child_changed)
def new_player(self, manager, player):
new_player = Player(player=player)
new_player.set_name(player.props.player_name)
self.players.append(new_player)
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.players) == 1:
if len(self.player) == 1:
self.stack.remove(self.placeholder)
self.player_switch_container.add_center(
@@ -611,6 +496,12 @@ class PlayerContainer(Box):
on_clicked=lambda b: self.switch_player(player.props.player_name, b),
)
)
# If this is the first player, make it active
if not self.active_player:
self.active_player = player
self.active_player_changed(player)
self.update_player_list()
def switch_player(self, player_name, button):
@@ -620,63 +511,47 @@ class PlayerContainer(Box):
btn.remove_style_class("active")
button.add_style_class("active")
# 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)
def on_player_vanish(self, manager, player):
for player_instance in self.players:
if player_instance.get_name() == player.props.player_name:
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.players.remove(player_instance)
self.player.remove(player_instance)
for btn in self.player_switch_container.center_children:
if btn.get_name() == player_instance.get_name():
if btn.get_name() == player_name:
self.player_switch_container.remove_center(btn)
self.update_player_list()
break
if len(self.players) == 0:
# Clean up player objects dictionary
if player_name in self.player_objects:
del self.player_objects[player_name]
# 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.update_player_list()
if len(self.player) == 0:
self.stack.add_named(self.placeholder, "placeholder")
def update_player_list(self):
curr = self.stack.get_visible_child()
if curr is None:
return
for btn in self.player_switch_container.center_children:
print(btn.get_name())
if btn.get_name() == curr.get_name():
btn.add_style_class("active")
else:
btn.remove_style_class("active")
def on_scroll(self, widget, event):
match event.direction:
case 0:
self.switch_relative_player(forward=False)
case 1:
self.switch_relative_player(forward=True)
def switch_relative_player(self, forward=True):
if not self.players:
return
current_player = self.stack.get_visible_child()
if current_player is None:
return
current_index = self.players.index(current_player)
next_index = (current_index + (1 if forward else -1)) % len(self.players)
next_player = self.players[next_index]
for btn in self.player_switch_container.center_children:
if btn.get_name() == next_player.get_name():
self.switch_player(next_player.get_name(), btn)
break
def on_visible_child_changed(self, *args):
curr_child = self.stack.get_visible_child()
if curr_child is None:
return
curr_player = curr_child.get_name()
self.tile_icon.set_name(curr_player)
self.tile_icon.set_icon_name(self._get_player_icon_name(curr_player))
def get_view(self):
return self.tile_view
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")
styles/player.css
+15 -27
View File
@@ -1,11 +1,9 @@
.player-box {
min-height: 60px;
min-width: 300px;
padding: 1px;
box-shadow:
inset 0 0 100px rgba(148, 148, 148, 0.3),
inset 70px 0 20px rgba(255, 255, 255, 0.3),
inset 0px 0px 20px 5px rgba(255, 255, 255, 0.3);
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;
@@ -14,7 +12,7 @@
#player-container {
min-height: 60px;
border-radius: 10px;
padding: 4px;
padding: 2px;
margin-left: 2px;
background-position: center;
background-size: cover;
@@ -31,12 +29,11 @@
border-radius: 10px;
}
#artist {
color: var(--primary);
}
/* #artist { */
/* color: var(--primary); */
/* } */
#song {
color: var(--surface);
font-size: 15px;
margin-top: 5px;
}
@@ -52,27 +49,17 @@
}
#pause-button {
background-color: whitesmoke;
/* background-color: whitesmoke; */
border-radius: 16px;
padding: 10px;
}
#pause-label,
#play-label {
color: var(--shadow);
/* color: var(--shadow); */
font-size: 20px;
padding: 2px;
border-radius: 999px;
min-width: 25px;
}
#pause-label.mini,
#play-label.mini {
color: white;
font-size: 20px;
padding: 2px;
border-radius: 999px;
min-width: 25px;
border-radius: 9px;
}
#shuffle-button,
@@ -80,24 +67,25 @@
#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 {
border-radius: 100%;
box-shadow: inset rgba(255, 255, 255, 0.5) 0 0 10px;
}
.player-button {
border-radius: 5px;
background-color: var(--shadow);
/* background-color: white; */
/* background-color: var(--shadow); */
background-color: white;
opacity: 1;
min-width: 10px;
min-height: 10px;