Alcea
bcfa051e4a
Fixed Next Album breaking due to incorrect start command on album skip Added info for first run to instruct user to start adb !
762 lines
35 KiB
Python
762 lines
35 KiB
Python
import subprocess
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox, scrolledtext
|
|
from pynput import keyboard
|
|
import threading
|
|
import requests
|
|
from datetime import datetime
|
|
import os
|
|
import json
|
|
from urllib.parse import quote
|
|
import re
|
|
from plyer import notification
|
|
|
|
class ADBAudioControl:
|
|
def __init__(self):
|
|
self.window = tk.Tk()
|
|
self.window.title("Poweramp Remote")
|
|
self.window.geometry("500x900")
|
|
self.window.configure(bg='#1a1a1a')
|
|
|
|
# Initialize variables
|
|
self.ip_address = tk.StringVar(value="192.168.0.101")
|
|
self.port = tk.StringVar(value="5555")
|
|
self.status_text = tk.StringVar(value="Ready")
|
|
self.connection_status = tk.StringVar(value="Not Connected (Press RestartADB if this is first run")
|
|
self.files_list = []
|
|
self.selected_file_path = tk.StringVar(value="")
|
|
|
|
# Track info and checkboxes
|
|
self.track_info_text = tk.StringVar(value="No track info available")
|
|
self.fetch_track_info = tk.BooleanVar(value=True) # Checked by default
|
|
self.show_notifications = tk.BooleanVar(value=True) # Taskbar notifications checkbox
|
|
|
|
# Lock to prevent double keypress handling
|
|
self.is_processing_keypress = False
|
|
self.last_track_info = "" # To avoid duplicate notifications
|
|
self.pending_notification = None # Track if we're waiting for track info
|
|
|
|
# Set up keyboard listener
|
|
self.keyboard_listener = keyboard.Listener(on_press=self.handle_key_press)
|
|
self.keyboard_listener.start()
|
|
|
|
self.setup_ui()
|
|
self.auto_fetch_ip()
|
|
|
|
def setup_ui(self):
|
|
# Main container with padding
|
|
main_frame = tk.Frame(self.window, bg='#1a1a1a', padx=20, pady=20)
|
|
main_frame.pack(fill=tk.BOTH, expand=True)
|
|
|
|
# Title
|
|
title_label = tk.Label(main_frame,
|
|
text="Poweramp Remote",
|
|
font=('Arial', 24, 'bold'),
|
|
fg='white',
|
|
bg='#1a1a1a')
|
|
title_label.pack(pady=(0, 2))
|
|
|
|
# Connection Status
|
|
status_frame = tk.Frame(main_frame, bg='#2d2d2d', padx=10, pady=10)
|
|
status_frame.pack(fill=tk.X, pady=(0, 10))
|
|
|
|
status_label = tk.Label(status_frame,
|
|
text="Connection Status:",
|
|
font=('Arial', 10),
|
|
fg='#cccccc',
|
|
bg='#2d2d2d')
|
|
status_label.pack(side=tk.LEFT)
|
|
|
|
self.connection_status_label = tk.Label(status_frame,
|
|
textvariable=self.connection_status,
|
|
font=('Arial', 10, 'bold'),
|
|
fg='#ff6b6b',
|
|
bg='#2d2d2d')
|
|
self.connection_status_label.pack(side=tk.RIGHT)
|
|
|
|
# IP Address Section
|
|
ip_frame = tk.Frame(main_frame, bg='#1a1a1a')
|
|
ip_frame.pack(fill=tk.X, pady=(0, 15))
|
|
|
|
ip_label = tk.Label(ip_frame,
|
|
text="IP Address:",
|
|
font=('Arial', 11),
|
|
fg='white',
|
|
bg='#1a1a1a')
|
|
ip_label.pack(anchor=tk.W)
|
|
|
|
ip_entry = tk.Entry(ip_frame,
|
|
textvariable=self.ip_address,
|
|
font=('Arial', 11),
|
|
bg='#333333',
|
|
fg='white',
|
|
insertbackground='white',
|
|
relief=tk.FLAT)
|
|
ip_entry.pack(fill=tk.X, pady=(5, 0), ipady=8)
|
|
|
|
# Port Section
|
|
port_frame = tk.Frame(main_frame, bg='#1a1a1a')
|
|
port_frame.pack(fill=tk.X, pady=(0, 30))
|
|
|
|
port_label = tk.Label(port_frame,
|
|
text="Port:",
|
|
font=('Arial', 11),
|
|
fg='white',
|
|
bg='#1a1a1a')
|
|
port_label.pack(anchor=tk.W)
|
|
|
|
port_entry = tk.Entry(port_frame,
|
|
textvariable=self.port,
|
|
font=('Arial', 11),
|
|
bg='#333333',
|
|
fg='white',
|
|
insertbackground='white',
|
|
relief=tk.FLAT)
|
|
port_entry.pack(fill=tk.X, pady=(5, 0), ipady=8)
|
|
|
|
# Control Buttons Frame
|
|
control_frame = tk.Frame(main_frame, bg='#1a1a1a')
|
|
control_frame.pack(pady=(0, 20))
|
|
|
|
# Control Buttons
|
|
button_style = {
|
|
'font': ('Arial', 12, 'bold'),
|
|
'bg': '#404040',
|
|
'fg': 'white',
|
|
'activebackground': '#505050',
|
|
'activeforeground': 'white',
|
|
'relief': tk.FLAT,
|
|
'padx': 10,
|
|
'pady': 5,
|
|
'width': 5
|
|
}
|
|
|
|
# Previous Album Button
|
|
self.prev_album_button = tk.Button(control_frame,
|
|
text="⏪",
|
|
command=lambda: self.send_adb_command(6),
|
|
**button_style)
|
|
self.prev_album_button.grid(row=0, column=0, padx=2, pady=5)
|
|
|
|
# Previous Track Button
|
|
self.prev_button = tk.Button(control_frame,
|
|
text="⏮",
|
|
command=lambda: self.send_adb_command(5),
|
|
**button_style)
|
|
self.prev_button.grid(row=0, column=1, padx=2, pady=5)
|
|
|
|
# Play/Pause Button
|
|
self.play_pause_button = tk.Button(control_frame,
|
|
text="⏯",
|
|
command=lambda: self.send_adb_command(1),
|
|
**button_style)
|
|
self.play_pause_button.grid(row=0, column=2, padx=2, pady=5)
|
|
|
|
# Next Track Button
|
|
self.next_button = tk.Button(control_frame,
|
|
text="⏭",
|
|
command=lambda: self.send_adb_command(4),
|
|
**button_style)
|
|
self.next_button.grid(row=0, column=3, padx=2, pady=5)
|
|
|
|
# Next Album Button
|
|
self.next_album_button = tk.Button(control_frame,
|
|
text="⏩",
|
|
command=lambda: self.send_adb_command(7),
|
|
**button_style)
|
|
self.next_album_button.grid(row=0, column=4, padx=2, pady=5)
|
|
|
|
# Track Info Section
|
|
track_info_frame = tk.LabelFrame(main_frame,
|
|
text="Now Playing",
|
|
font=('Arial', 11, 'bold'),
|
|
bg='#1a1a1a',
|
|
fg='white',
|
|
padx=10,
|
|
pady=10)
|
|
track_info_frame.pack(fill=tk.X, pady=(0, 20))
|
|
|
|
# Track Info Checkboxes Frame
|
|
checkbox_frame = tk.Frame(track_info_frame, bg='#1a1a1a')
|
|
checkbox_frame.pack(fill=tk.X, pady=(0, 10))
|
|
|
|
# Left side checkboxes
|
|
left_checkbox_frame = tk.Frame(checkbox_frame, bg='#1a1a1a')
|
|
left_checkbox_frame.pack(side=tk.LEFT, anchor=tk.W)
|
|
|
|
self.track_info_checkbox = tk.Checkbutton(left_checkbox_frame,
|
|
text="Fetch track info",
|
|
variable=self.fetch_track_info,
|
|
font=('Arial', 10),
|
|
bg='#1a1a1a',
|
|
fg='white',
|
|
selectcolor='#333333',
|
|
activebackground='#1a1a1a',
|
|
activeforeground='white')
|
|
self.track_info_checkbox.pack(anchor=tk.W)
|
|
|
|
# Notification checkbox
|
|
self.notification_checkbox = tk.Checkbutton(left_checkbox_frame,
|
|
text="Show taskbar notifications",
|
|
variable=self.show_notifications,
|
|
font=('Arial', 10),
|
|
bg='#1a1a1a',
|
|
fg='white',
|
|
selectcolor='#333333',
|
|
activebackground='#1a1a1a',
|
|
activeforeground='white')
|
|
self.notification_checkbox.pack(anchor=tk.W, pady=(5, 0))
|
|
|
|
# Right side refresh button
|
|
refresh_button = tk.Button(checkbox_frame,
|
|
text="🔄 Refresh",
|
|
command=self.fetch_current_track_info,
|
|
font=('Arial', 9),
|
|
bg='#333333',
|
|
fg='white',
|
|
activebackground='#444444',
|
|
activeforeground='white',
|
|
relief=tk.FLAT,
|
|
padx=10,
|
|
pady=2)
|
|
refresh_button.pack(side=tk.RIGHT)
|
|
|
|
# Track Info Display
|
|
track_display_frame = tk.Frame(track_info_frame, bg='#2d2d2d', padx=10, pady=10)
|
|
track_display_frame.pack(fill=tk.X)
|
|
|
|
self.track_info_label = tk.Label(track_display_frame,
|
|
textvariable=self.track_info_text,
|
|
font=('Arial', 10),
|
|
fg='#4CAF50',
|
|
bg='#2d2d2d',
|
|
justify=tk.LEFT,
|
|
wraplength=400,
|
|
height=3)
|
|
self.track_info_label.pack(anchor=tk.W)
|
|
|
|
# File Picker Section
|
|
file_picker_frame = tk.LabelFrame(main_frame,
|
|
text="File Picker",
|
|
font=('Arial', 11, 'bold'),
|
|
bg='#1a1a1a',
|
|
fg='white',
|
|
padx=10,
|
|
pady=10)
|
|
file_picker_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 20))
|
|
|
|
# Load Files Button
|
|
load_files_button = tk.Button(file_picker_frame,
|
|
text="📂 Load Music Files",
|
|
command=self.load_music_files,
|
|
font=('Arial', 10),
|
|
bg='#2d2d2d',
|
|
fg='white',
|
|
activebackground='#3d3d3d',
|
|
activeforeground='white',
|
|
relief=tk.FLAT,
|
|
padx=10,
|
|
pady=4)
|
|
load_files_button.pack(pady=(0, 5))
|
|
|
|
# Send Button
|
|
send_button = tk.Button(file_picker_frame,
|
|
text="▶ Play Selected File",
|
|
command=self.send_selected_file,
|
|
font=('Arial', 11, 'bold'),
|
|
bg='#4CAF50',
|
|
fg='white',
|
|
activebackground='#45a049',
|
|
activeforeground='white',
|
|
relief=tk.FLAT,
|
|
padx=10,
|
|
pady=4)
|
|
send_button.pack(pady=(5, 0))
|
|
|
|
# Files Listbox with Scrollbar
|
|
listbox_frame = tk.Frame(file_picker_frame, bg='#1a1a1a')
|
|
listbox_frame.pack(fill=tk.BOTH, expand=True)
|
|
|
|
scrollbar = tk.Scrollbar(listbox_frame, bg='#333333')
|
|
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
|
|
|
|
self.files_listbox = tk.Listbox(listbox_frame,
|
|
yscrollcommand=scrollbar.set,
|
|
bg='#333333',
|
|
fg='white',
|
|
selectbackground='#4CAF50',
|
|
selectforeground='white',
|
|
font=('Arial', 9),
|
|
height=6)
|
|
self.files_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
|
self.files_listbox.bind('<<ListboxSelect>>', self.on_file_select)
|
|
|
|
scrollbar.config(command=self.files_listbox.yview)
|
|
|
|
# Selected File Info
|
|
selected_frame = tk.Frame(file_picker_frame, bg='#1a1a1a')
|
|
selected_frame.pack(fill=tk.X, pady=(10, 5))
|
|
|
|
selected_label = tk.Label(selected_frame,
|
|
text="Selected:",
|
|
font=('Arial', 9),
|
|
fg='#cccccc',
|
|
bg='#1a1a1a')
|
|
selected_label.pack(side=tk.LEFT)
|
|
|
|
self.selected_file_label = tk.Label(selected_frame,
|
|
textvariable=self.selected_file_path,
|
|
font=('Arial', 9),
|
|
fg='#4CAF50',
|
|
bg='#1a1a1a',
|
|
wraplength=350)
|
|
self.selected_file_label.pack(side=tk.LEFT, padx=(5, 0))
|
|
|
|
# Restart Button
|
|
self.restart_button = tk.Button(main_frame,
|
|
text="🔄 Restart ADB",
|
|
command=self.restart,
|
|
font=('Arial', 11),
|
|
bg='#2d2d2d',
|
|
fg='white',
|
|
activebackground='#3d3d3d',
|
|
activeforeground='white',
|
|
relief=tk.FLAT,
|
|
padx=20,
|
|
pady=12)
|
|
self.restart_button.pack(pady=(0, 10))
|
|
|
|
# Fetch IP Button
|
|
self.fetch_button = tk.Button(main_frame,
|
|
text="📡 Fetch IP Automatically",
|
|
command=self.auto_fetch_ip,
|
|
font=('Arial', 11),
|
|
bg='#2d2d2d',
|
|
fg='white',
|
|
activebackground='#3d3d3d',
|
|
activeforeground='white',
|
|
relief=tk.FLAT,
|
|
padx=20,
|
|
pady=12)
|
|
self.fetch_button.pack(pady=(0, 20))
|
|
|
|
# Status Frame
|
|
status_display_frame = tk.Frame(main_frame, bg='#2d2d2d', padx=15, pady=15)
|
|
status_display_frame.pack(fill=tk.X)
|
|
|
|
status_title = tk.Label(status_display_frame,
|
|
text="Status",
|
|
font=('Arial', 10, 'bold'),
|
|
fg='#cccccc',
|
|
bg='#2d2d2d')
|
|
status_title.pack(anchor=tk.W)
|
|
|
|
self.status_label = tk.Label(status_display_frame,
|
|
textvariable=self.status_text,
|
|
font=('Arial', 10),
|
|
fg='#4CAF50',
|
|
bg='#2d2d2d',
|
|
justify=tk.LEFT,
|
|
wraplength=400)
|
|
self.status_label.pack(anchor=tk.W, pady=(5, 0))
|
|
|
|
# Key Bindings Info
|
|
keybind_frame = tk.Frame(main_frame, bg='#1a1a1a', padx=10, pady=10)
|
|
keybind_frame.pack(fill=tk.X, pady=(10, 0))
|
|
|
|
keybind_title = tk.Label(keybind_frame,
|
|
text="Media Key Bindings (pynput only):",
|
|
font=('Arial', 10, 'bold'),
|
|
fg='#cccccc',
|
|
bg='#1a1a1a')
|
|
keybind_title.pack(anchor=tk.W)
|
|
|
|
keybind_info = tk.Label(keybind_frame,
|
|
text="• Play/Pause: Media Play/Pause Key\n• Next Track: Media Next Key\n• Previous Track: Media Previous Key",
|
|
font=('Arial', 9),
|
|
fg='#999999',
|
|
bg='#1a1a1a',
|
|
justify=tk.LEFT)
|
|
keybind_info.pack(anchor=tk.W, pady=(5, 0))
|
|
|
|
# Version/Attribution
|
|
attribution = tk.Label(main_frame,
|
|
text="Poweramp Remote",
|
|
font=('Arial', 8),
|
|
fg='#666666',
|
|
bg='#1a1a1a')
|
|
attribution.pack(side=tk.BOTTOM, pady=(10, 0))
|
|
|
|
def show_notification(self, title, message):
|
|
"""Show taskbar notification"""
|
|
try:
|
|
notification.notify(
|
|
title=title,
|
|
message=message,
|
|
app_name="Poweramp Remote",
|
|
timeout=3, # 3 seconds
|
|
app_icon=None # You can add an icon file path here if desired
|
|
)
|
|
except Exception as e:
|
|
print(f"Error showing notification: {e}")
|
|
|
|
def encode_uri_like_js(self, uri):
|
|
"""Encode URI like JavaScript's encodeURI function"""
|
|
safe_chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789;,/?:@&=+$-_.!~*'()#"
|
|
|
|
encoded_parts = []
|
|
for char in uri:
|
|
if char in safe_chars:
|
|
encoded_parts.append(char)
|
|
elif char == ' ':
|
|
encoded_parts.append('%20')
|
|
elif char == "'":
|
|
encoded_parts.append("%27")
|
|
else:
|
|
encoded_parts.append(quote(char, safe=''))
|
|
|
|
return ''.join(encoded_parts)
|
|
|
|
def load_music_files(self):
|
|
"""Load music files from the remote URL"""
|
|
list_url = 'https://alcea-wisteria.de/PHP/0demo/2025-12-22-DeviceNAppCtrl/[Android]_Poweramp%E2%80%90(Tasker)(20251222)/0ld/list.txt'
|
|
|
|
def fetch_files():
|
|
try:
|
|
self.update_status("Loading music files list...")
|
|
response = requests.get(list_url, timeout=10)
|
|
if response.status_code == 200:
|
|
lines = response.text.strip().split('\n')
|
|
self.files_list = [line.strip() for line in lines if line.strip()]
|
|
|
|
self.files_listbox.delete(0, tk.END)
|
|
|
|
for file_path in self.files_list:
|
|
display_name = self.extract_display_name(file_path)
|
|
self.files_listbox.insert(tk.END, display_name)
|
|
|
|
self.update_status(f"Loaded {len(self.files_list)} music files")
|
|
else:
|
|
self.update_status(f"Failed to load files. Status code: {response.status_code}", "error")
|
|
except requests.exceptions.RequestException as e:
|
|
self.update_status(f"Network error: {str(e)}", "error")
|
|
except Exception as e:
|
|
self.update_status(f"Error loading files: {str(e)}", "error")
|
|
|
|
threading.Thread(target=fetch_files, daemon=True).start()
|
|
|
|
def extract_display_name(self, file_path):
|
|
"""Extract display name from file path (matches HTML logic)"""
|
|
if '/mobile/' in file_path:
|
|
parts = file_path.split('/mobile/')
|
|
if len(parts) > 1:
|
|
return parts[1]
|
|
path_parts = [part for part in file_path.split('/') if part.strip() != '']
|
|
return '/'.join(path_parts[-2:]) if len(path_parts) >= 2 else file_path
|
|
|
|
def convert_file_path(self, file_path):
|
|
"""Convert file path from /mnt/media_rw to /sdcard (matches HTML logic)"""
|
|
return file_path.replace('/mnt/media_rw', '/sdcard')
|
|
|
|
def on_file_select(self, event):
|
|
"""Handle file selection from listbox"""
|
|
selection = self.files_listbox.curselection()
|
|
if selection:
|
|
index = selection[0]
|
|
if 0 <= index < len(self.files_list):
|
|
original_path = self.files_list[index]
|
|
display_name = self.extract_display_name(original_path)
|
|
self.selected_file_path.set(display_name)
|
|
|
|
self.files_listbox.selected_index = index
|
|
|
|
def send_selected_file(self):
|
|
"""Send selected file to Poweramp via ADB"""
|
|
if not hasattr(self.files_listbox, 'selected_index'):
|
|
self.update_status("No file selected", "error")
|
|
return
|
|
|
|
index = self.files_listbox.selected_index
|
|
if 0 <= index < len(self.files_list):
|
|
original_path = self.files_list[index]
|
|
self.play_file_via_adb(original_path)
|
|
|
|
def play_file_via_adb(self, original_path):
|
|
"""Play a file via ADB using EXACTLY the method from the HTML"""
|
|
try:
|
|
ip_address = self.ip_address.get()
|
|
port = self.port.get()
|
|
|
|
if not ip_address or not port:
|
|
self.update_status("Error: IP and Port must be specified", "error")
|
|
return
|
|
|
|
android_path = self.convert_file_path(original_path)
|
|
file_url = f"file://{android_path}"
|
|
encoded_file_url = self.encode_uri_like_js(file_url)
|
|
adb_command = f'am start -a android.intent.action.VIEW -d "{encoded_file_url}" -t "audio/mpeg" -n com.maxmpz.audioplayer/.MainActivity'
|
|
# adb_command = f'am start -n com.maxmpz.audioplayer/.PlayerUIActivity -a android.intent.action.VIEW -d "{encoded_file_url}" -t audio/mpeg'
|
|
full_adb_command = f'adb -s {ip_address}:{port} shell "{adb_command}"'
|
|
|
|
display_name = self.extract_display_name(original_path)
|
|
self.update_status(f"Sending file to Poweramp: {display_name}...")
|
|
|
|
def run_command():
|
|
try:
|
|
result = subprocess.run(full_adb_command, shell=True, capture_output=True, text=True, timeout=10)
|
|
|
|
if result.returncode == 0:
|
|
self.update_status(f"Playing: {display_name}")
|
|
self.connection_status.set("Connected")
|
|
self.connection_status_label.config(fg='#4CAF50')
|
|
|
|
# Show notification with file name immediately
|
|
if self.show_notifications.get():
|
|
self.show_notification("Poweramp Remote", f"Now Playing:\n{display_name}")
|
|
|
|
# Fetch track info if enabled
|
|
if self.fetch_track_info.get():
|
|
self.fetch_current_track_info()
|
|
|
|
else:
|
|
alt_command = f'adb -s {ip_address}:{port} shell \'{adb_command}\''
|
|
alt_result = subprocess.run(alt_command, shell=True, capture_output=True, text=True, timeout=10)
|
|
|
|
if alt_result.returncode == 0:
|
|
self.update_status(f"Playing: {display_name}")
|
|
self.connection_status.set("Connected")
|
|
self.connection_status_label.config(fg='#4CAF50')
|
|
|
|
# Show notification with file name immediately
|
|
if self.show_notifications.get():
|
|
self.show_notification("Poweramp Remote", f"Now Playing:\n{display_name}")
|
|
|
|
if self.fetch_track_info.get():
|
|
self.fetch_current_track_info()
|
|
|
|
else:
|
|
error_msg = alt_result.stderr if alt_result.stderr else alt_result.stdout
|
|
self.update_status(f"Error: {error_msg[:100]}...", "error")
|
|
self.connection_status.set("Connection Failed")
|
|
|
|
except subprocess.TimeoutExpired:
|
|
self.update_status("Command timed out", "error")
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
threading.Thread(target=run_command, daemon=True).start()
|
|
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
def fetch_current_track_info(self, from_command=False):
|
|
"""Fetch current playing track info using adb shell dumpsys media_session"""
|
|
try:
|
|
ip_address = self.ip_address.get()
|
|
port = self.port.get()
|
|
|
|
if not ip_address or not port:
|
|
self.update_status("Error: IP and Port must be specified", "error")
|
|
return
|
|
|
|
dumpsys_command = f'adb -s {ip_address}:{port} shell "dumpsys media_session | grep \'description=\' | sed \'s/.*description=\\([^,]*\\), \\([^,]*\\),.*/\\1 - \\2/\'"'
|
|
|
|
def run_track_info():
|
|
try:
|
|
result = subprocess.run(dumpsys_command, shell=True, capture_output=True, text=True, timeout=5)
|
|
|
|
if result.returncode == 0:
|
|
output = result.stdout.strip()
|
|
if output:
|
|
track_info = output.strip()
|
|
self.track_info_text.set(track_info)
|
|
self.update_status(f"Track info updated")
|
|
|
|
# Show notification if enabled and track info changed
|
|
if (self.show_notifications.get() and
|
|
track_info != self.last_track_info and
|
|
track_info != "No track info available (not playing)"):
|
|
|
|
# Only show "Now Playing" notifications (not generic action notifications)
|
|
self.show_notification("Poweramp Remote", f"Now Playing:\n{track_info}")
|
|
self.last_track_info = track_info
|
|
|
|
else:
|
|
self.track_info_text.set("No track info available (not playing)")
|
|
else:
|
|
self.track_info_text.set(f"Error fetching track info")
|
|
|
|
except subprocess.TimeoutExpired:
|
|
self.track_info_text.set("Track info fetch timed out")
|
|
except Exception as e:
|
|
self.track_info_text.set(f"Error: {str(e)[:100]}")
|
|
|
|
threading.Thread(target=run_track_info, daemon=True).start()
|
|
|
|
except Exception as e:
|
|
self.track_info_text.set(f"Error: {str(e)[:100]}")
|
|
|
|
def handle_key_press(self, key):
|
|
"""Handle media key presses - Only use pynput, no double triggers"""
|
|
# Prevent rapid-fire keypresses
|
|
if self.is_processing_keypress:
|
|
return
|
|
|
|
self.is_processing_keypress = True
|
|
|
|
try:
|
|
# Standard media keys
|
|
if key == keyboard.Key.media_play_pause:
|
|
self.send_adb_command(1) # Poweramp API command for play/pause
|
|
elif key == keyboard.Key.media_next:
|
|
self.send_adb_command(4) # Poweramp API command for next track
|
|
elif key == keyboard.Key.media_previous:
|
|
self.send_adb_command(5) # Poweramp API command for previous track
|
|
|
|
# Album navigation - NOT standard media keys, use number keys if needed
|
|
# elif hasattr(key, 'char') and key.char == '7':
|
|
# self.send_adb_command(7) # Next album
|
|
# elif hasattr(key, 'char') and key.char == '6':
|
|
# self.send_adb_command(6) # Previous album
|
|
|
|
except Exception as e:
|
|
print(f"Key press error: {e}")
|
|
finally:
|
|
# Reset after a short delay to prevent rapid-fire presses
|
|
self.window.after(300, lambda: setattr(self, 'is_processing_keypress', False))
|
|
|
|
def send_adb_command(self, command):
|
|
"""Send standard Poweramp API commands with debouncing"""
|
|
try:
|
|
ip_address = self.ip_address.get()
|
|
port = self.port.get()
|
|
|
|
if not ip_address or not port:
|
|
self.update_status("Error: IP and Port must be specified", "error")
|
|
return
|
|
|
|
adb_shell_command = f"adb -s {ip_address}:{port} shell am broadcast 'intent:#Intent;action=com.maxmpz.audioplayer.API_COMMAND;package=com.maxmpz.audioplayer;i.cmd={command};end'"
|
|
|
|
# Update status
|
|
commands = {
|
|
1: "Play/Pause",
|
|
4: "Next Track",
|
|
5: "Previous Track",
|
|
6: "Previous Album",
|
|
7: "Next Album"
|
|
}
|
|
command_name = commands.get(command, 'Unknown')
|
|
self.update_status(f"Sending command: {command_name}...")
|
|
|
|
# REMOVED: Generic notification for track navigation - we'll wait for actual track info
|
|
|
|
# Run command in thread to prevent UI freezing
|
|
def run_command():
|
|
try:
|
|
result = subprocess.run(adb_shell_command, shell=True, capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
self.update_status(f"Command sent successfully: {command_name}")
|
|
self.connection_status.set("Connected")
|
|
self.connection_status_label.config(fg='#4CAF50')
|
|
|
|
# Fetch track info after command if checkbox is checked
|
|
if self.fetch_track_info.get():
|
|
import time
|
|
time.sleep(0.5)
|
|
self.fetch_current_track_info()
|
|
|
|
else:
|
|
self.update_status(f"Error sending command: {result.stderr}", "error")
|
|
self.connection_status.set("Connection Failed")
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
threading.Thread(target=run_command, daemon=True).start()
|
|
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
def restart(self):
|
|
"""Restart ADB server"""
|
|
try:
|
|
ip_address = self.ip_address.get()
|
|
port = self.port.get()
|
|
|
|
if not ip_address or not port:
|
|
self.update_status("Error: IP and Port must be specified", "error")
|
|
return
|
|
|
|
command = f'adb kill-server && adb start-server && adb connect {ip_address}:{port}'
|
|
|
|
self.update_status("Restarting ADB server...")
|
|
|
|
def run_restart():
|
|
try:
|
|
subprocess.Popen(['cmd.exe', '/c', 'start', 'cmd.exe', '/k', command])
|
|
self.update_status("ADB restart initiated in new terminal")
|
|
except Exception as e:
|
|
self.update_status(f"Error restarting ADB: {str(e)}", "error")
|
|
|
|
threading.Thread(target=run_restart, daemon=True).start()
|
|
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
def auto_fetch_ip(self):
|
|
"""Fetch IP from the remote URL"""
|
|
url = "https://alcea-wisteria.de/PHP/0demo/2025-12-22-DeviceNAppCtrl/%5BAndroid%5D_General-Fetch-DeviceIP-%2820251223%29/address.txt"
|
|
|
|
def fetch():
|
|
try:
|
|
self.update_status("Fetching IP address from remote server...")
|
|
response = requests.get(url, timeout=10)
|
|
if response.status_code == 200:
|
|
ip = response.text.strip()
|
|
if ip:
|
|
self.ip_address.set(ip)
|
|
self.update_status(f"IP fetched successfully: {ip}")
|
|
else:
|
|
self.update_status("No IP address found in response", "warning")
|
|
else:
|
|
self.update_status(f"Failed to fetch IP. Status code: {response.status_code}", "error")
|
|
except requests.exceptions.RequestException as e:
|
|
self.update_status(f"Network error: {str(e)}", "error")
|
|
except Exception as e:
|
|
self.update_status(f"Error: {str(e)}", "error")
|
|
|
|
threading.Thread(target=fetch, daemon=True).start()
|
|
|
|
def update_status(self, message, msg_type="info"):
|
|
"""Update status message with color coding"""
|
|
colors = {
|
|
"info": "#4CAF50",
|
|
"error": "#ff6b6b",
|
|
"warning": "#ffa726"
|
|
}
|
|
|
|
timestamp = datetime.now().strftime("%H:%M:%S")
|
|
self.status_text.set(f"[{timestamp}] {message}")
|
|
self.status_label.config(fg=colors.get(msg_type, "#4CAF50"))
|
|
|
|
# Auto-clear info messages after 5 seconds
|
|
if msg_type == "info":
|
|
self.window.after(5000, lambda: self.clear_status_if_old(timestamp))
|
|
|
|
def clear_status_if_old(self, timestamp):
|
|
"""Clear status if it's still showing the old timestamp"""
|
|
current_status = self.status_text.get()
|
|
if f"[{timestamp}]" in current_status:
|
|
self.status_text.set("Ready")
|
|
self.status_label.config(fg="#4CAF50")
|
|
|
|
def run(self):
|
|
self.window.mainloop()
|
|
|
|
def __del__(self):
|
|
if hasattr(self, 'keyboard_listener'):
|
|
self.keyboard_listener.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = ADBAudioControl()
|
|
app.run() |