11 Commits
2 changed files with 172 additions and 73 deletions
+1 -1
View File
@@ -8,7 +8,7 @@ Simply run the application, and a local web server will run in the background. B
[http://127.0.0.1:5000/now-playing/](http://127.0.0.1:5000/now-playing/) [http://127.0.0.1:5000/now-playing/](http://127.0.0.1:5000/now-playing/)
A ready-to-use "Now Playing" widget utilizing SMTC Bridge is available to try here:<br> A ready-to-use "Now Playing" widget utilizing SMTC Bridge is available to try here:<br>
**[https://widgets.nutty.gg/now-playing/settings/](https://widgets.nutty.gg/now-playing/settings/)** **[IF YOU FOUND THIS PAGE EARLY, DON'T SHARE IT WITH ANYONE YET OR I'LL FUCKING CUM](https://widgets.nutty.gg/now-playing/settings/)**
## Quick Start ## Quick Start
1. Download the latest `.exe` from the [Releases page](releases). 1. Download the latest `.exe` from the [Releases page](releases).
+163 -64
View File
@@ -1,5 +1,5 @@
# Versioning # Versioning
APP_VERSION = "0.0.2" APP_VERSION = "0.0.6"
DEVELOPER = "nutty" DEVELOPER = "nutty"
@@ -62,6 +62,9 @@ try:
import pystray import pystray
import webbrowser import webbrowser
import configparser import configparser
import time
import platform
import socket
from PIL import Image from PIL import Image
from flask import Flask, jsonify from flask import Flask, jsonify
from flask_cors import CORS from flask_cors import CORS
@@ -144,9 +147,25 @@ try:
HOST = settings.get('SERVER', 'Host') HOST = settings.get('SERVER', 'Host')
PORT = settings.getint('SERVER', 'Port') PORT = settings.getint('SERVER', 'Port')
def get_local_ip():
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "127.0.0.1"
DISPLAY_HOST = get_local_ip() if HOST == "0.0.0.0" else HOST
app = Flask(__name__) app = Flask(__name__)
CORS(app) CORS(app)
ARTWORK_CACHE = {}
# Create a local temp directory for thumbnails
THUMB_DIR = os.path.join(tempfile.gettempdir(), 'smtc_bridge_thumbs')
if not os.path.exists(THUMB_DIR):
os.makedirs(THUMB_DIR)
@@ -154,30 +173,66 @@ try:
### CORE FUNCTIONS ### ### CORE FUNCTIONS ###
###################### ######################
async def get_all_media_info(): # Cache variables for the SMTC manager and rate limiting
global ARTWORK_CACHE smtc_manager = None
last_execution_time = 0.0
last_payload = None
thumb_cache = {}
async def get_all_media_info():
try: try:
# We will cache the last execution time and payload to avoid redundant parsing if requests flood in faster than 0.5 seconds.
current_time = time.time()
global smtc_manager, last_execution_time, last_payload, thumb_cache
# If requests flood in faster than 0.5 seconds, return the cached result
# to completely spare the CPU from redundant parsing.
if (current_time - last_execution_time) < 0.5 and last_payload:
# print("Using cached media info.")
return last_payload
# else:
# print("Fetching fresh media info.")
last_execution_time = current_time
# Instantiate the SMTC manager -> This allows use to "talk" to the Windows Media API # Instantiate the SMTC manager -> This allows use to "talk" to the Windows Media API
manager = await SMTC.request_async() # Reuse the manager if we already have it, otherwise request it once
if not smtc_manager:
print("Instantiating new SMTC manager...")
smtc_manager = await SMTC.request_async()
manager = smtc_manager
# If it returns null, then no media is playing, or something fucked up and I have no # If it returns null, then no media is playing, or something fucked up and I have no
# idea what to do, so just return an empty session list # idea what to do, so just return an empty session list
if not manager: if not manager:
return {"current_session_id": None, "sessions": []} return {"current_session_id": None, "sessions": []}
# This is the session for the current media player -> Whatever Windows deems is "in focus" # current_focused: This is the session for the current media player -> Whatever Windows deems is "in focus"
# will be the current session. # will be the current session.
# all_sessions: We will also get all sessions, not just the current session.
# This will provide the client with all the necessary info if they want to target just one application.
# We will store the SourceAppUserModelId, which we all add to the final payload. # We will store the SourceAppUserModelId, which we all add to the final payload.
# For all available properties/methods/events, see the official docs: # For all available properties/methods/events, see the official docs:
# https://learn.microsoft.com/en-us/uwp/api/windows.media.control.globalsystemmediatransportcontrolssession?view=winrt-28000 # https://learn.microsoft.com/en-us/uwp/api/windows.media.control.globalsystemmediatransportcontrolssession?view=winrt-28000
# Try to pull sessions using the cached manager
try:
current_focused = manager.get_current_session() current_focused = manager.get_current_session()
all_sessions = manager.get_sessions()
except Exception:
# If the COM context dropped or invalidated, reset it and retry once
print("Instantiating new SMTC manager...")
smtc_manager = await SMTC.request_async()
manager = smtc_manager
if not manager:
return {"current_session_id": None, "sessions": []}
current_focused = manager.get_current_session()
all_sessions = manager.get_sessions()
current_session_id = current_focused.source_app_user_model_id if current_focused else None current_session_id = current_focused.source_app_user_model_id if current_focused else None
# We will also get all sessions, not just the current session. # We will store all the session info in a list of dictionaries, which we will return as JSON
# This will provide the client with all the necessary info if they want to target just
# one application.
all_sessions = manager.get_sessions()
sessions_list = [] sessions_list = []
# We will not iterate over all the sessions, and grab all the available info # We will not iterate over all the sessions, and grab all the available info
@@ -262,67 +317,44 @@ try:
"Thumbnail": None "Thumbnail": None
} }
# CACHE THE THUMBNAIL, DON'T ENCODE IT EVERY TIME, IT'S SLOW AS FUCK # Read thumbnail and convert straight to Base64
# # Add the album art (i.e. "Thumbnail") as a Base64 string. thumb_url = None
# # We will cache this so we don't have to read it every single call.
# # Use the track title and artist as the key -> {title} - {artist}
# # Build the key
# title = raw_media.title if raw_media else "Unknown"
# artist = raw_media.artist if raw_media else "Unknown"
# track_key = f"{title} - {artist}"
# # Check if the album art is cached.
# # If yes:
# # - Retrieve it from the cache.
# # - Add it to the payload
# if app_id in ARTWORK_CACHE and ARTWORK_CACHE[app_id]["track_key"] == track_key:
# media_data["Thumbnail"] = ARTWORK_CACHE[app_id]["base64"]
# # If not
# # - Read the Base64 string
# # - Cache the Base64 string
# # - Add it to the payload
# elif raw_media and raw_media.thumbnail:
# try:
# # Read the Base64 string
# stream_ref = raw_media.thumbnail
# stream = await stream_ref.open_read_async()
# reader = DataReader(stream.get_input_stream_at(0))
# await reader.load_async(stream.size)
# buffer = bytearray(stream.size)
# reader.read_bytes(buffer)
# img = Image.open(io.BytesIO(buffer))
# base64_art = f"data:image/png;base64,{base64.b64encode(buffer).decode('utf-8')}"
# # Cache the Base64 string
# ARTWORK_CACHE[app_id] = {"track_key": track_key, "base64": base64_art}
# # Add it to the payload
# media_data["Thumbnail"] = base64_art
# except Exception:
# pass
# Read the Base64 string
base64_art = None
if raw_media and raw_media.thumbnail: if raw_media and raw_media.thumbnail:
try: try:
stream_ref = raw_media.thumbnail stream_ref = raw_media.thumbnail
stream = await stream_ref.open_read_async() stream = await stream_ref.open_read_async()
if stream.size > 0:
reader = DataReader(stream.get_input_stream_at(0)) reader = DataReader(stream.get_input_stream_at(0))
await reader.load_async(stream.size) await reader.load_async(stream.size)
buffer = bytearray(stream.size) buffer = bytearray(stream.size)
reader.read_bytes(buffer) reader.read_bytes(buffer)
img = Image.open(io.BytesIO(buffer)) import hashlib
base64_art = f"data:image/png;base64,{base64.b64encode(buffer).decode('utf-8')}" import base64
except Exception:
# Fallback if the stream fails to open or read
base64_art = None
# Add it to the payload # Hash the raw bytes to see if the artwork is unique
media_data["Thumbnail"] = base64_art img_hash = hashlib.md5(buffer).hexdigest()
# If we've already processed this exact image bytes, grab it from cache instantly
if img_hash in thumb_cache:
thumb_url = thumb_cache[img_hash]
else:
print(f"Processing new artwork for: {media_data['Artist']} - {media_data['Title']}")
# Convert raw Windows bytes straight to Base64
encoded_img = base64.b64encode(bytes(buffer)).decode('utf-8')
thumb_url = f"data:image/jpeg;base64,{encoded_img}"
# Store it in the cache so we never process it again for this track
thumb_cache[img_hash] = thumb_url
except Exception as e:
thumb_url = None
# Add the Base64 data string to the payload
media_data["Thumbnail"] = thumb_url
# Finally, assemble all the data into a session object, and add it to the session list # Finally, assemble all the data into a session object, and add it to the session list
sessions_list.append({ sessions_list.append({
@@ -333,20 +365,81 @@ try:
}) })
# Build the final payload # Build the final payload
return { payload = {
"app_version": APP_VERSION, "app_version": APP_VERSION,
"os": f"{platform.system()} {platform.release()}",
"current_session_id": current_session_id, "current_session_id": current_session_id,
"sessions": sessions_list "sessions": sessions_list
} }
# Save to global payload cache
last_payload = payload
return payload
except Exception as e: except Exception as e:
return {"current_session_id": None, "sessions": [], "error": str(e)} return {"current_session_id": None, "sessions": [], "error": str(e)}
########################
### STARTUP SHORTCUT ###
########################
def get_startup_shortcut_path():
# Gets the path to the current user's Startup folder
startup_dir = os.path.join(os.environ['APPDATA'], r'Microsoft\Windows\Start Menu\Programs\Startup')
return os.path.join(startup_dir, 'SMTCBridge.lnk')
def is_start_with_windows():
return os.path.exists(get_startup_shortcut_path())
def set_start_with_windows(enable: bool):
shortcut_path = get_startup_shortcut_path()
if enable:
# Determine the correct path (handles both raw script and PyInstaller .exe)
if getattr(sys, 'frozen', False):
target_path = sys.executable
working_dir = os.path.dirname(sys.executable)
else:
target_path = sys.executable # python.exe
working_dir = os.path.dirname(os.path.abspath(__file__))
# If running as script, you might want to point to the script instead,
# but usually this feature is intended for the built .exe
# Use a quick PowerShell command to create a proper Windows .lnk shortcut
# This avoids needing external libraries like winshell
script_args = f"""
$WshShell = New-Object -ComObject WScript.Shell
$Shortcut = $WshShell.CreateShortcut('{shortcut_path}')
$Shortcut.TargetPath = '{target_path}'
$Shortcut.WorkingDirectory = '{working_dir}'
$Shortcut.Save()
"""
import subprocess
subprocess.run(["powershell", "-Command", script_args], capture_output=True, creationflags=subprocess.CREATE_NO_WINDOW)
else:
if os.path.exists(shortcut_path):
try:
os.remove(shortcut_path)
except OSError:
pass
def toggle_startup(icon, item):
current_state = is_start_with_windows()
set_start_with_windows(not current_state)
################# #################
### ENDPOINTS ### ### ENDPOINTS ###
################# #################
@app.route('/artwork/<app_identifier>')
def serve_artwork(app_identifier):
from flask import send_from_directory
# Serve the cached thumbnail directly from the temp directory
return send_from_directory(THUMB_DIR, f"{app_identifier}.jpg")
@app.route('/now-playing') @app.route('/now-playing')
def now_playing(): def now_playing():
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
@@ -410,12 +503,18 @@ try:
menu = pystray.Menu( menu = pystray.Menu(
pystray.MenuItem(f"SMTC Bridge v{APP_VERSION} by {DEVELOPER}", None, enabled=False), pystray.MenuItem(f"SMTC Bridge v{APP_VERSION} by {DEVELOPER}", None, enabled=False),
pystray.Menu.SEPARATOR, pystray.Menu.SEPARATOR,
pystray.MenuItem("View Data (JSON)", lambda: webbrowser.open(f"http://{HOST}:{PORT}/now-playing")), pystray.MenuItem("View Data (JSON)", lambda: webbrowser.open(f"http://{DISPLAY_HOST}:{PORT}/now-playing")),
pystray.MenuItem("View Active Sessions", lambda: webbrowser.open(f"http://{HOST}:{PORT}/sessions")), pystray.MenuItem("View Active Sessions", lambda: webbrowser.open(f"http://{DISPLAY_HOST}:{PORT}/sessions")),
pystray.Menu.SEPARATOR, pystray.Menu.SEPARATOR,
pystray.MenuItem("★ Customize Overlay", lambda: webbrowser.open(f"https://widgets.nutty.gg/now-playing/settings/")), pystray.MenuItem("★ Customize Overlay", lambda: webbrowser.open(f"https://widgets.nutty.gg/now-playing/settings/")),
pystray.MenuItem("★ Try my stream widgets!", lambda: webbrowser.open(f"https://nutty.gg/collections/member-exclusive-widgets")), pystray.MenuItem("★ Try my stream widgets!", lambda: webbrowser.open(f"https://nutty.gg/collections/member-exclusive-widgets")),
pystray.Menu.SEPARATOR, pystray.Menu.SEPARATOR,
pystray.MenuItem(
"Start with Windows",
toggle_startup,
checked=lambda item: is_start_with_windows()
),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Quit", on_quit) pystray.MenuItem("Quit", on_quit)
) )