282 lines
9.9 KiB
Python
Executable File
282 lines
9.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import requests
|
|
import time
|
|
import re
|
|
from threading import Thread
|
|
from pydbus import SessionBus
|
|
from pydbus.generic import signal
|
|
from gi.repository import GLib
|
|
from bs4 import BeautifulSoup
|
|
|
|
# --- CONFIGURATION ---
|
|
BASE_URL = "http://localhost:5666"
|
|
AUTH = ('winamp', 'llama')
|
|
APP_ID = "org.mpris.MediaPlayer2.winamp"
|
|
|
|
class WinampMPRIS:
|
|
"""
|
|
MPRIS2 specification implementation for Winamp Web Interface.
|
|
Optimized for Plasma 6 property change detection.
|
|
"""
|
|
dbus = f"""
|
|
<node>
|
|
<interface name="org.mpris.MediaPlayer2">
|
|
<property name="CanQuit" type="b" access="read" />
|
|
<property name="CanRaise" type="b" access="read" />
|
|
<property name="HasTrackList" type="b" access="read" />
|
|
<property name="Identity" type="s" access="read" />
|
|
<property name="DesktopEntry" type="s" access="read" />
|
|
<property name="SupportedUriSchemes" type="as" access="read" />
|
|
<property name="SupportedMimeTypes" type="as" access="read" />
|
|
</interface>
|
|
<interface name="org.mpris.MediaPlayer2.Player">
|
|
<method name="Next" />
|
|
<method name="Previous" />
|
|
<method name="Pause" />
|
|
<method name="PlayPause" />
|
|
<method name="Stop" />
|
|
<method name="Play" />
|
|
<method name="Seek">
|
|
<arg type="x" name="Offset" direction="in" />
|
|
</method>
|
|
<method name="SetPosition">
|
|
<arg type="o" name="TrackId" direction="in" />
|
|
<arg type="x" name="Position" direction="in" />
|
|
</method>
|
|
<method name="OpenUri">
|
|
<arg type="s" name="Uri" direction="in" />
|
|
</method>
|
|
<property name="PlaybackStatus" type="s" access="read">
|
|
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
|
</property>
|
|
<property name="LoopStatus" type="s" access="read" />
|
|
<property name="Rate" type="d" access="read" />
|
|
<property name="Shuffle" type="b" access="read" />
|
|
<property name="Metadata" type="a{{sv}}" access="read">
|
|
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
|
|
</property>
|
|
<property name="CanGoNext" type="b" access="read" />
|
|
<property name="CanGoPrevious" type="b" access="read" />
|
|
<property name="CanPlay" type="b" access="read" />
|
|
<property name="CanPause" type="b" access="read" />
|
|
<property name="CanControl" type="b" access="read" />
|
|
<property name="CanSeek" type="b" access="read" />
|
|
<property name="Position" type="x" access="read">
|
|
<annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false"/>
|
|
</property>
|
|
<property name="Volume" type="d" access="read" />
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
# Signal for org.freedesktop.DBus.Properties.PropertiesChanged
|
|
PropertiesChanged = signal()
|
|
|
|
def __init__(self):
|
|
self._title = "Winamp"
|
|
self._artist = "Unknown"
|
|
self._status = "Stopped"
|
|
self._last_position_us = 0
|
|
self._total_length_us = 0
|
|
self._last_update_ts = time.time()
|
|
|
|
def _request(self, endpoint):
|
|
try:
|
|
requests.get(f"{BASE_URL}/{endpoint}", auth=AUTH, timeout=2)
|
|
except Exception as e:
|
|
print(f"Request failed: {e}")
|
|
|
|
# MPRIS Methods
|
|
def Next(self): self._request("next")
|
|
def Previous(self): self._request("prev")
|
|
def Pause(self): self._request("pause")
|
|
def PlayPause(self): self._request("pause")
|
|
def Play(self): self._request("play")
|
|
def Stop(self): self._request("stop")
|
|
def Seek(self, offset): pass
|
|
def SetPosition(self, track_id, position): pass
|
|
def OpenUri(self, uri): pass
|
|
|
|
# --- Root Properties ---
|
|
@property
|
|
def CanQuit(self): return False
|
|
@property
|
|
def CanRaise(self): return False
|
|
@property
|
|
def HasTrackList(self): return False
|
|
@property
|
|
def Identity(self): return "Winamp Web Bridge"
|
|
@property
|
|
def DesktopEntry(self): return ""
|
|
@property
|
|
def SupportedUriSchemes(self): return []
|
|
@property
|
|
def SupportedMimeTypes(self): return []
|
|
|
|
# --- Player Properties ---
|
|
@property
|
|
def PlaybackStatus(self): return self._status
|
|
@property
|
|
def LoopStatus(self): return "None"
|
|
@property
|
|
def Rate(self): return 1.0
|
|
@property
|
|
def Shuffle(self): return False
|
|
@property
|
|
def CanGoNext(self): return True
|
|
@property
|
|
def CanGoPrevious(self): return True
|
|
@property
|
|
def CanPlay(self): return True
|
|
@property
|
|
def CanPause(self): return True
|
|
@property
|
|
def CanControl(self): return True
|
|
@property
|
|
def CanSeek(self): return False
|
|
|
|
@property
|
|
def Position(self):
|
|
if self._status == "Playing":
|
|
elapsed_us = int((time.time() - self._last_update_ts) * 1e6)
|
|
current_pos = self._last_position_us + elapsed_us
|
|
if self._total_length_us > 0:
|
|
return min(current_pos, self._total_length_us)
|
|
return current_pos
|
|
return self._last_position_us
|
|
|
|
@property
|
|
def Volume(self): return 1.0
|
|
|
|
@property
|
|
def Metadata(self):
|
|
# Metadata is a{sv}, so values MUST be wrapped in GLib.Variant
|
|
return {
|
|
'mpris:trackid': GLib.Variant('o', '/org/mpris/MediaPlayer2/winamp/track/0'),
|
|
'mpris:length': GLib.Variant('x', self._total_length_us),
|
|
'xesam:title': GLib.Variant('s', self._title),
|
|
'xesam:artist': GLib.Variant('as', [self._artist]),
|
|
'xesam:album': GLib.Variant('s', ''),
|
|
'mpris:artUrl': GLib.Variant('s', 'https://webamp.org/favicon.ico')
|
|
}
|
|
|
|
def parse_time_to_us(time_str):
|
|
"""Parses MM:SS or H:MM:SS to microseconds."""
|
|
parts = list(map(int, time_str.split(':')))
|
|
if len(parts) == 2: # MM:SS
|
|
return (parts[0] * 60 + parts[1]) * 1_000_000
|
|
elif len(parts) == 3: # H:MM:SS
|
|
return (parts[0] * 3600 + parts[1] * 60 + parts[2]) * 1_000_000
|
|
return 0
|
|
|
|
def update_loop(player):
|
|
last_known_title = ""
|
|
last_known_status = ""
|
|
offline_logged = False
|
|
|
|
while True:
|
|
try:
|
|
r = requests.get(f"{BASE_URL}/main", auth=AUTH, timeout=2)
|
|
if r.status_code == 200:
|
|
offline_logged = False
|
|
soup = BeautifulSoup(r.text, 'html.parser')
|
|
p_tags = soup.find_all('p')
|
|
status_raw = p_tags[0].text if p_tags else ""
|
|
|
|
new_status = "Stopped"
|
|
if "Playing" in status_raw: new_status = "Playing"
|
|
elif "Paused" in status_raw: new_status = "Paused"
|
|
|
|
new_artist = "Unknown"
|
|
new_title = "Winamp"
|
|
new_pos_us = 0
|
|
new_len_us = 0
|
|
|
|
# Metadata match
|
|
match_meta = re.search(r"track \d+ - (.*?) - (.*?) -", status_raw)
|
|
if match_meta:
|
|
new_artist = match_meta.group(1).strip()
|
|
new_title = match_meta.group(2).strip()
|
|
elif "track" in status_raw:
|
|
new_title = status_raw.split('-')[0].replace("Playing track ", "").strip()
|
|
|
|
# Time match: (1:06 / 2:48)
|
|
match_time = re.search(r"\((\d+:?\d*:\d+) / (\d+:?\d*:\d+)\)", status_raw)
|
|
if match_time:
|
|
new_pos_us = parse_time_to_us(match_time.group(1))
|
|
new_len_us = parse_time_to_us(match_time.group(2))
|
|
|
|
# Update internal state
|
|
player._status = new_status
|
|
player._artist = new_artist
|
|
player._title = new_title
|
|
player._last_position_us = new_pos_us
|
|
player._total_length_us = new_len_us
|
|
player._last_update_ts = time.time()
|
|
|
|
if new_status != last_known_status or new_title != last_known_title:
|
|
last_known_status = new_status
|
|
last_known_title = new_title
|
|
|
|
print(f"UPDATE: [{player._status}] {player._artist} - {player._title} ({match_time.group(1) if match_time else '0:00'})")
|
|
|
|
player.PropertiesChanged(
|
|
"org.mpris.MediaPlayer2.Player",
|
|
{
|
|
"PlaybackStatus": player.PlaybackStatus,
|
|
"Metadata": player.Metadata
|
|
},
|
|
[]
|
|
)
|
|
|
|
except (requests.exceptions.RequestException, Exception) as e:
|
|
if not offline_logged:
|
|
print(f"Winamp Web Interface offline or unreachable: {e}")
|
|
offline_logged = True
|
|
|
|
# Reset state when offline
|
|
player._status = "Stopped"
|
|
player._artist = "Unknown"
|
|
player._title = "Winamp (Offline)"
|
|
player._last_position_us = 0
|
|
player._total_length_us = 0
|
|
player._last_update_ts = time.time()
|
|
|
|
if last_known_status != "Stopped":
|
|
last_known_status = "Stopped"
|
|
player.PropertiesChanged(
|
|
"org.mpris.MediaPlayer2.Player",
|
|
{
|
|
"PlaybackStatus": player.PlaybackStatus,
|
|
"Metadata": player.Metadata
|
|
},
|
|
[]
|
|
)
|
|
|
|
time.sleep(2)
|
|
|
|
if __name__ == "__main__":
|
|
bus = SessionBus()
|
|
player_logic = WinampMPRIS()
|
|
bus.publish(APP_ID, ("/org/mpris/MediaPlayer2", player_logic))
|
|
|
|
# Delayed wake-up call
|
|
GLib.timeout_add(1500, lambda: player_logic.PropertiesChanged(
|
|
"org.mpris.MediaPlayer2.Player",
|
|
{
|
|
"PlaybackStatus": player_logic.PlaybackStatus,
|
|
"Metadata": player_logic.Metadata
|
|
},
|
|
[]
|
|
))
|
|
|
|
thread = Thread(target=update_loop, args=(player_logic,), daemon=True)
|
|
thread.start()
|
|
|
|
print(f"--- Winamp Bridge Started (Plasma Detection Fix) ---")
|
|
loop = GLib.MainLoop()
|
|
try:
|
|
loop.run()
|
|
except KeyboardInterrupt:
|
|
pass
|