diff --git a/config/data.py b/config/data.py
index d4c5e8e..0df7c23 100644
--- a/config/data.py
+++ b/config/data.py
@@ -10,6 +10,7 @@ gi.require_version("Gtk", "3.0")
APP_NAME = "modus"
APP_NAME_CAP = "Modus"
+ALLOWED_PLAYERS = ["vlc", "mpv", "cmus", "firefox", "spotify", "chromium"]
NOTIF_POS_KEY = "notif_pos"
NOTIF_POS_DEFAULT = "Top"
@@ -80,6 +81,7 @@ if os.path.exists(CONFIG_FILE):
"date_time": config.get("dock_date_time_visible", True),
"controls": config.get("dock_controls_visible", True),
"indicators": config.get("dock_indicators_visible", True),
+ "music_player": config.get("dock_music_player_visible", True),
"notifications": config.get("dock_notifications_visible", True),
"systray": config.get("dock_tray_visible", True),
"applications": config.get("dock_applications_visible", True),
@@ -112,6 +114,7 @@ else:
"date_time": True,
"controls": True,
"indicators": True,
+ "music_player": True,
"notifications": True,
"systray": True,
"applications": True,
diff --git a/config/hypr/colors.conf b/config/hypr/colors.conf
index ff92bcb..7697e90 100644
--- a/config/hypr/colors.conf
+++ b/config/hypr/colors.conf
@@ -1,13 +1,13 @@
-$wallpaper = /home/sankalp/Pictures/wallpapers/-13.webp
-$background = 191114
-$foreground = eedfe3
+$wallpaper = /home/sankalp/Pictures/wallpapers/110.jpg
+$background = 171216
+$foreground = ebdfe6
-$primary = fdb0d4
-$secondary = e0bdcb
-$tertiary = f2bb9a
-$surface = 191114
-$surface_bright = 40373a
-$outline = 9d8d92
+$primary = eeb4ea
+$secondary = d9bfd4
+$tertiary = f6b8aa
+$surface = 171216
+$surface_bright = 3e373c
+$outline = 998d96
$error = ff5945
$shadow = 000000
@@ -16,12 +16,12 @@ $red = ffb2b9
$green = b7d085
$yellow = dec56e
$blue = cebdfe
-$magenta = f7b1de
+$magenta = efb3e8
$cyan = 82d3e2
$white = 82d3e0
$red_dim = ff4c5c
$green_dim = 89ad42
$yellow_dim = bb9b2b
$blue_dim = 8359fc
-$magenta_dim = ee54b7
+$magenta_dim = dd5fcf
$cyan_dim = 30b5ce
diff --git a/config/settings/constants.py b/config/settings/constants.py
index 1e2214b..b5325bc 100644
--- a/config/settings/constants.py
+++ b/config/settings/constants.py
@@ -54,6 +54,7 @@ DEFAULTS = {
"dock_date_time_visible": True,
"dock_controls_visible": True,
"dock_indicators_visible": True,
+ "dock_music_player_visible": True,
"dock_notifications_visible": True,
"dock_applications_visible": True,
"dock_language_visible": True,
diff --git a/config/settings/gui.py b/config/settings/gui.py
index 1660fe2..e8f3133 100644
--- a/config/settings/gui.py
+++ b/config/settings/gui.py
@@ -523,6 +523,7 @@ class HyprConfGUI(Window):
"battery": "Battery Indicator",
"controls": "Control Panel",
"indicators": "Indicators",
+ "music_player": "Music Player",
"applications": "Taskbar",
"language": "Language Indicator",
}
diff --git a/example_player.py b/example_player.py
new file mode 100644
index 0000000..4fd3652
--- /dev/null
+++ b/example_player.py
@@ -0,0 +1,357 @@
+from fabric.widgets.box import Box
+from fabric.widgets.label import Label
+from fabric.widgets.centerbox import CenterBox
+from fabric.widgets.button import Button
+from fabric.widgets.stack import Stack
+
+from player_service import PlayerManager, PlayerService
+from wiggle_bar import WigglyWidget
+
+import icons as icons
+import info as info
+
+from loguru import logger
+
+
+class Player(Box):
+ def __init__(self, player, **kwargs):
+ super().__init__(style_classes="player", orientation="v", **kwargs)
+
+ if not info.VERTICAL:
+ # vertical class binding from unknown source
+ self.remove_style_class("vertical")
+
+ self._player = PlayerService(player=player)
+
+ 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)
+
+ self.player_name = Label(
+ name=player.props.player_name,
+ style_classes="player-icon",
+ markup=getattr(icons, player.props.player_name, icons.disc),
+ )
+
+ self.set_style(
+ f"background-image:url('{info.HOME_DIR}/.cache/walls/low_rez.png')"
+ )
+
+ self.song = Label(
+ name="song",
+ label="song",
+ justification="left",
+ h_align="start",
+ max_chars_width=10,
+ )
+ self.artist = Label(
+ name="artist", label="artist", justification="left", h_align="start"
+ )
+ self.music = Box(
+ name="music",
+ orientation="v",
+ h_expand=True,
+ v_expand=True,
+ children=[self.song, self.artist],
+ )
+
+ self.play_pause_button = Button(
+ name="pause-button",
+ child=Label(name="pause-label", markup=icons.play),
+ style_classes="pause-track",
+ tooltip_text="Exit",
+ on_clicked=lambda b, *_: self.handle_play_pause(player),
+ )
+
+ self.shuffle_button = Button(
+ name="shuffle-button",
+ child=Label(name="shuffle", markup=icons.shuffle),
+ on_clicked=lambda b, *_: self.handle_shuffle(b, player),
+ )
+
+ self.wiggly = WigglyWidget()
+ self.wiggly.connect("on-seek", self.on_seek)
+ self.gtk_wrapper = Box(
+ orientation="v",
+ h_expand=True,
+ v_expand=True,
+ h_align="fill",
+ v_align="fill",
+ children=self.wiggly,
+ )
+
+ self.children = [
+ Box(name="source", h_expand=True, v_expand=True, children=self.player_name),
+ CenterBox(
+ name="details",
+ # h_expand=True,
+ # v_expand=True,
+ start_children=self.music,
+ end_children=self.play_pause_button if not info.VERTICAL else [],
+ ),
+ Box(
+ name="controls",
+ # h_expand=True,
+ # v_expand=True,
+ style_classes="horizontal" if not info.VERTICAL else "vertical",
+ spacing=5,
+ children=[
+ Button(
+ name="prev-button",
+ child=Label(name="play-previous", markup=icons.previous),
+ on_clicked=lambda b, *_: self.handle_prev(player),
+ ),
+ CenterBox(
+ name="progress-container",
+ h_expand=True,
+ v_expand=True,
+ orientation="v",
+ center_children=[self.gtk_wrapper],
+ ),
+ Button(
+ name="next-button",
+ child=Label(name="play-next", markup=icons.next),
+ on_clicked=lambda b, *_: self.handle_next(player),
+ ),
+ self.shuffle_button,
+ ]
+ if not info.VERTICAL
+ else [
+ CenterBox(
+ orientation="v",
+ h_expand=True,
+ start_children=[
+ CenterBox(
+ h_expand=True,
+ v_expand=True,
+ v_align="end",
+ start_children=[
+ Button(
+ name="prev-button",
+ child=Label(
+ name="play-previous", markup=icons.previous
+ ),
+ on_clicked=lambda b, *_: self.handle_prev(
+ player
+ ),
+ ),
+ Button(
+ name="next-button",
+ child=Label(
+ name="play-next", markup=icons.next
+ ),
+ on_clicked=lambda b, *_: self.handle_next(
+ player
+ ),
+ ),
+ self.shuffle_button,
+ ],
+ end_children=self.play_pause_button,
+ )
+ ],
+ end_children=CenterBox(
+ name="progress-container",
+ h_expand=True,
+ v_expand=True,
+ orientation="v",
+ center_children=[self.gtk_wrapper],
+ ),
+ )
+ ],
+ ),
+ ]
+
+ self.on_metadata(self._player, metadata=player.props.metadata, player=player)
+
+ def on_update_track_position(self, sender, pos, dur):
+ self.duration = dur
+ self.wiggly.update_value_from_signal(pos / 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 on_metadata(self, sender, metadata, player):
+ keys = metadata.keys()
+ if "xesam:artist" in keys and "xesam:title" in keys:
+ _max_chars = 43 if not info.VERTICAL else 30
+ song_title = metadata["xesam:title"]
+ if len(song_title) > _max_chars:
+ song_title = song_title[: _max_chars - 1] + "…"
+ self.song.set_label(song_title)
+
+ artist_list = metadata["xesam:artist"]
+ artist_name = artist_list[0] if artist_list else "Unknown Artist"
+ if len(artist_name) > _max_chars:
+ artist_name = artist_name[: _max_chars - 1] + "…"
+ self.artist.set_label(artist_name)
+ if "mpris:artUrl" in keys:
+ self.set_style(f"background-image:url('{metadata['mpris:artUrl']}')")
+
+ if (
+ player.props.playback_status.value_name
+ == "PLAYERCTL_PLAYBACK_STATUS_PLAYING"
+ ):
+ self.on_play(self._player)
+
+ if player.props.shuffle == True:
+ self.shuffle_button.get_child().set_markup(icons.disable_shuffle)
+ self.shuffle_button.get_child().set_name("disable-shuffle")
+ else:
+ self.shuffle_button.get_child().set_markup(icons.shuffle)
+ self.shuffle_button.get_child().set_name("shuffle")
+
+ def on_pause(self, sender):
+ self.play_pause_button.get_child().set_markup(icons.play)
+ self.play_pause_button.get_child().set_name("pause-label")
+ self.wiggly.dragging = True
+ self.wiggly.update_amplitude(True)
+ self.wiggly.pause = True
+ self.play_pause_button.add_style_class("pause-track")
+
+ def on_play(self, sender):
+ self.play_pause_button.get_child().set_markup(icons.pause)
+ self.play_pause_button.get_child().set_name("play-label")
+ self.wiggly.pause = False
+ self.wiggly.dragging = False
+ self.wiggly.update_amplitude(False)
+ self.play_pause_button.remove_style_class("pause-track")
+
+ def on_shuffle(self, sender, player, status):
+ print("callback status", status)
+ if status == False:
+ self.shuffle_button.get_child().set_markup(icons.shuffle)
+ self.shuffle_button.get_child().set_name("shuffle")
+ else:
+ self.shuffle_button.get_child().set_markup(icons.disable_shuffle)
+ self.shuffle_button.get_child().set_name("disable-shuffle")
+
+ self.shuffle_button.get_child().set_style("color: white")
+
+ def handle_next(self, player):
+ player.next()
+
+ def handle_prev(self, player):
+ player.previous()
+
+ def handle_play_pause(self, player):
+ is_playing = self.play_pause_button.get_child().get_name() == "play-label"
+
+ def _set_play_ui():
+ self.play_pause_button.get_child().set_markup(icons.pause)
+ self.play_pause_button.remove_style_class("pause-track")
+ self.play_pause_button.get_child().set_name("pause-label")
+
+ def _set_pause_ui():
+ self.play_pause_button.get_child().set_markup(icons.play)
+ self.play_pause_button.add_style_class("pause-track")
+ self.play_pause_button.get_child().set_name("play-label")
+
+ if is_playing:
+ _set_pause_ui()
+ else:
+ _set_play_ui()
+
+ try:
+ player.play_pause()
+ except Exception as e:
+ # revert if signal failed
+ if is_playing:
+ _set_pause_ui()
+ else:
+ _set_play_ui()
+ logger.warning("Failed to toggle playback:", e)
+
+ def handle_shuffle(self, shuffle_button, player):
+ print("shuffle", player.props.shuffle)
+ if player.props.shuffle == False:
+ player.set_shuffle(True)
+ print("setting to true", player.props.player_name)
+ else:
+ player.set_shuffle(False)
+ shuffle_button.get_child().set_style("color: var(--outline)")
+
+
+class PlayerContainer(Box):
+ 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.stack = Stack(
+ name="player-container",
+ transition_type="crossfade",
+ transition_duration=100,
+ children=[],
+ )
+
+ self.player_stack = Stack(
+ name="player-stack",
+ transition_type="crossfade",
+ transition_duration=100,
+ children=[],
+ )
+
+ self.player_switch_container = CenterBox(
+ name="player-switch-container",
+ orientation="h",
+ style_classes="horizontal-player"
+ if not info.VERTICAL
+ else "vertical-player",
+ center_children=[],
+ )
+ self.children = [self.stack, self.player_switch_container]
+ self.player = []
+ self.manager.init_all_players()
+
+ def new_player(self, manager, player):
+ print(player.props.player_name, "new player")
+ print(player)
+ new_player = Player(player=player)
+ new_player.gtk_wrapper.queue_draw()
+ new_player.set_name(player.props.player_name)
+ self.player.append(new_player)
+ print("stacking dis bitvch")
+ self.stack.add_named(new_player, player.props.player_name)
+
+ 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),
+ )
+ )
+ self.update_player_list()
+
+ def switch_player(self, player_name, button):
+ self.stack.set_visible_child_name(player_name)
+
+ for btn in self.player_switch_container.center_children:
+ btn.remove_style_class("active")
+ button.add_style_class("active")
+
+ def on_player_vanish(self, manager, player):
+ for player_instance in self.player:
+ if player_instance.get_name() == player.props.player_name:
+ self.stack.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():
+ self.player_switch_container.remove_center(btn)
+ self.update_player_list()
+ break
+
+ def update_player_list(self):
+ curr = self.stack.get_visible_child()
+ 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")
diff --git a/install.sh b/install.sh
index 062dc56..21a4741 100755
--- a/install.sh
+++ b/install.sh
@@ -49,6 +49,7 @@ PACKAGES=(
python-ijson
python-watchdog
python-pyotp
+ python-pyzbar
ttf-nerd-fonts-symbols-mono
swww
swappy
diff --git a/main.css b/main.css
index ceb5766..4d53f78 100644
--- a/main.css
+++ b/main.css
@@ -10,6 +10,7 @@
@import url("./styles/kanban.css");
@import url("./styles/calendar.css");
@import url("./styles/notification.css");
+@import url("./styles/player.css");
* {
all: unset;
diff --git a/modules/dock/components/__init__.py b/modules/dock/components/__init__.py
index 20c9a1a..d2bbbb8 100644
--- a/modules/dock/components/__init__.py
+++ b/modules/dock/components/__init__.py
@@ -18,6 +18,7 @@ from .battery import Battery
from .controls import Controls
from .indicators import Indicators
from .metrics import Metrics
+from .music_player import MusicPlayer
from .workspaces import workspace
import gi
@@ -110,6 +111,7 @@ class DockComponents(Box):
)
self.controls = Controls()
self.indicators = Indicators()
+ self.music_player = MusicPlayer()
self.systray = SystemTray(
icon_size=18,
spacing=4,
@@ -122,6 +124,7 @@ class DockComponents(Box):
self.workspaces,
self.metrics,
self.controls,
+ self.music_player,
self.applications,
self.indicators,
self.battery,
@@ -155,6 +158,7 @@ class DockComponents(Box):
"date_time": self.date_time,
"controls": self.controls,
"indicators": self.indicators,
+ "music_player": self.music_player,
"systray": self.systray,
"applications": self.applications,
"language": self.language,
@@ -186,6 +190,7 @@ class DockComponents(Box):
"date_time": self.date_time,
"controls": self.controls,
"indicators": self.indicators,
+ "music_player": self.music_player,
"systray": self.systray,
"applications": self.applications,
"language": self.language,
diff --git a/modules/dock/components/music_player.py b/modules/dock/components/music_player.py
new file mode 100644
index 0000000..03e3a7f
--- /dev/null
+++ b/modules/dock/components/music_player.py
@@ -0,0 +1,333 @@
+from fabric.widgets.box import Box
+from fabric.widgets.button import Button
+from fabric.widgets.circularprogressbar import CircularProgressBar
+from fabric.widgets.eventbox import EventBox
+from fabric.widgets.label import Label
+from fabric.widgets.overlay import Overlay
+from gi.repository import Gdk
+
+import config.data as data
+import utils.icons as icons
+from services.mpris import PlayerManager, PlayerService
+from utils.wayland import WaylandWindow as Window
+from player import PlayerContainer
+
+
+class MusicPlayerPopup(Window):
+ """Popup window that shows the full music player interface"""
+
+ def __init__(self, **kwargs):
+ # Get dock position for proper anchoring
+ dock_position = data.DOCK_POSITION
+ popup_anchor = self._get_popup_anchor(dock_position)
+ margin = self._get_popup_margin(dock_position)
+
+ super().__init__(
+ name="music-player-popup",
+ layer="top",
+ anchor=popup_anchor,
+ margin=margin,
+ exclusive=False,
+ keyboard_mode="on-demand",
+ visible=False,
+ all_visible=False,
+ **kwargs,
+ )
+
+ self.player_container = PlayerContainer()
+ self.add(self.player_container)
+
+ # Connect to escape key to close
+ self.connect("key-press-event", self.on_key_press)
+ self.connect("button-press-event", self.on_button_press)
+ self.set_can_focus(True)
+
+ def on_key_press(self, widget, event):
+ """Handle key press events"""
+ if event.keyval == Gdk.KEY_Escape:
+ self.hide_popup()
+ return True
+ return False
+
+ def show_popup(self):
+ """Show the music player popup"""
+ self.set_visible(True)
+ self.grab_focus()
+
+ def _get_popup_anchor(self, dock_position):
+ """Get popup anchor based on dock position"""
+ anchor_map = {
+ "Top": "top",
+ "Bottom": "bottom",
+ "Left": "left",
+ "Right": "right",
+ }
+ return anchor_map.get(dock_position, "bottom")
+
+ def _get_popup_margin(self, dock_position):
+ """Get popup margin based on dock position"""
+ margin_map = {
+ "Top": "60px 10px 10px 10px",
+ "Bottom": "10px 10px 60px 10px",
+ "Left": "10px 10px 10px 60px",
+ "Right": "10px 60px 10px 10px",
+ }
+ return margin_map.get(dock_position, "10px 10px 60px 10px")
+
+ def on_button_press(self, widget, event):
+ """Handle button press events"""
+ return False
+
+ def hide_popup(self):
+ """Hide the music player popup"""
+ self.set_visible(False)
+
+
+class MusicPlayer(Box):
+ """Music player component for the dock with circular progress bar"""
+
+ def __init__(self, **kwargs):
+ super().__init__(name="music-player", **kwargs)
+
+ # Initialize player manager
+ self.manager = PlayerManager()
+ self.manager.connect("new-player", self.on_new_player)
+ self.manager.connect("player-vanish", self.on_player_vanish)
+
+ # Current player tracking
+ self.current_player = None
+ self.current_player_service = None
+ self.duration = 0.0
+ self.position = 0.0
+ self._current_track_info = "Music Player"
+
+ # Create circular progress bar
+ self.progress_bar = CircularProgressBar(
+ name="player-circle",
+ size=28,
+ line_width=2,
+ start_angle=150,
+ end_angle=390,
+ value=0.0, # Initialize with 0 progress
+ )
+
+ # Create music icon label - always use generic music icon
+ self.music_label = Label(name="music-label", markup=icons.music)
+
+ # Create button for the music icon
+ self.music_button = Button(
+ on_clicked=self.on_music_clicked, child=self.music_label
+ )
+
+ # Create event box for scroll events (volume control)
+ self.event_box = EventBox(
+ events=["scroll", "smooth-scroll"],
+ child=Overlay(child=self.progress_bar, overlays=self.music_button),
+ )
+
+ # Connect scroll events for seeking
+ self.event_box.connect("scroll-event", self.on_scroll)
+ self.add_events(Gdk.EventMask.SCROLL_MASK | Gdk.EventMask.SMOOTH_SCROLL_MASK)
+
+ self.add(self.event_box)
+
+ # Popup window will be created on-demand
+ self.popup = None
+
+ # Initialize players
+ self.manager.init_all_players()
+
+ # Set initial visibility
+ self.update_visibility()
+
+ # Connect to destroy signal for cleanup
+ self.connect("destroy", self.on_destroy)
+
+ def on_destroy(self, widget):
+ """Clean up popup when component is destroyed"""
+ if self.popup:
+ self.popup.destroy()
+ self.popup = None
+
+ def on_new_player(self, manager, player):
+ """Handle new player detection"""
+ print(f"Music dock component: New player detected - {player.props.player_name}")
+
+ # If this is our first player, make it current
+ if not self.current_player:
+ self.set_current_player(player)
+
+ self.update_visibility()
+
+ def on_player_vanish(self, manager, player):
+ """Handle player disappearing"""
+ print(f"Music dock component: Player vanished - {player.props.player_name}")
+
+ # If the current player vanished, find another one
+ if (
+ self.current_player
+ and self.current_player.props.player_name == player.props.player_name
+ ):
+ self.current_player = None
+ self.current_player_service = None
+
+ # Try to find another active player
+ # This is a simple approach - in a real implementation you might want more sophisticated logic
+
+ self.update_visibility()
+
+ def set_current_player(self, player):
+ """Set the current active player"""
+ # Disconnect from previous player
+ if self.current_player_service:
+ try:
+ self.current_player_service.disconnect_by_func(self.on_track_position)
+ self.current_player_service.disconnect_by_func(self.on_play)
+ self.current_player_service.disconnect_by_func(self.on_pause)
+ self.current_player_service.disconnect_by_func(self.on_metadata)
+ except:
+ pass
+
+ self.current_player = player
+ self.current_player_service = PlayerService(player=player)
+
+ # Connect to player events
+ self.current_player_service.connect("track-position", self.on_track_position)
+ self.current_player_service.connect("play", self.on_play)
+ self.current_player_service.connect("pause", self.on_pause)
+ self.current_player_service.connect("meta-change", self.on_metadata)
+
+ # Always use generic music icon, not player-specific icons
+ self.music_label.set_markup(icons.music)
+
+ # Update initial state
+ self.update_playback_state()
+
+ def on_track_position(self, service, pos, dur):
+ """Handle track position updates"""
+ self.position = pos
+ self.duration = dur
+
+ if dur > 0:
+ progress = pos / dur
+ # Use value property instead of set_percentage for CircularProgressBar
+ self.progress_bar.value = progress
+
+ # Update tooltip with timestamp info
+ pos_min = int(pos // 60)
+ pos_sec = int(pos % 60)
+ dur_min = int(dur // 60)
+ dur_sec = int(dur % 60)
+ timestamp_text = f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}"
+
+ # Get current track info if available
+ if hasattr(self, '_current_track_info'):
+ tooltip = f"{self._current_track_info}\n{timestamp_text}"
+ else:
+ tooltip = f"Music Player\n{timestamp_text}"
+ self.event_box.set_tooltip_text(tooltip)
+ else:
+ self.progress_bar.value = 0
+ self.event_box.set_tooltip_text("Music Player")
+
+ def on_play(self, service):
+ """Handle play state"""
+ self.music_label.set_markup(icons.music)
+ self.progress_bar.remove_style_class("paused")
+ # Update tooltip to show playing state
+ if hasattr(self, '_current_track_info'):
+ self.event_box.set_tooltip_text(f"{self._current_track_info} (Playing)")
+
+ def on_pause(self, service):
+ """Handle pause state"""
+ # Always use generic music icon when paused
+ self.music_label.set_markup(icons.music)
+ self.progress_bar.add_style_class("paused")
+ # Update tooltip to show paused state
+ if hasattr(self, '_current_track_info'):
+ self.event_box.set_tooltip_text(f"{self._current_track_info} (Paused)")
+
+ def on_metadata(self, service, metadata, player):
+ """Handle metadata changes"""
+ # Update tooltip with current track info
+ keys = metadata.keys()
+ if "xesam:artist" in keys and "xesam:title" in keys:
+ artist = (
+ metadata["xesam:artist"][0]
+ if metadata["xesam:artist"]
+ else "Unknown Artist"
+ )
+ title = metadata["xesam:title"]
+ self._current_track_info = f"{artist} - {title}"
+
+ # If we have position info, include timestamp
+ if hasattr(self, 'position') and hasattr(self, 'duration') and self.duration > 0:
+ pos_min = int(self.position // 60)
+ pos_sec = int(self.position % 60)
+ dur_min = int(self.duration // 60)
+ dur_sec = int(self.duration % 60)
+ timestamp_text = f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}"
+ tooltip = f"{self._current_track_info}\n{timestamp_text}"
+ else:
+ tooltip = self._current_track_info
+
+ self.event_box.set_tooltip_text(tooltip)
+ else:
+ self._current_track_info = "Music Player"
+ self.event_box.set_tooltip_text("Music Player")
+
+ def update_playback_state(self):
+ """Update the visual state based on current playback"""
+ if not self.current_player:
+ return
+
+ if (
+ self.current_player.props.playback_status.value_name
+ == "PLAYERCTL_PLAYBACK_STATUS_PLAYING"
+ ):
+ self.on_play(None)
+ else:
+ self.on_pause(None)
+
+ def on_music_clicked(self, button):
+ """Handle music button click - show popup"""
+ if self.popup and self.popup.get_visible():
+ self.popup.hide_popup()
+ else:
+ # Create popup on-demand if it doesn't exist
+ if not self.popup:
+ self.popup = MusicPlayerPopup()
+ self.popup.show_popup()
+
+ def on_scroll(self, widget, event):
+ """Handle scroll events for seeking"""
+ if not self.current_player or self.duration <= 0:
+ return False
+
+ # Calculate seek amount (5 seconds per scroll)
+ seek_amount = 5.0
+
+ if event.direction == Gdk.ScrollDirection.UP:
+ new_position = min(self.position + seek_amount, self.duration)
+ elif event.direction == Gdk.ScrollDirection.DOWN:
+ new_position = max(self.position - seek_amount, 0)
+ else:
+ return False
+
+ # Seek to new position
+ if self.current_player_service:
+ self.current_player_service.set_position(int(new_position))
+
+ return True
+
+ def update_visibility(self):
+ """Update component visibility based on available players"""
+ # Show component only if there are active players
+ has_players = self.current_player is not None
+ self.set_visible(has_players)
+
+ # Hide in vertical mode if user preference is set
+ if data.VERTICAL and hasattr(data, "DOCK_COMPONENTS_VISIBILITY"):
+ music_visible = data.DOCK_COMPONENTS_VISIBILITY.get("music_player", True)
+ if not music_visible:
+ self.set_visible(False)
diff --git a/modules/launcher/plugins/otp.py b/modules/launcher/plugins/otp.py
index b657367..84b42e7 100644
--- a/modules/launcher/plugins/otp.py
+++ b/modules/launcher/plugins/otp.py
@@ -1,4 +1,3 @@
-import base64
import json
import subprocess
import threading
@@ -11,13 +10,13 @@ from fabric.utils import get_relative_path
import utils.icons as icons
from modules.launcher.plugin_base import PluginBase
from modules.launcher.result import Result
-
-try:
- import pyotp
-
- PYOTP_AVAILABLE = True
-except ImportError:
- PYOTP_AVAILABLE = False
+from services.auth import (
+ generate_totp,
+ get_time_remaining_with_blink,
+ validate_base32_secret,
+ parse_otpauth_uri,
+ scan_qr_and_add_account,
+)
class OTPPlugin(PluginBase):
@@ -40,11 +39,6 @@ class OTPPlugin(PluginBase):
def initialize(self):
"""Initialize the OTP plugin."""
- if not PYOTP_AVAILABLE:
- raise RuntimeError(
- "pyotp is required for OTP functionality. Install with: pip install pyotp"
- )
-
self.set_triggers(["otp"])
self._load_secrets()
self._ensure_config_file()
@@ -301,26 +295,11 @@ class OTPPlugin(PluginBase):
def _generate_totp(self, secret: str) -> Optional[str]:
"""Generate TOTP code from secret."""
- try:
- return pyotp.TOTP(secret).now()
- except Exception as e:
- print(f"Error generating TOTP: {e}")
- return None
-
- def _get_time_remaining(self) -> int:
- """Get seconds remaining until next token refresh."""
- return 30 - (int(time.time()) % 30)
+ return generate_totp(secret)
def _get_time_remaining_with_blink(self) -> str:
"""Get time remaining with blinking effect."""
- time_remaining = self._get_time_remaining()
- current_second = int(time.time())
- should_blink = current_second % 2 == 0
-
- if should_blink:
- return f"{time_remaining}s"
- else:
- return f"{time_remaining}s"
+ return get_time_remaining_with_blink()
def query(self, query_string: str) -> List[Result]:
"""Process OTP queries."""
@@ -357,6 +336,13 @@ class OTPPlugin(PluginBase):
else:
remove_content = query[7:].strip()
return self._handle_remove_command(remove_content)
+ elif query_lower == "qr" or query_lower.startswith("qr "):
+ # Handle QR scanning command
+ if query_lower == "qr":
+ qr_content = ""
+ else:
+ qr_content = query[3:].strip()
+ return self._handle_qr_command(qr_content)
else:
return self._search_accounts(query)
@@ -413,7 +399,7 @@ class OTPPlugin(PluginBase):
results.append(
Result(
title="Available commands:",
- subtitle="add | remove ",
+ subtitle="add | remove | qr ",
icon_markup=icons.info,
action=lambda: None,
relevance=0.9,
@@ -435,7 +421,7 @@ class OTPPlugin(PluginBase):
results.append(
Result(
title=f"{totp_code}",
- subtitle_markup=f"{display_name} • {time_display} remaining",
+ subtitle_markup=f"{display_name} • {time_display} remaining • Shift+Enter: remove",
icon_markup=icons.key,
action=lambda code=totp_code: self._copy_to_clipboard(code),
relevance=1.0,
@@ -444,6 +430,7 @@ class OTPPlugin(PluginBase):
"type": "totp",
"account": account_name,
"code": totp_code,
+ "alt_action": lambda acc=account_name: self._remove_account_and_refresh(acc),
},
)
)
@@ -492,7 +479,7 @@ class OTPPlugin(PluginBase):
title=f"{totp_code}",
subtitle_markup=f"{display_name} • {
time_display
- } remaining",
+ } remaining • Shift+Enter: remove",
icon_markup=icons.key,
action=lambda code=totp_code: self._copy_to_clipboard(code),
relevance=1.0,
@@ -501,6 +488,7 @@ class OTPPlugin(PluginBase):
"type": "totp",
"account": account_name,
"code": totp_code,
+ "alt_action": lambda acc=account_name: self._remove_account_and_refresh(acc),
},
)
)
@@ -520,7 +508,7 @@ class OTPPlugin(PluginBase):
results.append(
Result(
title="Available commands:",
- subtitle="add | remove ",
+ subtitle="add | remove | qr ",
icon_markup=icons.info,
action=lambda: None,
relevance=0.4,
@@ -589,6 +577,66 @@ class OTPPlugin(PluginBase):
),
]
+ def _handle_qr_command(self, account_name: str) -> List[Result]:
+ """Handle QR scanning command."""
+ if not account_name:
+ return [
+ Result(
+ title="Scan QR Code",
+ subtitle="Click to scan QR code from screen",
+ icon_markup=icons.connect,
+ action=lambda: self._scan_qr_and_add_account(""),
+ relevance=1.0,
+ plugin_name=self.display_name,
+ data={"type": "qr_scan"},
+ ),
+ Result(
+ title="QR Scan Instructions:",
+ subtitle="Use 'qr ' to specify account name",
+ icon_markup=icons.info,
+ action=lambda: None,
+ relevance=0.9,
+ plugin_name=self.display_name,
+ data={"type": "help", "keep_launcher_open": True},
+ ),
+ ]
+ else:
+ return [
+ Result(
+ title=f"Scan QR Code for '{account_name}'",
+ subtitle="Click to scan QR code from screen",
+ icon_markup=icons.connect,
+ action=lambda name=account_name: self._scan_qr_and_add_account(name),
+ relevance=1.0,
+ plugin_name=self.display_name,
+ data={"type": "qr_scan"},
+ ),
+ ]
+
+ def _scan_qr_and_add_account(self, account_name: str):
+ """Scan QR code and add OTP account."""
+ print(f"QR scan action called for account: '{account_name}'")
+ print("Starting QR scan process asynchronously...")
+
+ # Run QR scanning asynchronously so launcher closes immediately
+ import threading
+ thread = threading.Thread(target=self._scan_qr_async, args=(account_name,))
+ thread.daemon = True
+ thread.start()
+
+ def _scan_qr_async(self, account_name: str):
+ """Async QR scanning process."""
+ result = scan_qr_and_add_account(account_name, str(self.secrets_file))
+
+ if result["success"]:
+ print(result["message"])
+ # Reload secrets from file
+ self._load_secrets()
+ # Trigger refresh to show the new account
+ self._trigger_refresh()
+ else:
+ print(f"QR scan failed: {result['error']}")
+
def _handle_remove_command(self, account_name: str) -> List[Result]:
"""Handle removal of OTP account."""
if not account_name:
@@ -636,18 +684,18 @@ class OTPPlugin(PluginBase):
results.append(
Result(
title=f"{totp_code}",
- subtitle_markup=f"Type: remove {acc_name} • {
+ subtitle_markup=f"Press Enter to remove • {
time_display
} remaining",
icon_markup=icons.trash,
- action=lambda: None, # No action - shows instruction only
+ action=lambda acc=acc_name: self._remove_account_and_refresh(acc),
relevance=0.9,
plugin_name=self.display_name,
data={
"type": "remove_instruction",
"account": acc_name,
"code": totp_code,
- "keep_launcher_open": True,
+ "keep_launcher_open": True
},
)
)
@@ -655,9 +703,9 @@ class OTPPlugin(PluginBase):
results.append(
Result(
title=f"Error: {acc_name}",
- subtitle=f"Type: remove {acc_name} (Invalid secret)",
+ subtitle=f"Press Enter to remove (Invalid secret)",
icon_markup=icons.trash,
- action=lambda: None, # No action - shows instruction only
+ action=lambda acc=acc_name: self._remove_account_and_refresh(acc),
relevance=0.8,
plugin_name=self.display_name,
data={
@@ -689,114 +737,28 @@ class OTPPlugin(PluginBase):
issuer = account_data.get("issuer", "")
display_name = f"{issuer} - {account_name}" if issuer else account_name
- # Remove the account
- try:
- del self.secrets[account_name]
- self._save_secrets()
-
- return [
- Result(
- title=f"✓ Removed '{display_name}'",
- subtitle="OTP account removed successfully",
- icon_markup=icons.check,
- action=lambda: self._trigger_refresh(),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"type": "success", "keep_launcher_open": True},
- )
- ]
- except Exception as e:
- return [
- Result(
- title="Error removing account",
- subtitle=f"Failed to remove account: {str(e)}",
- icon_markup=icons.cancel,
- action=lambda: None,
- relevance=0.5,
- plugin_name=self.display_name,
- data={"type": "error", "keep_launcher_open": True},
- )
- ]
+ # Show confirmation for removal
+ return [
+ Result(
+ title=f"Remove '{display_name}'?",
+ subtitle="Press Enter to confirm removal",
+ icon_markup=icons.trash,
+ action=lambda acc=account_name: self._remove_account_and_refresh(acc),
+ relevance=1.0,
+ plugin_name=self.display_name,
+ data={"type": "remove_confirm", "account": account_name,"keep_launcher_open": True,},
+ )
+ ]
def _handle_base32_secret(self, account_name: str, secret: str) -> List[Result]:
"""Handle raw Base32 secret."""
- try:
- # Clean up the secret - remove spaces, dashes, and convert to uppercase
- clean_secret = (
- secret.replace(" ", "").replace("-", "").replace("_", "").upper()
- )
-
- # Remove any non-base32 characters
- import re
-
- clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret)
-
- # Add padding if needed (Base32 requires padding to multiple of 8)
- while len(clean_secret) % 8 != 0:
- clean_secret += "="
-
- # Validate Base32 format
- try:
- base64.b32decode(clean_secret)
- except Exception as e:
- return [
- Result(
- title="Invalid Base32 secret",
- subtitle=f"Error: {str(e)}. Please check the secret format",
- icon_markup=icons.cancel,
- action=lambda: None,
- relevance=0.5,
- plugin_name=self.display_name,
- data={"type": "error", "keep_launcher_open": True},
- )
- ]
-
- # Test if the secret can generate a valid TOTP
- try:
- test_totp = pyotp.TOTP(clean_secret)
- test_code = test_totp.now()
- if not test_code or len(test_code) != 6:
- raise ValueError("Generated invalid TOTP code")
- except Exception as e:
- return [
- Result(
- title="Invalid secret for TOTP",
- subtitle=f"Cannot generate TOTP: {str(e)}",
- icon_markup=icons.cancel,
- action=lambda: None,
- relevance=0.5,
- plugin_name=self.display_name,
- data={"type": "error", "keep_launcher_open": True},
- )
- ]
-
- self.secrets[account_name] = {
- "secret": clean_secret,
- "issuer": "",
- "algorithm": "SHA1",
- "digits": 6,
- "period": 30,
- }
- self._save_secrets()
+ result = validate_base32_secret(secret)
+ if not result["success"]:
return [
Result(
- title=f"✓ Added '{account_name}'",
- subtitle=f"OTP account added successfully (secret: {
- clean_secret[:4]
- }...)",
- icon_markup=icons.check,
- action=lambda: self._trigger_refresh(),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"type": "success", "keep_launcher_open": True},
- )
- ]
- except Exception as e:
- return [
- Result(
- title="Error adding account",
- subtitle=f"Unexpected error: {str(e)}",
+ title="Invalid Base32 secret",
+ subtitle=result["error"],
icon_markup=icons.cancel,
action=lambda: None,
relevance=0.5,
@@ -805,79 +767,36 @@ class OTPPlugin(PluginBase):
)
]
+ self.secrets[account_name] = {
+ "secret": result["secret"],
+ "issuer": "",
+ "algorithm": "SHA1",
+ "digits": 6,
+ "period": 30,
+ }
+ self._save_secrets()
+
+ return [
+ Result(
+ title=f"✓ Added '{account_name}'",
+ subtitle=f"OTP account added successfully (secret: {result['secret'][:4]}...)",
+ icon_markup=icons.check,
+ action=lambda: self._trigger_refresh(),
+ relevance=1.0,
+ plugin_name=self.display_name,
+ data={"type": "success", "keep_launcher_open": True},
+ )
+ ]
+
def _handle_otpauth_uri(self, account_name: str, uri: str) -> List[Result]:
"""Handle otpauth:// URI."""
- try:
- from urllib.parse import parse_qs, urlparse
+ result = parse_otpauth_uri(uri, account_name)
- parsed = urlparse(uri)
- if parsed.scheme != "otpauth" or parsed.netloc != "totp":
- return [
- Result(
- title="Invalid otpauth URI",
- subtitle="Only otpauth://totp/ URIs are supported",
- icon_markup=icons.cancel,
- action=lambda: None,
- relevance=0.5,
- plugin_name=self.display_name,
- data={"type": "error", "keep_launcher_open": True},
- )
- ]
-
- if not account_name:
- account_path = parsed.path.lstrip("/")
- if ":" in account_path:
- issuer, extracted_name = account_path.split(":", 1)
- account_name = extracted_name
- else:
- account_name = account_path
-
- params = parse_qs(parsed.query)
- secret = params.get("secret", [""])[0]
- issuer = params.get("issuer", [""])[0]
- algorithm = params.get("algorithm", ["SHA1"])[0]
- digits = int(params.get("digits", ["6"])[0])
- period = int(params.get("period", ["30"])[0])
-
- if not secret:
- return [
- Result(
- title="No secret in URI",
- subtitle="otpauth URI must contain a secret parameter",
- icon_markup=icons.cancel,
- action=lambda: None,
- relevance=0.5,
- plugin_name=self.display_name,
- data={"type": "error", "keep_launcher_open": True},
- )
- ]
-
- self.secrets[account_name] = {
- "secret": secret,
- "issuer": issuer,
- "algorithm": algorithm,
- "digits": digits,
- "period": period,
- }
- self._save_secrets()
-
- display_name = f"{issuer} - {account_name}" if issuer else account_name
- return [
- Result(
- title=f"✓ Added '{display_name}'",
- subtitle="OTP account added from URI",
- icon_markup=icons.check,
- action=lambda: self._trigger_refresh(),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"type": "success", "keep_launcher_open": True},
- )
- ]
- except Exception as e:
+ if not result["success"]:
return [
Result(
title="Error parsing otpauth URI",
- subtitle=str(e),
+ subtitle=result["error"],
icon_markup=icons.cancel,
action=lambda: None,
relevance=0.5,
@@ -885,3 +804,25 @@ class OTPPlugin(PluginBase):
data={"type": "error", "keep_launcher_open": True},
)
]
+
+ self.secrets[result["account_name"]] = {
+ "secret": result["secret"],
+ "issuer": result["issuer"],
+ "algorithm": result["algorithm"],
+ "digits": result["digits"],
+ "period": result["period"],
+ }
+ self._save_secrets()
+
+ display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name']
+ return [
+ Result(
+ title=f"✓ Added '{display_name}'",
+ subtitle="OTP account added from URI",
+ icon_markup=icons.check,
+ action=lambda: self._trigger_refresh(),
+ relevance=1.0,
+ plugin_name=self.display_name,
+ data={"type": "success", "keep_launcher_open": True},
+ )
+ ]
diff --git a/player.py b/player.py
new file mode 100644
index 0000000..8577ad4
--- /dev/null
+++ b/player.py
@@ -0,0 +1,366 @@
+from fabric.widgets.box import Box
+from fabric.widgets.label import Label
+from fabric.widgets.centerbox import CenterBox
+from fabric.widgets.button import Button
+from fabric.widgets.stack import Stack
+
+from services.mpris import PlayerManager, PlayerService
+from utils.wiggle_bar import WigglyWidget
+
+import utils.icons as icons
+import config.data as data
+
+from loguru import logger
+
+
+class Player(Box):
+ def __init__(self, player, **kwargs):
+ super().__init__(style_classes="player", orientation="v", **kwargs)
+
+ # Check if we're in vertical mode based on dock position
+ vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False
+
+ if not vertical_mode:
+ # vertical class binding from unknown source
+ self.remove_style_class("vertical")
+
+ self._player = PlayerService(player=player)
+
+ 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)
+
+ self.player_name = Label(
+ name=player.props.player_name,
+ style_classes="player-icon",
+ markup=getattr(icons, player.props.player_name, icons.disc),
+ )
+
+ # Use a fallback background image
+ self.set_style(
+ f"background-image:url('{data.HOME_DIR}/.cache/walls/low_rez.png')"
+ )
+
+ self.song = Label(
+ name="song",
+ label="song",
+ justification="left",
+ h_align="start",
+ max_chars_width=10,
+ )
+ self.artist = Label(
+ name="artist", label="artist", justification="left", h_align="start"
+ )
+ self.music = Box(
+ name="music",
+ orientation="v",
+ h_expand=True,
+ v_expand=True,
+ children=[self.song, self.artist],
+ )
+
+ self.play_pause_button = Button(
+ name="pause-button",
+ child=Label(name="pause-label", markup=icons.play),
+ style_classes="pause-track",
+ tooltip_text="Play/Pause",
+ on_clicked=lambda b, *_: self.handle_play_pause(player),
+ )
+
+ self.shuffle_button = Button(
+ name="shuffle-button",
+ child=Label(name="shuffle", markup=icons.shuffle),
+ on_clicked=lambda b, *_: self.handle_shuffle(b, player),
+ )
+
+ self.wiggly = WigglyWidget()
+ self.wiggly.connect("on-seek", self.on_seek)
+ self.gtk_wrapper = Box(
+ orientation="v",
+ h_expand=True,
+ v_expand=True,
+ h_align="fill",
+ v_align="fill",
+ children=self.wiggly,
+ )
+
+ self.children = [
+ Box(name="source", h_expand=True, v_expand=True, children=self.player_name),
+ CenterBox(
+ name="details",
+ start_children=self.music,
+ end_children=self.play_pause_button if not vertical_mode else [],
+ ),
+ Box(
+ name="controls",
+ style_classes="horizontal" if not vertical_mode else "vertical",
+ spacing=5,
+ children=[
+ Button(
+ name="prev-button",
+ child=Label(name="play-previous", markup=icons.prev),
+ on_clicked=lambda b, *_: self.handle_prev(player),
+ ),
+ CenterBox(
+ name="progress-container",
+ h_expand=True,
+ v_expand=True,
+ orientation="v",
+ center_children=[self.gtk_wrapper],
+ ),
+ Button(
+ name="next-button",
+ child=Label(name="play-next", markup=icons.next),
+ on_clicked=lambda b, *_: self.handle_next(player),
+ ),
+ self.shuffle_button,
+ ]
+ if not vertical_mode
+ else [
+ CenterBox(
+ orientation="v",
+ h_expand=True,
+ start_children=[
+ CenterBox(
+ h_expand=True,
+ v_expand=True,
+ v_align="end",
+ start_children=[
+ Button(
+ name="prev-button",
+ child=Label(
+ name="play-previous", markup=icons.prev
+ ),
+ on_clicked=lambda b, *_: self.handle_prev(
+ player
+ ),
+ ),
+ Button(
+ name="next-button",
+ child=Label(
+ name="play-next", markup=icons.next
+ ),
+ on_clicked=lambda b, *_: self.handle_next(
+ player
+ ),
+ ),
+ self.shuffle_button,
+ ],
+ end_children=self.play_pause_button,
+ )
+ ],
+ end_children=CenterBox(
+ name="progress-container",
+ h_expand=True,
+ v_expand=True,
+ orientation="v",
+ center_children=[self.gtk_wrapper],
+ ),
+ )
+ ],
+ ),
+ ]
+
+ self.on_metadata(self._player, metadata=player.props.metadata, player=player)
+
+ def on_update_track_position(self, sender, pos, dur):
+ self.duration = dur
+ if dur > 0:
+ self.wiggly.update_value_from_signal(pos / 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 on_metadata(self, sender, metadata, player):
+ keys = metadata.keys()
+ if "xesam:artist" in keys and "xesam:title" in keys:
+ # Check if we're in vertical mode
+ vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False
+ _max_chars = 43 if not vertical_mode else 30
+
+ song_title = metadata["xesam:title"]
+ if len(song_title) > _max_chars:
+ song_title = song_title[: _max_chars - 1] + "…"
+ self.song.set_label(song_title)
+
+ artist_list = metadata["xesam:artist"]
+ artist_name = artist_list[0] if artist_list else "Unknown Artist"
+ if len(artist_name) > _max_chars:
+ artist_name = artist_name[: _max_chars - 1] + "…"
+ self.artist.set_label(artist_name)
+
+ if "mpris:artUrl" in keys:
+ self.set_style(f"background-image:url('{metadata['mpris:artUrl']}')")
+
+ if (
+ player.props.playback_status.value_name
+ == "PLAYERCTL_PLAYBACK_STATUS_PLAYING"
+ ):
+ self.on_play(self._player)
+
+ if player.props.shuffle == True:
+ self.shuffle_button.get_child().set_markup(icons.shuffle) # Use same icon for now
+ self.shuffle_button.get_child().set_name("disable-shuffle")
+ else:
+ self.shuffle_button.get_child().set_markup(icons.shuffle)
+ self.shuffle_button.get_child().set_name("shuffle")
+
+ def on_pause(self, sender):
+ self.play_pause_button.get_child().set_markup(icons.play)
+ self.play_pause_button.get_child().set_name("pause-label")
+ self.wiggly.dragging = True
+ self.wiggly.update_amplitude(True)
+ self.wiggly.pause = True
+ self.play_pause_button.add_style_class("pause-track")
+
+ def on_play(self, sender):
+ self.play_pause_button.get_child().set_markup(icons.pause)
+ self.play_pause_button.get_child().set_name("play-label")
+ self.wiggly.pause = False
+ self.wiggly.dragging = False
+ self.wiggly.update_amplitude(False)
+ self.play_pause_button.remove_style_class("pause-track")
+
+ def on_shuffle(self, sender, player, status):
+ print("callback status", status)
+ if status == False:
+ self.shuffle_button.get_child().set_markup(icons.shuffle)
+ self.shuffle_button.get_child().set_name("shuffle")
+ else:
+ self.shuffle_button.get_child().set_markup(icons.shuffle) # Use same icon for now
+ self.shuffle_button.get_child().set_name("disable-shuffle")
+
+ self.shuffle_button.get_child().set_style("color: white")
+
+ def handle_next(self, player):
+ player.next()
+
+ def handle_prev(self, player):
+ player.previous()
+
+ def handle_play_pause(self, player):
+ is_playing = self.play_pause_button.get_child().get_name() == "play-label"
+
+ def _set_play_ui():
+ self.play_pause_button.get_child().set_markup(icons.pause)
+ self.play_pause_button.remove_style_class("pause-track")
+ self.play_pause_button.get_child().set_name("pause-label")
+
+ def _set_pause_ui():
+ self.play_pause_button.get_child().set_markup(icons.play)
+ self.play_pause_button.add_style_class("pause-track")
+ self.play_pause_button.get_child().set_name("play-label")
+
+ if is_playing:
+ _set_pause_ui()
+ else:
+ _set_play_ui()
+
+ try:
+ player.play_pause()
+ except Exception as e:
+ # revert if signal failed
+ if is_playing:
+ _set_pause_ui()
+ else:
+ _set_play_ui()
+ logger.warning("Failed to toggle playback:", e)
+
+ def handle_shuffle(self, shuffle_button, player):
+ print("shuffle", player.props.shuffle)
+ if player.props.shuffle == False:
+ player.set_shuffle(True)
+ print("setting to true", player.props.player_name)
+ else:
+ player.set_shuffle(False)
+ shuffle_button.get_child().set_style("color: var(--outline)")
+
+
+class PlayerContainer(Box):
+ 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.stack = Stack(
+ name="player-container",
+ transition_type="crossfade",
+ transition_duration=100,
+ children=[],
+ )
+
+ self.player_stack = Stack(
+ name="player-stack",
+ transition_type="crossfade",
+ transition_duration=100,
+ children=[],
+ )
+
+ # Check if we're in vertical mode
+ vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False
+
+ self.player_switch_container = CenterBox(
+ name="player-switch-container",
+ orientation="h",
+ style_classes="horizontal-player"
+ if not vertical_mode
+ else "vertical-player",
+ center_children=[],
+ )
+ self.children = [self.stack, self.player_switch_container]
+ self.player = []
+ self.manager.init_all_players()
+
+ def new_player(self, manager, player):
+ print(player.props.player_name, "new player")
+ print(player)
+ new_player = Player(player=player)
+ new_player.gtk_wrapper.queue_draw()
+ new_player.set_name(player.props.player_name)
+ self.player.append(new_player)
+ print("stacking player")
+ self.stack.add_named(new_player, player.props.player_name)
+
+ 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),
+ )
+ )
+ self.update_player_list()
+
+ def switch_player(self, player_name, button):
+ self.stack.set_visible_child_name(player_name)
+
+ for btn in self.player_switch_container.center_children:
+ btn.remove_style_class("active")
+ button.add_style_class("active")
+
+ def on_player_vanish(self, manager, player):
+ for player_instance in self.player:
+ if player_instance.get_name() == player.props.player_name:
+ self.stack.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():
+ self.player_switch_container.remove_center(btn)
+ self.update_player_list()
+ break
+
+ def update_player_list(self):
+ curr = self.stack.get_visible_child()
+ if curr:
+ 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")
diff --git a/services/auth.py b/services/auth.py
new file mode 100644
index 0000000..c003288
--- /dev/null
+++ b/services/auth.py
@@ -0,0 +1,335 @@
+import subprocess
+import time
+import json
+import os
+from datetime import datetime
+from PIL import Image
+from pyzbar.pyzbar import decode
+import pyotp
+from urllib.parse import urlparse, parse_qs
+from pathlib import Path
+import config.data as data
+
+
+def get_otp_file_path():
+ """Returns the path to the OTP file inside the cache directory."""
+ cache_dir = Path(data.CACHE_DIR) / "otp"
+ # Create the directory if it doesn't exist
+ cache_dir.mkdir(parents=True, exist_ok=True)
+
+ file_path = cache_dir / "otp_codes.json"
+
+ # Create the file if it doesn't exist
+ if not file_path.exists():
+ with open(file_path, "w") as f:
+ json.dump([], f)
+ print(f"File {file_path} created successfully!")
+
+ return file_path
+
+
+def capture_selected_area(filename="/tmp/screenshot.png"):
+ """
+ Uses slurp to let the user select an area of the screen,
+ then captures that area using grim.
+ """
+ try:
+ # slurp returns coordinates in format: x,y widthxheight (e.g. 100,200 300x400)
+ result = subprocess.run(["slurp"], check=True, capture_output=True, text=True)
+ geometry = result.stdout.strip()
+ if not geometry:
+ print("No area selected with slurp.")
+ return None
+ except subprocess.CalledProcessError as e:
+ print("Error selecting area with slurp:", e)
+ return None
+
+ try:
+ # grim uses -g to capture a specific region
+ subprocess.run(["grim", "-g", geometry, filename], check=True)
+ except subprocess.CalledProcessError as e:
+ print("Error capturing screenshot with grim:", e)
+ return None
+
+ return filename
+
+
+def read_and_save_to_json():
+ # Get the full path to the JSON file
+ json_file = get_otp_file_path()
+
+ screenshot = capture_selected_area()
+ if screenshot is None:
+ print("Failed to capture the selected area.")
+ return False
+
+ # Short delay to ensure the file is written
+ time.sleep(1)
+
+ # Open the captured image
+ try:
+ img = Image.open(screenshot)
+ except Exception as e:
+ print("Error opening the image:", e)
+ return False
+
+ # Decode QR Code(s) from the image
+ decoded_objects = decode(img)
+ if not decoded_objects:
+ print("No QR Code detected in the selected area.")
+ return False
+
+ results = []
+
+ for obj in decoded_objects:
+ data = obj.data.decode("utf-8")
+ print("QR Code detected:", data)
+
+ result_entry = {"timestamp": datetime.now().isoformat(), "qr_data": data}
+
+ if data.startswith("otpauth://"):
+ parsed = urlparse(data)
+ query = parse_qs(parsed.query)
+
+ # Extract the secret properly
+ secret = query.get("secret", [None])[0]
+ issuer_from_query = query.get("issuer", [None])[0]
+
+ # Extract the label (path), which may contain issuer and account
+ label = parsed.path.lstrip("/") if parsed.path else ""
+ account_name = label
+ issuer_from_path = None
+
+ if ":" in label:
+ parts = label.split(":", 1)
+ issuer_from_path = parts[0]
+ account_name = parts[1]
+
+ # Prefer issuer from path if available, otherwise use from query
+ issuer = issuer_from_path or issuer_from_query
+
+ # Extract the period (default is 30 seconds)
+ period = int(query.get("period", ["30"])[0])
+
+ # Create TOTP object with the correct interval
+ totp = pyotp.TOTP(secret, interval=period)
+ current_otp = totp.now()
+ print(f"Generated OTP: {current_otp} (valid for {period} seconds)")
+
+ result_entry.update(
+ {
+ "type": "otp",
+ "secret": secret,
+ "issuer": issuer,
+ "account_name": account_name,
+ }
+ )
+ else:
+ result_entry["type"] = "unknown"
+ print("Unrecognized format. Expected a URI like otpauth://")
+ return False
+
+ results.append(result_entry)
+
+ # Load existing data if the file exists
+ existing_data = []
+ if os.path.exists(json_file):
+ try:
+ with open(json_file, "r") as f:
+ existing_data = json.load(f)
+ except json.JSONDecodeError:
+ print(f"Error reading the file {json_file}. Creating a new one.")
+
+ # Append new results
+ existing_data.extend(results)
+
+ # Save to JSON file
+ with open(json_file, "w") as f:
+ json.dump(existing_data, f, indent=4)
+
+ print(f"OTP data saved to {json_file}")
+ return True
+
+
+def CodeOTP(uri):
+ parsed = urlparse(uri)
+ query = parse_qs(parsed.query)
+ secret = query.get("secret", [None])[0]
+
+ if secret is None:
+ return None
+ else:
+ totp = pyotp.TOTP(secret)
+ return totp.now()
+
+
+# TOTP/OTP utility functions
+def generate_totp(secret: str) -> str:
+ """Generate TOTP code from secret."""
+ try:
+ return pyotp.TOTP(secret).now()
+ except Exception as e:
+ print(f"Error generating TOTP: {e}")
+ return None
+
+
+def get_time_remaining() -> int:
+ """Get seconds remaining until next token refresh."""
+ return 30 - (int(time.time()) % 30)
+
+
+def get_time_remaining_with_blink() -> str:
+ """Get time remaining with blinking effect."""
+ time_remaining = get_time_remaining()
+ current_second = int(time.time())
+ should_blink = current_second % 2 == 0
+
+ if should_blink:
+ return f"{time_remaining}s"
+ else:
+ return f"{time_remaining}s"
+
+
+def validate_base32_secret(secret: str) -> dict:
+ """Validate and clean Base32 secret."""
+ import base64
+ import re
+
+ try:
+ # Clean up the secret - remove spaces, dashes, and convert to uppercase
+ clean_secret = (
+ secret.replace(" ", "").replace("-", "").replace("_", "").upper()
+ )
+
+ # Remove any non-base32 characters
+ clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret)
+
+ # Add padding if needed (Base32 requires padding to multiple of 8)
+ while len(clean_secret) % 8 != 0:
+ clean_secret += "="
+
+ # Validate Base32 format
+ try:
+ base64.b32decode(clean_secret)
+ except Exception as e:
+ return {"success": False, "error": f"Invalid Base32 secret: {str(e)}"}
+
+ # Test if the secret can generate a valid TOTP
+ try:
+ test_totp = pyotp.TOTP(clean_secret)
+ test_code = test_totp.now()
+ if not test_code or len(test_code) != 6:
+ raise ValueError("Generated invalid TOTP code")
+ except Exception as e:
+ return {"success": False, "error": f"Cannot generate TOTP: {str(e)}"}
+
+ return {"success": True, "secret": clean_secret}
+ except Exception as e:
+ return {"success": False, "error": f"Unexpected error: {str(e)}"}
+
+
+def parse_otpauth_uri(uri: str, account_name: str = "") -> dict:
+ """Parse otpauth URI and extract account information."""
+ try:
+ parsed = urlparse(uri)
+ if parsed.scheme != "otpauth" or parsed.netloc != "totp":
+ return {"success": False, "error": "Only otpauth://totp/ URIs are supported"}
+
+ if not account_name:
+ account_path = parsed.path.lstrip("/")
+ if ":" in account_path:
+ issuer, extracted_name = account_path.split(":", 1)
+ account_name = extracted_name
+ else:
+ account_name = account_path
+
+ params = parse_qs(parsed.query)
+ secret = params.get("secret", [""])[0]
+ issuer = params.get("issuer", [""])[0]
+ algorithm = params.get("algorithm", ["SHA1"])[0]
+ digits = int(params.get("digits", ["6"])[0])
+ period = int(params.get("period", ["30"])[0])
+
+ if not secret:
+ return {"success": False, "error": "No secret found in URI"}
+
+ return {
+ "success": True,
+ "account_name": account_name,
+ "secret": secret,
+ "issuer": issuer,
+ "algorithm": algorithm,
+ "digits": digits,
+ "period": period,
+ }
+ except Exception as e:
+ return {"success": False, "error": f"Error parsing otpauth URI: {str(e)}"}
+
+
+def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict:
+ """Scan QR code and add OTP account to secrets file."""
+ try:
+ # Capture QR code from screen
+ screenshot_path = capture_selected_area()
+ if not screenshot_path:
+ return {"success": False, "error": "QR scan cancelled or failed"}
+
+ # Decode QR code
+ try:
+ img = Image.open(screenshot_path)
+ decoded_objects = decode(img)
+
+ if not decoded_objects:
+ return {"success": False, "error": "No QR code detected in selected area"}
+
+ # Process the first QR code found
+ qr_data = decoded_objects[0].data.decode("utf-8")
+ print(f"QR Code detected: {qr_data}")
+
+ if qr_data.startswith("otpauth://"):
+ # Parse otpauth URI
+ result = parse_otpauth_uri(qr_data, account_name)
+ if not result["success"]:
+ return result
+
+ # Load existing secrets
+ secrets = {}
+ if os.path.exists(secrets_file_path):
+ try:
+ with open(secrets_file_path, "r", encoding="utf-8") as f:
+ secrets = json.load(f)
+ except Exception as e:
+ print(f"Error loading secrets: {e}")
+
+ # Add new account
+ secrets[result["account_name"]] = {
+ "secret": result["secret"],
+ "issuer": result["issuer"],
+ "algorithm": result["algorithm"],
+ "digits": result["digits"],
+ "period": result["period"],
+ }
+
+ # Save secrets
+ try:
+ os.makedirs(os.path.dirname(secrets_file_path), exist_ok=True)
+ with open(secrets_file_path, "w", encoding="utf-8") as f:
+ json.dump(secrets, f, indent=2)
+ except Exception as e:
+ return {"success": False, "error": f"Error saving secrets: {str(e)}"}
+
+ display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name']
+ return {
+ "success": True,
+ "account_name": result["account_name"],
+ "display_name": display_name,
+ "message": f"Successfully added OTP account: {display_name}"
+ }
+ else:
+ return {"success": False, "error": "QR code is not an otpauth URI"}
+
+ except Exception as e:
+ return {"success": False, "error": f"Error processing QR code: {str(e)}"}
+
+ except Exception as e:
+ return {"success": False, "error": f"Error during QR scan: {str(e)}"}
diff --git a/services/mpris.py b/services/mpris.py
index 3125890..14a7a0b 100644
--- a/services/mpris.py
+++ b/services/mpris.py
@@ -1,209 +1,141 @@
-import contextlib
-from typing import Optional
+from gi.repository import Playerctl, GLib
+from fabric.utils import bulk_connect
+from fabric.core.service import Service, Signal
+from fabric import Fabricator
+
+from config.data import ALLOWED_PLAYERS
import gi
-from fabric.core.service import Property, Service, Signal
-from fabric.utils.helpers import get_enum_member_name, snake_case_to_kebab_case
-from gi.repository import Playerctl
-from loguru import logger
gi.require_version("Playerctl", "2.0")
-class MediaPlayer(Service):
- """A service to manage a media player."""
+class PlayerService(Service):
+ @Signal
+ def shuffle_toggle(self, player: Playerctl.Player, status: bool) -> None: ...
@Signal
- def exit(self) -> None: ...
+ def meta_change(self, metadata: GLib.Variant, player: Playerctl.Player) -> None: ...
@Signal
- def metadata_changed(self) -> None: ...
+ def pause(self) -> None: ...
@Signal
- def playback_status_changed(self) -> None: ...
+ def play(self) -> None: ...
- @Property(str, flags="read-write")
- def player_name(self) -> str:
- return self._player.get_property("player_name")
+ @Signal
+ def track_position(self, pos: float, dur: float) -> None: ...
- @Property(int, "read-write", default_value=0)
- def track_position(self) -> int:
- return self._player.get_property("position")
+ def __init__(self, player: Playerctl.Player, **kwargs):
+ 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)
+ print(type(player))
- @track_position.setter
- def track_position(self, new_pos: int):
- self._player.set_position(new_pos)
-
- @Property(str, flags="read-write")
- def album_image_url(self) -> str:
- return self._player.get_property("metadata").unpack().get("mpris:artUrl", None)
-
- @Property(int, flags="read-write")
- def track_duration(self) -> int:
- default_duration = minutes_to_microseconds(5)
- metadata = self._player.get_property("metadata").unpack()
- duration = metadata.get("mpris:length", default_duration) # Default: 5 minutes
- return default_duration if duration == 0 else duration
-
- @Property(str, flags="readable")
- def status(self) -> str:
- return snake_case_to_kebab_case(
- get_enum_member_name(
- self._player.get_property("playback-status"), # type: ignore
- default="unknown",
- )
+ self.status = self._player.props.playback_status
+ self.pos_fabricator = Fabricator(
+ interval=1000, # 1s
+ poll_from=lambda f, *_: self._player.get_position(),
+ on_changed=lambda f, *_: self.fabricating(),
)
+ self.poll_progress()
- @Property(str, flags="readable")
- def track(self) -> str:
- return (
- f"{self.track_artist} {self.track_title}"
- if self.player_name == "spotify"
- else self.track_title or "Music"
- )
+ def on_seeked(self, player, position):
+ if self.status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING":
+ self.pos_fabricator.start()
- @Property(str, flags="readable")
- def track_title(self) -> str:
- return self._player.get_title()
+ def set_position(self, pos: float):
+ print("seeking in the service")
+ self.pos_fabricator.stop()
+ micro_pos = int(pos * 1_000_000)
+ try:
+ self._player.set_position(micro_pos)
+ print(f"Set position to {micro_pos}")
+ except GLib.Error as e:
+ print(f"Failed to seek: {e}")
- @Property(str, flags="readable")
- def track_artist(self) -> str:
- return self._player.get_artist()
+ def poll_progress(self):
+ if self.status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING":
+ self.pos_fabricator.start()
+ else:
+ self.pos_fabricator.stop()
- @Property(bool, "readable", default_value=False)
- def can_go_next(self) -> bool:
- return self._player.get_property("can_go_next")
+ def fabricating(self):
+ pos = self._player.get_position() / 1_000_000 # seconds
+ # seconds
+ dur = self._player.props.metadata["mpris:length"] / 1_000_000
+ # print(self._player.get_position())
+ # print(f"[progress] {pos:.2f}s / {dur:.2f}s")
+ self.track_position(pos, dur)
- @Property(bool, "readable", default_value=False)
- def can_go_previous(self) -> bool:
- return self._player.get_property("can_go_previous")
+ def on_play(self, player, status):
+ print("player is playing: {}".format(player.props.player_name))
+ self.status = player.props.playback_status
+ self.poll_progress()
+ self.play()
- @Property(bool, "readable", default_value=False)
- def can_seek(self) -> bool:
- return self._player.get_property("can_seek")
+ def on_pause(self, player, status):
+ print("player is paused: {}".format(player.props.player_name))
+ self.status = player.props.playback_status
+ self.poll_progress()
+ self.pause()
- @Property(bool, "readable", default_value=False)
- def can_pause(self) -> bool:
- return self._player.get_property("can_pause")
+ def on_shuffle(self, player, status):
+ print("suffle status changed for: {}".format(player.props.player_name))
+ print(type(status), "here is the status type")
+ self.shuffle_toggle(player, status)
- def __init__(self, player):
- super().__init__()
- self._player = player
- self._signal_connectors = {}
-
- signals = {
- "seeked": lambda *args: self.notify("track_position"),
- "playback-status": lambda *args: self._on_playback_status_changed(),
- "metadata": lambda *args: self._on_metadata_changed(),
- "exit": lambda *args: self._on_exit(),
- }
-
- for signal_name, handler in signals.items():
- self._signal_connectors[signal_name] = self._player.connect(
- signal_name, handler
+ def on_metadata(self, player, metadata):
+ keys = metadata.keys()
+ if "xesam:artist" in keys and "xesam:title" in keys:
+ print(
+ "{} - {}".format(metadata["xesam:artist"][0], metadata["xesam:title"])
)
-
- def _on_playback_status_changed(self):
- self.notify("status")
- self.playback_status_changed.emit()
-
- def _on_metadata_changed(self):
- for prop in [
- "track_duration",
- "track_position",
- "track_title",
- "track_artist",
- "track",
- "album_image_url",
- ]:
- self.notify(prop)
- self.metadata_changed.emit()
-
- def _on_exit(self):
- for signal_id in self._signal_connectors.values():
- with contextlib.suppress(Exception):
- self._player.disconnect(signal_id)
- del self._player
- self.exit.emit()
-
- def play_pause(self):
- if self.can_pause:
- self._player.play_pause()
-
- def next(self):
- if self.can_go_next:
- self._player.next()
-
- def previous(self):
- if self.can_go_previous:
- self._player.previous()
-
- def get_position(self):
- return self._player.get_position() if self._player else 0
+ self.meta_change(metadata, player)
-@staticmethod
-def minutes_to_microseconds(minutes):
- return minutes * 60 * 1000000
-
-
-class MediaManager(Service):
- """MediaManager Service using Playerctl to control media players"""
+class PlayerManager(Service):
+ @Signal
+ def new_player(self, player: Playerctl.Player) -> Playerctl.Player: ...
@Signal
- def player_appeared(self) -> None: ...
+ def player_vanish(self, player: Playerctl.Player) -> None: ...
- @Signal
- def player_vanished(self) -> None: ...
-
- @Property(list, "readable")
- def player_names(self):
- return self._manager.get_property("player_names")
-
- @Property(list[MediaPlayer], "readable")
- def players(self) -> list[MediaPlayer]:
- return self._players.values()
-
- @Property(MediaPlayer, "readable")
- def current_player(self) -> Optional[MediaPlayer]:
- current_player = None
- for player in self.players:
- if player.status == "playing":
- return player
- current_player = player
- return current_player
-
- def __init__(self):
- super().__init__()
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
self._manager = Playerctl.PlayerManager()
- self._players: dict[str, MediaPlayer] = {}
-
+ self._manager.connect("name-appeared", self._on_name_appeared, self._manager)
self._manager.connect(
- "name-appeared", lambda _, player: self._on_player_appeared(player=player)
- )
- self._manager.connect(
- "player-vanished", lambda _, player: self._on_player_vanished(player=player)
+ "player-vanished", self._on_player_vanished, self._manager
)
+ print("playerctl init done")
- for player in self._manager.get_property("player-names"):
- self._on_player_appeared(player=player)
+ self._players = {}
- def _on_player_appeared(self, player):
- """Callback when a new player appears"""
- player = Playerctl.Player.new_from_name(player)
- player_name = player.get_property("player_name")
+ def init_all_players(self):
+ # invoked in the UI
+ for player_obj in self._manager.props.player_names:
+ name_str = player_obj.name
+ print(f"{name_str} appeared")
+ if name_str in ALLOWED_PLAYERS:
+ player = Playerctl.Player.new_from_name(player_obj)
+ self._manager.manage_player(player)
+ self.new_player(player)
- logger.info(f"[Media] New player appeared: {player_name}")
+ def _on_name_appeared(self, sender, name, manager):
+ name_str = name.name
+ print(f"{name_str} appeared")
+ if name_str in ALLOWED_PLAYERS:
+ player = Playerctl.Player.new_from_name(name)
+ self._manager.manage_player(player)
+ self.new_player(player)
- self._manager.manage_player(player)
- self._players[player_name] = MediaPlayer(player=player)
- self.notify("current-player")
- self.player_appeared.emit()
-
- def _on_player_vanished(self, player):
- """Callback when a player disappears"""
- player_name = player.get_property("player_name") or "Unknown"
- logger.info(f"[Media] Player vanished: {player_name}")
- self._players.pop(player_name)
- self.notify("current-player")
- self.player_vanished.emit()
+ def _on_player_vanished(self, sender, player, manager):
+ print("player has exited: {}".format(player.props.player_name))
+ print(type(player))
+ self.player_vanish(player)
diff --git a/services/qrcode.py b/services/qrcode.py
new file mode 100644
index 0000000..c003288
--- /dev/null
+++ b/services/qrcode.py
@@ -0,0 +1,335 @@
+import subprocess
+import time
+import json
+import os
+from datetime import datetime
+from PIL import Image
+from pyzbar.pyzbar import decode
+import pyotp
+from urllib.parse import urlparse, parse_qs
+from pathlib import Path
+import config.data as data
+
+
+def get_otp_file_path():
+ """Returns the path to the OTP file inside the cache directory."""
+ cache_dir = Path(data.CACHE_DIR) / "otp"
+ # Create the directory if it doesn't exist
+ cache_dir.mkdir(parents=True, exist_ok=True)
+
+ file_path = cache_dir / "otp_codes.json"
+
+ # Create the file if it doesn't exist
+ if not file_path.exists():
+ with open(file_path, "w") as f:
+ json.dump([], f)
+ print(f"File {file_path} created successfully!")
+
+ return file_path
+
+
+def capture_selected_area(filename="/tmp/screenshot.png"):
+ """
+ Uses slurp to let the user select an area of the screen,
+ then captures that area using grim.
+ """
+ try:
+ # slurp returns coordinates in format: x,y widthxheight (e.g. 100,200 300x400)
+ result = subprocess.run(["slurp"], check=True, capture_output=True, text=True)
+ geometry = result.stdout.strip()
+ if not geometry:
+ print("No area selected with slurp.")
+ return None
+ except subprocess.CalledProcessError as e:
+ print("Error selecting area with slurp:", e)
+ return None
+
+ try:
+ # grim uses -g to capture a specific region
+ subprocess.run(["grim", "-g", geometry, filename], check=True)
+ except subprocess.CalledProcessError as e:
+ print("Error capturing screenshot with grim:", e)
+ return None
+
+ return filename
+
+
+def read_and_save_to_json():
+ # Get the full path to the JSON file
+ json_file = get_otp_file_path()
+
+ screenshot = capture_selected_area()
+ if screenshot is None:
+ print("Failed to capture the selected area.")
+ return False
+
+ # Short delay to ensure the file is written
+ time.sleep(1)
+
+ # Open the captured image
+ try:
+ img = Image.open(screenshot)
+ except Exception as e:
+ print("Error opening the image:", e)
+ return False
+
+ # Decode QR Code(s) from the image
+ decoded_objects = decode(img)
+ if not decoded_objects:
+ print("No QR Code detected in the selected area.")
+ return False
+
+ results = []
+
+ for obj in decoded_objects:
+ data = obj.data.decode("utf-8")
+ print("QR Code detected:", data)
+
+ result_entry = {"timestamp": datetime.now().isoformat(), "qr_data": data}
+
+ if data.startswith("otpauth://"):
+ parsed = urlparse(data)
+ query = parse_qs(parsed.query)
+
+ # Extract the secret properly
+ secret = query.get("secret", [None])[0]
+ issuer_from_query = query.get("issuer", [None])[0]
+
+ # Extract the label (path), which may contain issuer and account
+ label = parsed.path.lstrip("/") if parsed.path else ""
+ account_name = label
+ issuer_from_path = None
+
+ if ":" in label:
+ parts = label.split(":", 1)
+ issuer_from_path = parts[0]
+ account_name = parts[1]
+
+ # Prefer issuer from path if available, otherwise use from query
+ issuer = issuer_from_path or issuer_from_query
+
+ # Extract the period (default is 30 seconds)
+ period = int(query.get("period", ["30"])[0])
+
+ # Create TOTP object with the correct interval
+ totp = pyotp.TOTP(secret, interval=period)
+ current_otp = totp.now()
+ print(f"Generated OTP: {current_otp} (valid for {period} seconds)")
+
+ result_entry.update(
+ {
+ "type": "otp",
+ "secret": secret,
+ "issuer": issuer,
+ "account_name": account_name,
+ }
+ )
+ else:
+ result_entry["type"] = "unknown"
+ print("Unrecognized format. Expected a URI like otpauth://")
+ return False
+
+ results.append(result_entry)
+
+ # Load existing data if the file exists
+ existing_data = []
+ if os.path.exists(json_file):
+ try:
+ with open(json_file, "r") as f:
+ existing_data = json.load(f)
+ except json.JSONDecodeError:
+ print(f"Error reading the file {json_file}. Creating a new one.")
+
+ # Append new results
+ existing_data.extend(results)
+
+ # Save to JSON file
+ with open(json_file, "w") as f:
+ json.dump(existing_data, f, indent=4)
+
+ print(f"OTP data saved to {json_file}")
+ return True
+
+
+def CodeOTP(uri):
+ parsed = urlparse(uri)
+ query = parse_qs(parsed.query)
+ secret = query.get("secret", [None])[0]
+
+ if secret is None:
+ return None
+ else:
+ totp = pyotp.TOTP(secret)
+ return totp.now()
+
+
+# TOTP/OTP utility functions
+def generate_totp(secret: str) -> str:
+ """Generate TOTP code from secret."""
+ try:
+ return pyotp.TOTP(secret).now()
+ except Exception as e:
+ print(f"Error generating TOTP: {e}")
+ return None
+
+
+def get_time_remaining() -> int:
+ """Get seconds remaining until next token refresh."""
+ return 30 - (int(time.time()) % 30)
+
+
+def get_time_remaining_with_blink() -> str:
+ """Get time remaining with blinking effect."""
+ time_remaining = get_time_remaining()
+ current_second = int(time.time())
+ should_blink = current_second % 2 == 0
+
+ if should_blink:
+ return f"{time_remaining}s"
+ else:
+ return f"{time_remaining}s"
+
+
+def validate_base32_secret(secret: str) -> dict:
+ """Validate and clean Base32 secret."""
+ import base64
+ import re
+
+ try:
+ # Clean up the secret - remove spaces, dashes, and convert to uppercase
+ clean_secret = (
+ secret.replace(" ", "").replace("-", "").replace("_", "").upper()
+ )
+
+ # Remove any non-base32 characters
+ clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret)
+
+ # Add padding if needed (Base32 requires padding to multiple of 8)
+ while len(clean_secret) % 8 != 0:
+ clean_secret += "="
+
+ # Validate Base32 format
+ try:
+ base64.b32decode(clean_secret)
+ except Exception as e:
+ return {"success": False, "error": f"Invalid Base32 secret: {str(e)}"}
+
+ # Test if the secret can generate a valid TOTP
+ try:
+ test_totp = pyotp.TOTP(clean_secret)
+ test_code = test_totp.now()
+ if not test_code or len(test_code) != 6:
+ raise ValueError("Generated invalid TOTP code")
+ except Exception as e:
+ return {"success": False, "error": f"Cannot generate TOTP: {str(e)}"}
+
+ return {"success": True, "secret": clean_secret}
+ except Exception as e:
+ return {"success": False, "error": f"Unexpected error: {str(e)}"}
+
+
+def parse_otpauth_uri(uri: str, account_name: str = "") -> dict:
+ """Parse otpauth URI and extract account information."""
+ try:
+ parsed = urlparse(uri)
+ if parsed.scheme != "otpauth" or parsed.netloc != "totp":
+ return {"success": False, "error": "Only otpauth://totp/ URIs are supported"}
+
+ if not account_name:
+ account_path = parsed.path.lstrip("/")
+ if ":" in account_path:
+ issuer, extracted_name = account_path.split(":", 1)
+ account_name = extracted_name
+ else:
+ account_name = account_path
+
+ params = parse_qs(parsed.query)
+ secret = params.get("secret", [""])[0]
+ issuer = params.get("issuer", [""])[0]
+ algorithm = params.get("algorithm", ["SHA1"])[0]
+ digits = int(params.get("digits", ["6"])[0])
+ period = int(params.get("period", ["30"])[0])
+
+ if not secret:
+ return {"success": False, "error": "No secret found in URI"}
+
+ return {
+ "success": True,
+ "account_name": account_name,
+ "secret": secret,
+ "issuer": issuer,
+ "algorithm": algorithm,
+ "digits": digits,
+ "period": period,
+ }
+ except Exception as e:
+ return {"success": False, "error": f"Error parsing otpauth URI: {str(e)}"}
+
+
+def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict:
+ """Scan QR code and add OTP account to secrets file."""
+ try:
+ # Capture QR code from screen
+ screenshot_path = capture_selected_area()
+ if not screenshot_path:
+ return {"success": False, "error": "QR scan cancelled or failed"}
+
+ # Decode QR code
+ try:
+ img = Image.open(screenshot_path)
+ decoded_objects = decode(img)
+
+ if not decoded_objects:
+ return {"success": False, "error": "No QR code detected in selected area"}
+
+ # Process the first QR code found
+ qr_data = decoded_objects[0].data.decode("utf-8")
+ print(f"QR Code detected: {qr_data}")
+
+ if qr_data.startswith("otpauth://"):
+ # Parse otpauth URI
+ result = parse_otpauth_uri(qr_data, account_name)
+ if not result["success"]:
+ return result
+
+ # Load existing secrets
+ secrets = {}
+ if os.path.exists(secrets_file_path):
+ try:
+ with open(secrets_file_path, "r", encoding="utf-8") as f:
+ secrets = json.load(f)
+ except Exception as e:
+ print(f"Error loading secrets: {e}")
+
+ # Add new account
+ secrets[result["account_name"]] = {
+ "secret": result["secret"],
+ "issuer": result["issuer"],
+ "algorithm": result["algorithm"],
+ "digits": result["digits"],
+ "period": result["period"],
+ }
+
+ # Save secrets
+ try:
+ os.makedirs(os.path.dirname(secrets_file_path), exist_ok=True)
+ with open(secrets_file_path, "w", encoding="utf-8") as f:
+ json.dump(secrets, f, indent=2)
+ except Exception as e:
+ return {"success": False, "error": f"Error saving secrets: {str(e)}"}
+
+ display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name']
+ return {
+ "success": True,
+ "account_name": result["account_name"],
+ "display_name": display_name,
+ "message": f"Successfully added OTP account: {display_name}"
+ }
+ else:
+ return {"success": False, "error": "QR code is not an otpauth URI"}
+
+ except Exception as e:
+ return {"success": False, "error": f"Error processing QR code: {str(e)}"}
+
+ except Exception as e:
+ return {"success": False, "error": f"Error during QR scan: {str(e)}"}
diff --git a/styles/components.css b/styles/components.css
index f408bd6..3aa6fc9 100644
--- a/styles/components.css
+++ b/styles/components.css
@@ -1,3 +1,8 @@
+#player-circle {
+ padding: 4px;
+ color: var(--surface-bright);
+ border: 3px solid var(--primary);
+}
/*
* ── language indicator ────────────────────────────────────────────────
*/
@@ -176,5 +181,3 @@ menu > menuitem:hover {
menu > menuitem:hover > label {
color: var(--shadow);
}
-
-
diff --git a/styles/notification.css b/styles/notification.css
index 02a2584..465421b 100644
--- a/styles/notification.css
+++ b/styles/notification.css
@@ -51,6 +51,11 @@
margin: 4px;
min-width: 430px;
}
+
+#notification-history.vertical {
+ margin: 10px;
+}
+
#notification-stack-box {
border-radius: 32px;
padding: 16px;
diff --git a/styles/player.css b/styles/player.css
new file mode 100644
index 0000000..dc818fb
--- /dev/null
+++ b/styles/player.css
@@ -0,0 +1,149 @@
+.player {
+ /* background-color: aqua; */
+ min-height: 100px;
+ min-width: 400px;
+ box-shadow:
+ inset 0 100px 100px black,
+ inset 10px 0px 10px rgba(0, 0, 0, 0.5),
+ inset -10px 0px 10px rgba(0, 0, 0, 0.5);
+ border-bottom-right-radius: 20px;
+ border-bottom-left-radius: 20px;
+ padding: 20px;
+ background-position: center;
+ background-size: cover;
+ margin: 4px;
+}
+
+.player.vertical {
+ box-shadow:
+ inset 100px 0px 150px black,
+ inset 0px 10px 10px rgba(0, 0, 0, 0.5),
+ inset 0px -10px 10px rgba(0, 0, 0, 0.5);
+ border-top-right-radius: 20px;
+ border-bottom-right-radius: 20px;
+ min-width: 400px;
+ min-height: 150px;
+ margin: 10px;
+}
+
+.player-icon {
+ font-size: 24px;
+}
+#artist {
+ color: var(--primary);
+}
+
+#song {
+ color: whitesmoke;
+ font-size: 16px;
+ margin-top: 5px;
+}
+
+#wiggle-bar {
+ color: var(--primary);
+}
+
+#source {
+ /* background-color: antiquewhite; */
+ margin-top: -10px;
+}
+
+#controls {
+ margin-bottom: -10px;
+}
+
+#play-next,
+#play-previous {
+ font-size: 24px;
+}
+
+#progress {
+ /* min-height: 1px; */
+ background-color: antiquewhite;
+}
+
+#pause-button {
+ background-color: var(--surface);
+ border-radius: 20px;
+ padding: 8px;
+}
+
+#pause-button.pause-track {
+ border-radius: 100%;
+}
+
+#pause-label,
+#play-label {
+ color: var(--primary);
+ font-size: 20px;
+ padding: 2px;
+ border-radius: 999px;
+ min-width: 25px;
+}
+
+#shuffle-button,
+#next-button,
+#prev-button {
+ border: none;
+ padding: 0px;
+ padding-left: 11px;
+ padding-right: 11px;
+}
+
+#shuffle-button:hover,
+#next-button:hover,
+#prev-button:hover {
+ background-color: var(--surface-bright);
+}
+
+#shuffle {
+ font-size: 17px;
+}
+
+.player-button {
+ border-radius: 5px;
+ /* background-color: var(--secondary); */
+ background-color: white;
+ opacity: 0.5;
+ min-width: 10px;
+ min-height: 10px;
+ padding: 0px;
+ margin: 0px;
+ border: none;
+ margin-left: 5px;
+}
+
+.player-button.active {
+ opacity: 1;
+}
+
+#player-switch-container.vertical-player {
+ margin-top: -23px;
+ padding-bottom: 13px;
+}
+
+#player-switch-container {
+ margin-top: -18px;
+ padding-bottom: 13px;
+}
+
+#controls.vertical {
+ padding-bottom: 10px;
+ margin-bottom: 0px;
+}
+
+#disable-shuffle {
+ font-size: 27px;
+}
+
+#progress-slider {
+ margin-left: -3px;
+ margin-right: -2px;
+ margin-top: 8px;
+ margin-bottom: 8px;
+ border-radius: 2px;
+}
+
+#progress-other-end {
+ min-width: 50px;
+}
diff --git a/utils/icons.py b/utils/icons.py
index 7cea415..eeb0c54 100644
--- a/utils/icons.py
+++ b/utils/icons.py
@@ -189,6 +189,7 @@ star: str = ""
process: str = ""
+music: str = ""
# otp
check: str = ""
bulb: str = ""
diff --git a/utils/wiggle_bar.py b/utils/wiggle_bar.py
new file mode 100644
index 0000000..69298c2
--- /dev/null
+++ b/utils/wiggle_bar.py
@@ -0,0 +1,174 @@
+import math
+from fabric.core.service import Service, Signal
+from gi.repository import Gtk, Gdk, GLib
+import gi
+
+gi.require_version("Gtk", "3.0")
+
+
+class WigglyWidget(Gtk.DrawingArea, Service):
+ @Signal
+ def on_seek(self, ratio: float) -> None: ...
+
+ def __init__(self):
+ super().__init__(name="wiggle-bar")
+ self.phase = 0
+ self.set_size_request(-1, 20)
+ self.connect("draw", self.on_draw)
+
+ # mouse events
+ self.add_events(
+ Gdk.EventMask.BUTTON_PRESS_MASK
+ | Gdk.EventMask.BUTTON_RELEASE_MASK
+ | Gdk.EventMask.POINTER_MOTION_MASK
+ )
+
+ self.connect("button-press-event", self.on_button_press)
+ self.connect("motion-notify-event", self.on_motion)
+ self.connect("button-release-event", self.on_button_release)
+
+ GLib.timeout_add(16, self.update) # 60 FPS
+ self.value = 0.0
+ self.amplitude = 2
+ self.dragging = False
+ self.pause = False
+
+ self.show_all()
+
+ def animate_amplitude_to(self):
+ if abs(self.amplitude - self.amplitude_target) < 0.01:
+ self.amplitude = self.amplitude_target
+ self.queue_draw()
+ return False
+ else:
+ self.amplitude += self.amplitude_step
+ self.queue_draw()
+ return True
+
+ def update_amplitude(self, decrease):
+ if decrease:
+ self.amplitude_target = 0
+ self.amplitude_step = (0 - self.amplitude) / 10.0
+ else:
+ self.amplitude_target = 2
+ self.amplitude_step = (2 - self.amplitude) / 10.0
+
+ # animation stops when False is returned btw
+ GLib.timeout_add(16, self.animate_amplitude_to)
+
+ def update_value_from_x(self, x):
+ width = self.get_allocated_width()
+ self.value = max(0.0, min(1.0, x / width))
+ self.on_seek(self.value)
+ self.queue_draw()
+
+ def on_button_press(self, widget, event):
+ self.dragging = True
+ self.update_amplitude(True)
+ self.update_value_from_x(event.x)
+ return True
+
+ # def animate_progress_to(self):
+ # print("frame")
+ # if self.value < self.value_target:
+ # self.value = self.value_target
+ # print(self.value)
+ # self.queue_draw()
+ # return False
+ # else:
+ # self.value += self.value_step
+ # self.queue_draw()
+ # return True
+
+ # def update_value_from_signal(self, new_value):
+ # self.value_target = min(1.0, new_value)
+ # steps = 60 # 1s / 16ms ≈ 60 frames
+ # self.value_step = (self.value_target - self.value) / steps
+ # GLib.timeout_add(16, self.animate_progress_to)
+
+ def update_value_from_signal(self, new_value):
+ self.value = min(1.0, new_value)
+
+ def on_motion(self, widget, event):
+ if self.dragging and not self.pause:
+ self.update_value_from_x(event.x)
+
+ cursor = Gdk.Cursor.new(Gdk.CursorType.HAND1)
+ widget.get_window().set_cursor(cursor)
+
+ return True
+
+ def on_button_release(self, widget, event):
+ if self.pause == False:
+ self.dragging = False
+ self.update_amplitude(False)
+ return True
+
+ def update(self):
+ if self.dragging == False:
+ self.phase += 0.1
+ # if (self.value<1):
+ # self.value += 0.001
+ self.queue_draw()
+ return True
+
+ def on_draw(self, widget, cr):
+ alloc_width = self.get_allocated_width()
+ height = self.get_allocated_height()
+ center_y = height / 2
+ # self.amplitude = 2
+ frequency = 0.3
+ slider_diameter = 6
+
+ width = int(alloc_width * self.value) - slider_diameter + 1
+
+ cr.set_source_rgb(1, 1, 1) # color here btw
+ cr.set_line_width(2)
+
+ last_x = 0 # fallback values in case the slider gets dragged out of range
+ last_y = 0
+
+ cr.move_to(0, center_y)
+ for x in range(width):
+ y = center_y + self.amplitude * math.sin((x * frequency) + self.phase)
+ cr.line_to(x, y)
+ last_x, last_y = x, y
+
+ cr.stroke()
+
+ # Set the color using CSS - the widget should have CSS styling applied
+ # that sets the color to var(--primary). For now, use a fallback approach.
+ style_context = self.get_style_context()
+
+ try:
+ # Get the foreground color which should be styled with CSS
+ color = style_context.get_color(Gtk.StateFlags.NORMAL)
+ cr.set_source_rgba(color.red, color.green, color.blue, color.alpha)
+ except:
+ # Fallback to a nice blue color that matches the primary theme
+ cr.set_source_rgb(0.2, 0.6, 1.0)
+
+ rect_width = 6
+ arc_radius = slider_diameter / 2
+ self.draw_rounded_rect(cr, last_x, height, rect_width, rect_width, arc_radius)
+ cr.fill()
+
+ cr.stroke()
+
+ cr.set_source_rgb(1, 1, 1)
+ cr.set_line_width(1)
+
+ # This is for anti aliasing. When line width becomes 1px and the position is not an integer, two lines of opacity 50% is drawn on two pixel rows.
+ # This moves the line to make it pixel perfect hence not needing anti aliasing
+ cr.translate(0, 0.5)
+
+ cr.move_to(last_x + 2 * arc_radius, height / 2)
+ cr.line_to(alloc_width, height / 2)
+
+ cr.stroke()
+
+ def draw_rounded_rect(self, cr, x, y, width, _height, radius):
+ cr.new_sub_path()
+ cr.arc(x + width - radius, radius, radius, -math.pi, 0)
+ cr.arc(x + width - radius, y - radius, radius, 0, -math.pi)
+ cr.close_path()