Implement MPRIS-helper

openfreebuds/driver/huawei/handler/state_in_ear.py

@@ -1,3 +1,5 @@
+import json
+
 from openfreebuds.driver.huawei.driver.generic import OfbDriverHandlerHuawei
 from openfreebuds.driver.huawei.package import HuaweiSppPackage
 
@@ -16,4 +18,4 @@ async def on_init(self):
     async def on_package(self, package: HuaweiSppPackage):
         value = package.find_param(8, 9)
         if len(value) == 1:
-            await self.driver.put_property("state", "in_ear", value[0] == 1)
+            await self.driver.put_property("state", "in_ear", json.dumps(value[0] == 1))

openfreebuds_backend/linux/dbus/mpris.py

@@ -0,0 +1,36 @@
+from sdbus import DbusInterfaceCommonAsync, dbus_property_async, dbus_method_async
+from sdbus_async.dbus_daemon import FreedesktopDbus
+
+
+class MPRISPProxy(DbusInterfaceCommonAsync, interface_name="org.mpris.MediaPlayer2"):
+    def __init__(self, service_name):
+        super().__init__()
+        self._proxify(service_name, "/org/mpris/MediaPlayer2")
+        self.Player = MPRISPlayer2Proxy.new_proxy(service_name, "/org/mpris/MediaPlayer2")
+
+    @staticmethod
+    async def get_all():
+        items: list[MPRISPProxy] = []
+        bus = FreedesktopDbus()
+        for name in await bus.list_names():
+            if name.startswith("org.mpris.MediaPlayer2"):
+                items.append(MPRISPProxy(name))
+        return items
+
+    @dbus_property_async("s")
+    def Identity(self) -> str:
+        pass
+
+
+class MPRISPlayer2Proxy(DbusInterfaceCommonAsync, interface_name="org.mpris.MediaPlayer2.Player"):
+    @dbus_property_async("s")
+    def PlaybackStatus(self) -> str:
+        pass
+
+    @dbus_method_async()
+    async def Pause(self):
+        pass
+
+    @dbus_method_async()
+    async def Play(self):
+        pass

openfreebuds_qt/app/module/linux_related.py

@@ -2,23 +2,35 @@
 import webbrowser
 
 from PyQt6.QtCore import pyqtSlot
+from qasync import asyncSlot
 
 from openfreebuds_qt.app.module.common import OfbQtCommonModule
+from openfreebuds_qt.config import OfbQtConfigParser
 from openfreebuds_qt.designer.linux_extras import Ui_OfbQtLinuxExtrasModule
+from openfreebuds_qt.utils import blocked_signals
+from openfreebuds_qt.utils.mpris.service import OfbQtMPRISHelperService
 
 
 class OfbQtLinuxExtrasModule(Ui_OfbQtLinuxExtrasModule, OfbQtCommonModule):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        self.setupUi(self)
-
-        # TODO: Impl MPRIS-helper
+        self.config = OfbQtConfigParser.get_instance()
+        self.service = OfbQtMPRISHelperService.get_instance(self.ofb)
 
+        self.setupUi(self)
+        with blocked_signals(self.mpris_helper_checkbox):
+            self.mpris_helper_checkbox.setChecked(self.config.get("mpris", "enabled", False))
         if os.environ.get("XDG_SESSION_TYPE") != "wayland":
             self.wayland_root.setVisible(False)
 
     @pyqtSlot()
     def on_hotkeys_doc(self):
         url = "https://mmk.pw/en/openfreebuds/help"
         webbrowser.open(url)
+
+    @asyncSlot(bool)
+    async def on_mpris_toggle(self, value: bool):
+        self.config.set("mpris", "enabled", value)
+        self.config.save()
+        await self.service.start()

openfreebuds_qt/designer/linux_extras.ui

@@ -40,6 +40,35 @@
      </layout>
     </widget>
    </item>
+   <item>
+    <widget class="QGroupBox" name="groupBox_2">
+     <property name="title">
+      <string>Theme</string>
+     </property>
+     <layout class="QVBoxLayout" name="verticalLayout_3">
+      <item>
+       <widget class="QLabel" name="label_3">
+        <property name="text">
+         <string>Fresh versions of OpenFreebuds are written in Qt6, and uses system-wide Qt UI theme. So, if application color scheme didn't match with system, or it looks ugly, you should configure global Qt style settings.</string>
+        </property>
+        <property name="wordWrap">
+         <bool>true</bool>
+        </property>
+       </widget>
+      </item>
+      <item>
+       <widget class="QLabel" name="label_4">
+        <property name="text">
+         <string>In KDE, LxQT or other Qt-based desktop environments, use system appearance settings. Otherwise, configure qt manually or use any configuration tool like qt6ct.</string>
+        </property>
+        <property name="wordWrap">
+         <bool>true</bool>
+        </property>
+       </widget>
+      </item>
+     </layout>
+    </widget>
+   </item>
    <item>
     <widget class="QGroupBox" name="wayland_root">
      <property name="title">
@@ -105,8 +134,25 @@
     </hint>
    </hints>
   </connection>
+  <connection>
+   <sender>mpris_helper_checkbox</sender>
+   <signal>toggled(bool)</signal>
+   <receiver>OfbQtLinuxExtrasModule</receiver>
+   <slot>on_mpris_toggle(bool)</slot>
+   <hints>
+    <hint type="sourcelabel">
+     <x>293</x>
+     <y>52</y>
+    </hint>
+    <hint type="destinationlabel">
+     <x>293</x>
+     <y>206</y>
+    </hint>
+   </hints>
+  </connection>
  </connections>
  <slots>
   <slot>on_hotkeys_doc()</slot>
+  <slot>on_mpris_toggle(bool)</slot>
  </slots>
 </ui>

openfreebuds_qt/main.py

@@ -14,6 +14,7 @@
 from openfreebuds_qt.generic import IOfbQtApplication
 from openfreebuds_qt.tray.main import OfbTrayIcon
 from openfreebuds_qt.utils import OfbQtDeviceAutoSelect, OfbQtHotkeyService, list_available_locales
+from openfreebuds_qt.utils.mpris.service import OfbQtMPRISHelperService
 
 log = create_logger("OfbQtApplication")
 
@@ -33,6 +34,7 @@ def __init__(self, args):
         self.ofb: Optional[IOpenFreebuds] = None
         self.auto_select: Optional[OfbQtDeviceAutoSelect] = None
         self.hotkeys: Optional[OfbQtHotkeyService] = None
+        self.mpris: Optional[OfbQtMPRISHelperService] = None
         self.tray: Optional[OfbTrayIcon] = None
         self.main_window: Optional[OfbQtMainWindow] = None
 
@@ -79,6 +81,7 @@ async def boot(self):
 
             # Initialize services
             self.hotkeys = OfbQtHotkeyService.get_instance(self.ofb)
+            self.mpris = OfbQtMPRISHelperService.get_instance(self.ofb)
             self.auto_select = OfbQtDeviceAutoSelect(self.ofb)
             self.tray = OfbTrayIcon(self)
             self.main_window = OfbQtMainWindow(self)
@@ -87,6 +90,8 @@ async def boot(self):
                 await self.restore_device()
                 await self.auto_select.boot()
 
+            self.hotkeys.start()
+            await self.mpris.start()
             await self.tray.boot()
             await self.main_window.boot()
 

openfreebuds_qt/utils/hotkeys/service.py

@@ -23,9 +23,9 @@ def get_instance(ofb: IOpenFreebuds):
         return OfbQtHotkeyService.instance
 
     def start(self):
+        self.stop()
         if not self.config.get("hotkeys", "enabled", False):
             return
-        self.stop()
 
         try:
             from pynput.keyboard import GlobalHotKeys

openfreebuds_qt/utils/mpris/service.py

@@ -0,0 +1,75 @@
+import asyncio
+from contextlib import suppress
+from typing import Optional
+
+from openfreebuds import IOpenFreebuds, OfbEventKind
+from openfreebuds.utils.logger import create_logger
+from openfreebuds_backend.linux.dbus.mpris import MPRISPProxy
+from openfreebuds_qt.config import OfbQtConfigParser
+from openfreebuds_qt.utils import OfbCoreEvent
+
+log = create_logger("OfbQtMPRISHelperService")
+
+
+class OfbQtMPRISHelperService:
+    instance = None
+
+    def __init__(self, ofb: IOpenFreebuds):
+        self.ofb = ofb
+        self.config = OfbQtConfigParser.get_instance()
+
+        self._task: Optional[asyncio.Task] = None
+        self.paused_players: list[MPRISPProxy] = []
+        self.last_in_ear: bool = True
+
+    async def _trigger(self):
+        in_ear = await self.ofb.get_property("state", "in_ear", "false") == "true"
+        enabled = await self.ofb.get_property("config", "auto_pause", "false") == "true"
+
+        if in_ear != self.last_in_ear and enabled:
+            if self.last_in_ear is True and in_ear is False:
+                # Pause all
+                self.paused_players = []
+                for service in await MPRISPProxy.get_all():
+                    if await service.Player.PlaybackStatus == "Playing":
+                        log.info(f"Pause {service.Identity}")
+                        await service.Player.Pause()
+                        self.paused_players.append(service)
+            elif self.last_in_ear is False and in_ear is True:
+                for service in self.paused_players:
+                    log.info(f"Resume {service.Identity}")
+                    await service.Player.Play()
+                self.paused_players = []
+            self.last_in_ear = in_ear
+
+    @staticmethod
+    def get_instance(ofb: IOpenFreebuds):
+        if OfbQtMPRISHelperService.instance is None:
+            OfbQtMPRISHelperService.instance = OfbQtMPRISHelperService(ofb)
+        return OfbQtMPRISHelperService.instance
+
+    async def stop(self):
+        if self._task is not None:
+            with suppress(Exception):
+                self._task.cancel()
+                await self._task
+            self._task = None
+
+    async def start(self):
+        await self.stop()
+        if not self.config.get("mpris", "enabled", False):
+            return
+
+        self._task = asyncio.create_task(self._main())
+
+    async def _main(self):
+        log.info("Started")
+        member_id = await self.ofb.subscribe(kind_filters=[OfbEventKind.PROPERTY_CHANGED])
+
+        while True:
+            try:
+                event = OfbCoreEvent(*await self.ofb.wait_for_event(member_id))
+                if event.is_changed("state", "in_ear"):
+                    await self._trigger()
+            except Exception:
+                log.exception("Failure")