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 class ADBAudioControl: def __init__(self): self.window = tk.Tk() self.window.title("Poweramp Remote") self.window.geometry("500x800") 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") self.files_list = [] self.selected_file_path = tk.StringVar(value="") # NEW: Add variable for track info and checkbox self.track_info_text = tk.StringVar(value="No track info available") self.fetch_track_info = tk.BooleanVar(value=True) # Checked by default # NEW: Add lock to prevent double keypress handling self.is_processing_keypress = False # 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 - FIXED: Changed to send command 1 (Poweramp API) instead of keyevent 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 Checkbox checkbox_frame = tk.Frame(track_info_frame, bg='#1a1a1a') checkbox_frame.pack(fill=tk.X, pady=(0, 10)) self.track_info_checkbox = tk.Checkbutton(checkbox_frame, text="Fetch track info after actions", variable=self.fetch_track_info, font=('Arial', 10), bg='#1a1a1a', fg='white', selectcolor='#333333', activebackground='#1a1a1a', activeforeground='white') self.track_info_checkbox.pack(side=tk.LEFT) # 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('<>', 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 - UPDATED 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)) # REMOVED: Tkinter window key bindings to prevent double triggering # self.window.bind("", self.on_key_press) 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 -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') 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') 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): """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") 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 - FIXED: 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 - FIXED: Proper 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" } self.update_status(f"Sending command: {commands.get(command, 'Unknown')}...") # 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: {commands.get(command, 'Unknown')}") 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()