From 6d08df1807cdf1c3023fdc2650ddfc0159e43e9b Mon Sep 17 00:00:00 2001 From: Jason Ross Date: Mon, 22 Sep 2025 14:08:30 -0500 Subject: [PATCH] main window refactor --- src/gui/animation_handler.py | 124 ++++ src/gui/device_manager.py | 172 +++++ src/gui/dialog_manager.py | 430 ++++++++++++ src/gui/file_browser.py | 23 +- src/gui/main_window.py | 1268 ++++++++-------------------------- src/gui/transfer_manager.py | 394 +++++++++++ src/gui/ui_components.py | 319 +++++++++ 7 files changed, 1724 insertions(+), 1006 deletions(-) create mode 100644 src/gui/animation_handler.py create mode 100644 src/gui/device_manager.py create mode 100644 src/gui/dialog_manager.py create mode 100644 src/gui/transfer_manager.py create mode 100644 src/gui/ui_components.py diff --git a/src/gui/animation_handler.py b/src/gui/animation_handler.py new file mode 100644 index 0000000..859ef4a --- /dev/null +++ b/src/gui/animation_handler.py @@ -0,0 +1,124 @@ +""" +Animation Handler Module +Manages all GUI animations for the Android file transfer application. +""" + +import tkinter as tk +from typing import Optional, Dict, Any + + +class AnimationHandler: + """Handles all GUI animations including transfer and scanning animations.""" + + def __init__(self, parent_window: tk.Tk): + """Initialize the animation handler. + + Args: + parent_window: The main window instance + """ + self.parent = parent_window + self.animation_job: Optional[str] = None + self.animation_dots = 0 + self.scanning_active = False + + # File transfer progress tracking + self.transfer_file_progress = { + 'current': 0, + 'total': 0, + 'active': False + } + + def start_scanning_animation(self) -> None: + """Start the 'Scanning for duplicates...' animation.""" + self.animation_dots = 0 + self.scanning_active = True + self.animation_job = self.parent.after(0, self._animate_scanning_text) + + def start_transfer_animation(self) -> None: + """Start the 'Transferring...' animation.""" + self.animation_dots = 0 + self.scanning_active = False + self.animation_job = self.parent.after(0, self._animate_transfer_text) + + def stop_animation(self) -> None: + """Stop any running animation.""" + if self.animation_job is not None: + self.parent.after_cancel(self.animation_job) + self.animation_job = None + + self.scanning_active = False + self._reset_file_progress() + + def update_transfer_progress(self, current: int, total: int) -> None: + """Update the file transfer progress. + + Args: + current: Current number of files transferred + total: Total number of files to transfer + """ + self.transfer_file_progress['current'] = current + self.transfer_file_progress['total'] = total + self.transfer_file_progress['active'] = True + + def _animate_scanning_text(self) -> None: + """Animate the scanning text with dots.""" + if self.animation_job is not None and self.scanning_active: + dots = "." * (self.animation_dots + 1) + status_text = f"Scanning for duplicates{dots}" + self._update_status_label(status_text) + self.animation_dots = (self.animation_dots + 1) % 5 # Cycle 0-4 dots + # Schedule next update in 500ms + self.animation_job = self.parent.after(500, self._animate_scanning_text) + + def _animate_transfer_text(self) -> None: + """Animate the transfer text with dots.""" + if self.animation_job is not None and not self.scanning_active: + dots = "." * (self.animation_dots + 1) + + # Show file progress if available + if (self.transfer_file_progress['active'] and + self.transfer_file_progress['total'] > 0): + current = self.transfer_file_progress['current'] + total = self.transfer_file_progress['total'] + status_text = f"Transferring {current} of {total} files{dots}" + else: + status_text = f"Transferring{dots}" + + self._update_status_label(status_text) + self.animation_dots = (self.animation_dots + 1) % 5 # Cycle 0-4 dots + # Schedule next update in 500ms + self.animation_job = self.parent.after(500, self._animate_transfer_text) + + def _update_status_label(self, text: str) -> None: + """Update the status label with the given text. + + Args: + text: Text to display in the status label + """ + if hasattr(self.parent, 'status_label'): + self.parent.status_label.config(text=text) + self.parent.update_idletasks() + + def _reset_file_progress(self) -> None: + """Reset file progress tracking.""" + self.transfer_file_progress = { + 'current': 0, + 'total': 0, + 'active': False + } + + def is_animation_running(self) -> bool: + """Check if any animation is currently running. + + Returns: + True if animation is running, False otherwise + """ + return self.animation_job is not None + + def is_scanning(self) -> bool: + """Check if scanning animation is active. + + Returns: + True if scanning animation is active, False otherwise + """ + return self.scanning_active \ No newline at end of file diff --git a/src/gui/device_manager.py b/src/gui/device_manager.py new file mode 100644 index 0000000..92c5e1d --- /dev/null +++ b/src/gui/device_manager.py @@ -0,0 +1,172 @@ +""" +Device Manager Module +Handles Android device connection and ADB operations for the GUI. +""" + +import tkinter as tk +from tkinter import messagebox +from typing import Optional, Callable + +try: + from ..adb_manager import ADBManager, is_adb_available +except ImportError: + from adb_manager import ADBManager, is_adb_available + + +class DeviceManager: + """Manages Android device connections and ADB operations.""" + + def __init__(self, parent_window: tk.Tk, status_callback: Optional[Callable[[str], None]] = None): + """Initialize the device manager. + + Args: + parent_window: The main window instance + status_callback: Callback function for status updates + """ + self.parent = parent_window + self.status_callback = status_callback + self.adb_manager = ADBManager() + self.device_connected = False + + # Set up ADB callbacks + self.adb_manager.set_status_callback(self._on_adb_status_update) + self.adb_manager.set_progress_callback(self._on_adb_progress_update) + + def initialize_adb(self) -> bool: + """Initialize ADB and download tools if needed. + + Returns: + True if ADB is available and ready, False otherwise + """ + if not is_adb_available(): + # Show welcome message when platform-tools need to be downloaded + messagebox.showinfo( + "Welcome to Android File Transfer!", + "Welcome to Android File Transfer! This application simplifies " + "and speeds up file transfers over USB between computers and Android devices. " + "Please do not delete or move the platform-tools folder that will be " + "downloaded. These are tools written by Google " + "and they are required for this application to function properly." + ) + + self._update_status("ADB not found locally. Downloading...") + self.parent.update() + self.adb_manager.download_and_extract_adb() + success = is_adb_available() + + if success: + self._update_status("ADB downloaded and ready.") + return True + else: + self._update_status( + "Failed to download/access Android Debug Bridge tools. " + "Please check your internet and for any blocking security pop-ups and restart." + ) + messagebox.showerror("Error", "Failed to download ADB tools. Exiting.") + return False + + return True + + def check_device_connection(self) -> Optional[str]: + """Check for device connection and update status. + + Returns: + Device ID if connected, None otherwise + """ + self._update_status("Checking for connected device...") + self.parent.update() + + device = self.adb_manager.check_device() + if device: + self.device_connected = True + self._update_status(f"Device detected: {device}") + return device + else: + self.device_connected = False + self._update_status( + "No Android devices detected. Please check the USB connection at both ends is " + "securely inserted, USB debugging is enabled, and that File Transfer mode is turned on." + ) + return None + + def is_remote_file(self, remote_path: str) -> bool: + """Check if the remote path points to a file (not a directory). + + Args: + remote_path: Path on the Android device + + Returns: + True if it's a file, False if it's a directory or check fails + """ + try: + result = self.adb_manager.run_adb_command(["shell", "ls", "-la", remote_path]) + if isinstance(result, tuple) and len(result) == 3: + stdout, stderr, returncode = result + if returncode == 0 and stdout: + # If the output starts with '-', it's a regular file + return stdout.strip().startswith('-') + return False + except Exception: + return False + + def get_file_transfer_methods(self, direction: str, is_file: bool): + """Get the appropriate transfer methods based on direction and type. + + Args: + direction: Transfer direction ('pull' or 'push') + is_file: Whether transferring a file (True) or folder (False) + + Returns: + Tuple of (transfer_method, transfer_type) + """ + if direction == "pull": + if is_file: + return self.adb_manager.pull_file, "file" + else: + return self.adb_manager.pull_folder_with_dedup, "folder" + else: # push + if is_file: + return self.adb_manager.push_file, "file" + else: + return self.adb_manager.push_folder_with_dedup, "folder" + + def cancel_current_operation(self) -> None: + """Cancel the current ADB operation.""" + try: + self.adb_manager.cancel_current_operation() + except Exception as e: + print(f"Error cancelling operation: {e}") + + def cancel_transfer(self) -> bool: + """Cancel the current transfer operation. + + Returns: + True if transfer was cancelled successfully, False otherwise + """ + return self.adb_manager.cancel_transfer() + + def _update_status(self, message: str) -> None: + """Update status through callback if available. + + Args: + message: Status message to display + """ + if self.status_callback: + self.status_callback(message) + + def _on_adb_status_update(self, message: str) -> None: + """Handle status updates from ADB manager. + + Args: + message: Status message from ADB operations + """ + self._update_status(message) + + def _on_adb_progress_update(self, percentage: int) -> None: + """Handle progress updates from ADB manager. + + Args: + percentage: Progress percentage (0-100) + """ + # Currently we ignore progress updates and rely on status updates + pass \ No newline at end of file diff --git a/src/gui/dialog_manager.py b/src/gui/dialog_manager.py new file mode 100644 index 0000000..75a2637 --- /dev/null +++ b/src/gui/dialog_manager.py @@ -0,0 +1,430 @@ +""" +Dialog Manager Module +Handles various dialog boxes and message windows for the application. +""" + +import tkinter as tk +from tkinter import messagebox, filedialog, scrolledtext +import os +from typing import Optional, Tuple + + +class DialogManager: + """Manages various dialog boxes and user interactions.""" + + def __init__(self, parent_window: tk.Tk): + """Initialize the dialog manager. + + Args: + parent_window: The main window instance + """ + self.parent = parent_window + self.troubleshooting_steps = ( + "Android device appears to have been disconnected and/or USB debugging is disabled.\n" + "Please ensure your Android device is securely connected at both ends.\n\n" + + "To enable USB debugging:\n" + "1. Connect your device to the computer via USB\n" + "2. Open Settings → About phone\n" + "3. Tap 'Build number' seven times to unlock Developer Options\n" + " (You only need to do this once unless you disable it, reset settings, or wipe your device)\n" + "4. Navigate back and go to System → Developer Options\n" + "5. Find and enable 'USB debugging'\n" + " (Tip: Use the search icon at the top if you can't find it)\n" + "6. Connect via USB and tap 'Trust' when prompted\n" + " (Checking 'Remember' is recommended for future transfers)\n\n" + + "Ensure File Transfer mode is enabled:\n" + "1. After connecting, swipe down to view notifications\n" + "2. Look for a USB notification (often shows 'Charging over USB')\n" + "3. Tap the notification and select 'File Transfer' or 'MTP' mode\n\n" + + "Note: Menu names may vary by Android version:\n" + "• Some devices show 'Developer options' under 'System'\n" + "• Others may have it directly in the main Settings menu\n" + "• Samsung devices might show 'Software information' instead of 'About phone'\n\n" + + "If you're still having trouble:\n" + "• Try a different USB cable or port (some cables only support charging)\n" + "• Restart both your phone and computer\n" + "• Make sure your phone screen is unlocked when connecting\n" + "• Set your phone screen timeout to 30 minutes (especially for long transfers)\n\n" + "• Use a different computer to test if the issue is computer-specific\n" + + "Windows users: If you see a driver installation popup, please allow it to complete.\n" + "Linux users: You may need to run 'sudo usermod -a -G plugdev $USER' and reboot.\n" + + "After completing these steps, click 'Recheck for connected Android device' to try again." + ) + + def show_file_folder_selection_notice(self) -> bool: + """Show instructions for file and folder selection in a custom dialog. + + Returns: + True if user clicked OK, False if user cancelled or closed dialog + """ + # Create custom dialog window + dialog = tk.Toplevel(self.parent) + dialog.title("File/Folder Selection Notice") + dialog.geometry("600x250") + dialog.minsize(600, 250) + dialog.resizable(True, True) + dialog.transient(self.parent) + dialog.grab_set() + + # Center the dialog on the parent window + dialog.update_idletasks() + x = self.parent.winfo_x() + (self.parent.winfo_width() // 2) - (600 // 2) + y = self.parent.winfo_y() + (self.parent.winfo_height() // 2) - (250 // 2) + dialog.geometry(f"600x250+{x}+{y}") + + # Track if OK was clicked + dialog_confirmed = False + + # Create main frame + main_frame = tk.Frame(dialog) + main_frame.pack(fill="both", expand=True, padx=20, pady=20) + + # Selection notice text + notice_text = ( + "How file and folder selection works in this application\n\n" + "If you want to select a file for transfer, simply click on the file to select it.\n\n" + "If you want to select a folder for transfer, the folder that you navigate to " + "(the current directory you are viewing, not a highlighted folder) will be selected for transfer.\n\n" + "You can only transfer one file or one folder at a time." + ) + + # Create responsive text label + text_label = tk.Label( + main_frame, + text=notice_text, + justify="left", + anchor="nw", + wraplength=0, # Will be set dynamically + font=("Arial", 10) + ) + text_label.pack(fill="both", expand=True, pady=(0, 20)) + + # OK button with confirmation callback + def on_ok_clicked(): + nonlocal dialog_confirmed + dialog_confirmed = True + dialog.destroy() + + ok_button = tk.Button( + main_frame, + text="OK", + command=on_ok_clicked, + width=10, + font=("Arial", 10) + ) + ok_button.pack(pady=10) + + # Configure text wrapping on dialog resize + def on_dialog_configure(event): + if event.widget == dialog: + # Calculate available width for text (account for padding and margins) + available_width = dialog.winfo_width() - 60 # 20px padding * 2 + some margin + if available_width > 200: # Minimum reasonable width + text_label.config(wraplength=available_width) + + dialog.bind("", on_dialog_configure) + + # Set initial wrap length + dialog.after(10, lambda: on_dialog_configure(type('Event', (), {'widget': dialog})())) + + # Handle window close (X button) - treat as cancel + def on_dialog_close(): + nonlocal dialog_confirmed + dialog_confirmed = False + dialog.destroy() + + dialog.protocol("WM_DELETE_WINDOW", on_dialog_close) + + # Wait for dialog to close + dialog.wait_window() + + return dialog_confirmed + + def show_enable_debugging_instructions(self, callback: Optional[callable] = None) -> None: + """Show instructions to connect device, enable file transfer, and enable USB debugging. + + Args: + callback: Optional callback to execute after dialog is closed + """ + # Create custom dialog window + dialog = tk.Toplevel(self.parent) + dialog.title("Check device connection and enable USB Debugging") + dialog.geometry("700x800") # Increased height for better button spacing + dialog.minsize(700, 800) # Set minimum size to ensure all content is visible + dialog.resizable(True, True) + dialog.transient(self.parent) + dialog.grab_set() + + # Center the dialog on the parent window + dialog.update_idletasks() + x = self.parent.winfo_x() + (self.parent.winfo_width() // 2) - (700 // 2) + y = self.parent.winfo_y() + (self.parent.winfo_height() // 2) - (450 // 2) + dialog.geometry(f"700x450+{x}+{y}") + + # Create main frame + main_frame = tk.Frame(dialog) + main_frame.pack(fill="both", expand=True, padx=20, pady=20) + + # Create responsive text label + text_label = tk.Label( + main_frame, + text=self.troubleshooting_steps, + justify="left", + anchor="nw", + wraplength=0, # Will be set dynamically + font=("Arial", 10) + ) + text_label.pack(fill="both", expand=True, pady=(0, 20)) + + # OK button + ok_button = tk.Button( + main_frame, + text="OK", + command=dialog.destroy, + width=10, + font=("Arial", 10) + ) + ok_button.pack(pady=10) + + # Configure text wrapping on dialog resize + def on_dialog_configure(event): + if event.widget == dialog: + # Calculate available width for text (account for padding and margins) + available_width = dialog.winfo_width() - 60 # 20px padding * 2 + some margin + if available_width > 200: # Minimum reasonable width + text_label.config(wraplength=available_width) + + dialog.bind("", on_dialog_configure) + + # Set initial wrap length + dialog.after(10, lambda: on_dialog_configure(type('Event', (), {'widget': dialog})())) + + # Handle dialog close + def on_close(): + dialog.destroy() + if callback: + callback() + + # After user clicks OK, ensure callback is executed + dialog.protocol("WM_DELETE_WINDOW", on_close) + ok_button.config(command=on_close) + + def show_disable_debugging_reminder(self) -> None: + """Show reminder to disable USB debugging after transfer.""" + msg = ( + "Transfer completed!\n\n" + "For security, please disable USB debugging when done:\n" + "Settings → Developer Options → disable 'USB debugging'." + ) + messagebox.showinfo("Disable USB Debugging", msg) + + def show_transfer_stats(self, stats: dict, operation: str, deduplicator=None) -> None: + """Show transfer statistics dialog. + + Args: + stats: Dictionary containing transfer statistics + operation: Type of operation ("Pull" or "Push") + deduplicator: Optional deduplicator instance for byte formatting + """ + # Format bytes saved + bytes_saved_str = "" + if stats['bytes_saved'] > 0: + if deduplicator and hasattr(deduplicator, 'format_bytes'): + bytes_saved_str = f" ({deduplicator.format_bytes(stats['bytes_saved'])} saved)" + else: + bytes_saved_str = f" ({stats['bytes_saved']} bytes saved)" + + # Build message + title = f"{operation} Transfer Complete" + + if stats['total_files'] == 0: + message = "No files were found to transfer." + else: + message_parts = [ + f"Transfer completed successfully!\n", + f"Files found: {stats['total_files']}", + f"Files transferred: {stats['transferred']}", + f"Duplicate files skipped: {stats['skipped']}{bytes_saved_str}" + ] + + if stats['skipped'] > 0: + message_parts.append(f"\nDuplicate detection helped avoid unnecessary transfers!") + + message = "\n".join(message_parts) + + # Show dialog + messagebox.showinfo(title, message) + + def show_error(self, title: str, message: str) -> None: + """Show an error dialog. + + Args: + title: Dialog title + message: Error message to display + """ + messagebox.showerror(title, message) + + def show_info(self, title: str, message: str) -> None: + """Show an info dialog. + + Args: + title: Dialog title + message: Info message to display + """ + messagebox.showinfo(title, message) + + def browse_local_file_or_folder(self, direction: str, initial_dir: Optional[str] = None) -> Optional[str]: + """Browse for local file or folder based on transfer direction. + + Args: + direction: Transfer direction ('pull' or 'push') + initial_dir: Initial directory to open browser in + + Returns: + Selected path or None if cancelled + """ + if initial_dir is None: + initial_dir = os.path.expanduser("~") + + # Show helpful notification about folder selection behavior + if not self.show_file_folder_selection_notice(): + return None # User cancelled the notice dialog + + if direction == "push": + # For push, show file selection first, then folder selection if cancelled + + # First try file selection + selected_path = filedialog.askopenfilename( + title="Select file to push to Android device", + initialdir=initial_dir, + filetypes=[("All files", "*.*")] + ) + + # If no file was selected, offer folder selection as an alternative + if not selected_path: + selected_path = filedialog.askdirectory( + title="Select folder to push to Android device", + initialdir=initial_dir + ) + + return selected_path if selected_path else None + + else: # pull direction + # For pull, only allow folder selection (destination) + folder = filedialog.askdirectory( + title="Select destination folder for pulled files", + initialdir=initial_dir + ) + return folder if folder else None + + def show_file_folder_choice(self, on_file_callback, on_folder_callback): + """Show a dialog to choose between file or folder selection. + + Args: + on_file_callback: Callback for file selection + on_folder_callback: Callback for folder selection + """ + # Ask user whether they want to select a file or folder + choice_window = tk.Toplevel(self.parent) + choice_window.title("Select File or Folder") + choice_window.geometry("350x150") + choice_window.resizable(False, False) + choice_window.transient(self.parent) + choice_window.grab_set() + + # Center the window + choice_window.geometry("+{}+{}".format( + self.parent.winfo_x() + 100, + self.parent.winfo_y() + 100 + )) + + main_frame = tk.Frame(choice_window) + main_frame.pack(fill="both", expand=True, padx=20, pady=20) + + tk.Label(main_frame, text="What would you like to select?", + font=("Arial", 11)).pack(pady=(0, 15)) + + button_frame = tk.Frame(main_frame) + button_frame.pack(fill="x") + + def select_file(): + choice_window.destroy() + on_file_callback() + + def select_folder(): + choice_window.destroy() + on_folder_callback() + + def cancel_selection(): + choice_window.destroy() + + tk.Button(button_frame, text="📄 File", command=select_file, + width=12, font=("Arial", 10)).pack(side="left", padx=(0, 10)) + tk.Button(button_frame, text="📁 Folder", command=select_folder, + width=12, font=("Arial", 10)).pack(side="left", padx=(0, 10)) + tk.Button(button_frame, text="Cancel", command=cancel_selection, + width=12).pack(side="right") + + def show_transfer_stats(self, stats: dict, operation: str): + """Show transfer statistics in a dialog. + + Args: + stats: Dictionary containing transfer statistics + operation: Description of the operation performed + """ + if not stats: + return + + stats_window = tk.Toplevel(self.parent) + stats_window.title("Transfer Statistics") + stats_window.geometry("400x300") + stats_window.resizable(True, True) + stats_window.transient(self.parent) + stats_window.grab_set() + + # Center the window + stats_window.geometry("+{}+{}".format( + self.parent.winfo_x() + 60, + self.parent.winfo_y() + 60 + )) + + main_frame = tk.Frame(stats_window) + main_frame.pack(fill="both", expand=True, padx=15, pady=15) + + # Title + title_label = tk.Label(main_frame, text=f"📊 Transfer Complete - {operation.title()}", + font=("Arial", 14, "bold")) + title_label.pack(pady=(0, 15)) + + # Stats display + stats_text = tk.Text(main_frame, wrap=tk.WORD, font=("Courier", 10), + height=12, width=50) + stats_text.pack(fill="both", expand=True, pady=(0, 15)) + + # Format and insert stats + stats_content = [] + for key, value in stats.items(): + formatted_key = key.replace('_', ' ').title() + stats_content.append(f"{formatted_key}: {value}") + + stats_text.insert(tk.END, "\n".join(stats_content)) + stats_text.config(state=tk.DISABLED) + + # Close button + tk.Button(main_frame, text="Close", + command=stats_window.destroy).pack() + + def show_disable_debugging_reminder(self): + """Show reminder about disabling USB debugging after transfer.""" + messagebox.showinfo( + "Security Reminder", + "Transfer complete!\n\nFor security, consider disabling USB debugging when not needed:\n" + "Settings → Developer Options → USB debugging (toggle off)\n\n" + "Keep it enabled if you plan to transfer files again soon." + ) \ No newline at end of file diff --git a/src/gui/file_browser.py b/src/gui/file_browser.py index 2406ce6..d1e743f 100644 --- a/src/gui/file_browser.py +++ b/src/gui/file_browser.py @@ -11,11 +11,13 @@ from tkinter import messagebox, ttk class AndroidFileBrowser: """Android filesystem browser for Windows using ADB.""" - def __init__(self, parent_window, adb_manager, remote_path_var, path_changed_callback=None): + def __init__(self, parent_window, adb_manager, path_callback=None): self.parent = parent_window self.adb_manager = adb_manager - self.remote_path_var = remote_path_var - self.path_changed_callback = path_changed_callback + self.path_callback = path_callback + + # Show the browser immediately + self.show_browser() def show_browser(self, direction="pull"): """Show a browsable Android folder tree. @@ -469,25 +471,22 @@ class AndroidFileBrowser: return else: # It's a file and we're in pull mode - select the file path directly - self.remote_path_var.set(item_path) - if self.path_changed_callback: - self.path_changed_callback() + if self.path_callback: + self.path_callback(item_path) browser_window.destroy() return elif len(item_values) >= 2 and item_values[1] == "folder": # It's a folder - select the folder path directly - self.remote_path_var.set(item_path) - if self.path_changed_callback: - self.path_changed_callback() + if self.path_callback: + self.path_callback(item_path) browser_window.destroy() return # Fallback: use current path (for backwards compatibility or when no specific item is selected) current_path = current_path_var.get() if current_path and current_path.strip(): - self.remote_path_var.set(current_path) - if self.path_changed_callback: - self.path_changed_callback() + if self.path_callback: + self.path_callback(current_path) browser_window.destroy() else: selection_type = "folder" if direction == "push" else "file or folder" diff --git a/src/gui/main_window.py b/src/gui/main_window.py index 1d9ebbd..9213ed5 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -1,36 +1,42 @@ """ -Main Window Module -Core GUI application window for Android file transfers. +Main Window Module - Refactored +Core GUI application window for Android file transfers using modular components. """ import os -import sys import threading -import time import tkinter as tk -from tkinter import messagebox, filedialog, ttk, scrolledtext +from tkinter import messagebox, filedialog try: # Try relative import first (when used as module) - from ..adb_manager import ( - ADBManager, - is_adb_available, - ) + from ..adb_manager import ADBManager, is_adb_available except ImportError: # Fall back to direct import (when run directly) - from adb_manager import ( - ADBManager, - is_adb_available, - ) + from adb_manager import ADBManager, is_adb_available try: # Try relative imports first from .file_browser import AndroidFileBrowser - from .license_agreement import LicenseAgreementFrame, check_license_agreement + from .animation_handler import AnimationHandler + from .device_manager import DeviceManager + from .dialog_manager import DialogManager + from .transfer_manager import TransferManager + from .ui_components import ( + PathSelectorFrame, DirectionSelector, StatusLabel, + TransferButton, LicenseManager + ) except ImportError: # Fall back to direct imports from src.gui.file_browser import AndroidFileBrowser - from src.gui.license_agreement import LicenseAgreementFrame, check_license_agreement + from src.gui.animation_handler import AnimationHandler + from src.gui.device_manager import DeviceManager + from src.gui.dialog_manager import DialogManager + from src.gui.transfer_manager import TransferManager + from src.gui.ui_components import ( + PathSelectorFrame, DirectionSelector, StatusLabel, + TransferButton, LicenseManager + ) class AndroidFileHandlerGUI(tk.Tk): @@ -40,57 +46,26 @@ class AndroidFileHandlerGUI(tk.Tk): """Initialize the main GUI application.""" super().__init__() - # Initialize business logic - self.adb_manager = ADBManager() - # Transfer tracking for thread safety self.current_transfer_id = 0 self.device_connected = False - # License agreement tracking - self.license_agreed = check_license_agreement() - self.license_frame = None - self.main_ui_created = False - - self.troubleshooting_steps = ( - "Android device appears to have been disconnected and/or USB debugging is disabled.\n" - "Please ensure your Android device is securely connected at both ends.\n\n" - - "To enable USB debugging:\n" - "1. Connect your device to the computer via USB\n" - "2. Open Settings → About phone\n" - "3. Tap 'Build number' seven times to unlock Developer Options\n" - " (You only need to do this once unless you disable it, reset settings, or wipe your device)\n" - "4. Navigate back and go to System → Developer Options\n" - "5. Find and enable 'USB debugging'\n" - " (Tip: Use the search icon at the top if you can't find it)\n" - "6. Connect via USB and tap 'Trust' when prompted\n" - " (Checking 'Remember' is recommended for future transfers)\n\n" - - "Ensure File Transfer mode is enabled:\n" - "1. After connecting, swipe down to view notifications\n" - "2. Look for a USB notification (often shows 'Charging over USB')\n" - "3. Tap the notification and select 'File Transfer' or 'MTP' mode\n\n" - - "Note: Menu names may vary by Android version:\n" - "• Some devices show 'Developer options' under 'System'\n" - "• Others may have it directly in the main Settings menu\n" - "• Samsung devices might show 'Software information' instead of 'About phone'\n\n" - - "If you're still having trouble:\n" - "• Try a different USB cable or port (some cables only support charging)\n" - "• Restart both your phone and computer\n" - "• Make sure your phone screen is unlocked when connecting\n" - "• Set your phone screen timeout to 30 minutes (especially for long transfers)\n\n" - "• Use a different computer to test if the issue is computer-specific\n" - - "Windows users: If you see a driver installation popup, please allow it to complete.\n" - "Linux users: You may need to run 'sudo usermod -a -G plugdev $USER' and reboot.\n" - - "After completing these steps, click 'Recheck for connected Android device' to try again." + # Initialize modular components + self.license_manager = LicenseManager(self) + self.device_manager = DeviceManager(self) # Creates its own ADBManager + self.dialog_manager = DialogManager(self) + self.animation_handler = AnimationHandler(self) + self.transfer_manager = TransferManager( + self, + self.device_manager, + self.animation_handler, + self.dialog_manager ) - # Setup UI based on license status + # Get ADB manager reference from device manager + self.adb_manager = self.device_manager.adb_manager + + # Initialize UI and start application self._setup_initial_ui() def _setup_initial_ui(self): @@ -101,34 +76,13 @@ class AndroidFileHandlerGUI(tk.Tk): self.minsize(520, 320) self.resizable(True, True) - if not self.license_agreed: + if self.license_manager.needs_license_agreement(): # Show license agreement first - self._show_license_agreement() + self.license_manager.show_license_agreement(self._show_main_interface) else: # Show main interface directly self._show_main_interface() - def _show_license_agreement(self): - """Show the license agreement interface.""" - # Adjust window size for license agreement - self.geometry("700x600") - self.minsize(700, 600) - - # Create license agreement frame - self.license_frame = LicenseAgreementFrame(self, self._on_license_agreed) - - def _on_license_agreed(self): - """Handle when user agrees to license.""" - self.license_agreed = True - - # Remove license frame - if self.license_frame: - self.license_frame.destroy() - self.license_frame = None - - # Switch to main interface - self._show_main_interface() - def _show_main_interface(self): """Show the main application interface.""" # Reset window size for main interface @@ -143,89 +97,35 @@ class AndroidFileHandlerGUI(tk.Tk): def _setup_main_ui(self): """Setup the main user interface.""" # Direction selection - self.direction_var = tk.StringVar(value="pull") - direction_frame = tk.Frame(self) - direction_frame.pack(anchor="w", padx=10, pady=(10, 0)) - tk.Radiobutton( - direction_frame, - text="Pull (Android → Computer)", - variable=self.direction_var, - value="pull", - command=self._on_direction_change, - ).pack(side="left") - tk.Radiobutton( - direction_frame, - text="Push (Computer → Android)", - variable=self.direction_var, - value="push", - command=self._on_direction_change, - ).pack(side="left", padx=(20, 0)) - + self.direction_selector = DirectionSelector(self, self._on_direction_change) + # Create a container frame for the path sections that can be reordered self.path_container = tk.Frame(self) self.path_container.pack(fill="x", padx=10, pady=(10, 0)) - # Create the Android device path section - self.android_frame = tk.Frame(self.path_container) - self.android_label = tk.Label(self.android_frame, text="Android device:") - self.android_label.pack(anchor="w") + # Create path selector components + self.android_path_selector = PathSelectorFrame( + self.path_container, + "Android device:", + self.browse_remote_folder + ) - android_path_frame = tk.Frame(self.android_frame) - android_path_frame.pack(fill="x", pady=(0, 10)) - self.remote_path_var = tk.StringVar(value="Please select file or folder ->") - self.remote_path_display = tk.Label( - android_path_frame, - textvariable=self.remote_path_var, - anchor="w" + self.computer_path_selector = PathSelectorFrame( + self.path_container, + "Computer:", + self.browse_local_folder ) - self.remote_path_display.pack(side="left", fill="x", expand=True) - self.remote_browse_btn = tk.Button( - android_path_frame, text="Browse...", command=self.browse_remote_folder - ) - self.remote_browse_btn.pack(side="right", padx=(5, 0)) - - # Create the Computer path section - self.computer_frame = tk.Frame(self.path_container) - self.computer_label = tk.Label(self.computer_frame, text="Computer:") - self.computer_label.pack(anchor="w") - - computer_path_frame = tk.Frame(self.computer_frame) - computer_path_frame.pack(fill="x", pady=(0, 10)) - self.local_path_var = tk.StringVar(value="Please select file or folder ->") - self.local_path_display = tk.Label( - computer_path_frame, - textvariable=self.local_path_var, - anchor="w" - ) - self.local_path_display.pack(side="left", fill="x", expand=True) - self.local_browse_btn = tk.Button( - computer_path_frame, text="Browse...", command=self.browse_local_folder - ) - self.local_browse_btn.pack(side="right", padx=(5, 0)) # Initially arrange for pull (Android on top) self._arrange_path_sections() # Status label (responsive with word wrapping) - self.status_label = tk.Label( - self, - text="Status: Idle", - wraplength=0, # Will be set dynamically - justify="center", - anchor="center" - ) - self.status_label.pack(padx=10, fill="x", pady=(20, 5)) + self.status_label = StatusLabel(self, "Status: Idle") + + # Transfer button with multi-mode functionality + self.transfer_button = TransferButton(self) - # Bind window resize event to update label wrapping - self.bind("", self._on_window_configure) - - # Start/Recheck button (will change based on device state) - self.start_btn = tk.Button( - self, text="Start Transfer", command=self.handle_button_click, state="disabled" - ) - self.start_btn.pack(pady=10) - - # Initially disable browse buttons until device is connected + # Initially disable components until device is connected self._disable_browse_buttons() # Window close protocol @@ -233,907 +133,287 @@ class AndroidFileHandlerGUI(tk.Tk): def _initialize_components(self): """Initialize GUI components and handlers.""" - # Transfer animation state - self.transfer_animation_job = None - self.transfer_dots = 0 - - # File transfer progress tracking - self.transfer_file_progress = { - 'current': 0, - 'total': 0, - 'active': False - } - - # Set up ADB callbacks to our own methods - self.adb_manager.set_progress_callback(self._update_progress) + # Set up ADB callbacks directly + self.adb_manager.set_progress_callback(lambda x: None) # Ignore progress for now self.adb_manager.set_status_callback(self._update_status) - - # Browser component - self.browser = AndroidFileBrowser( - self, self.adb_manager, self.remote_path_var, self._validate_paths_and_update_button - ) - - def _update_progress(self, percentage): - """Handle progress updates (simplified).""" - # We ignore the percentage and just rely on status updates - pass + + # Connect transfer manager callbacks + self.transfer_manager.set_status_callback(self._update_status) + self.transfer_manager.set_ui_callback('disable_controls', self.disable_controls) + self.transfer_manager.set_ui_callback('enable_controls', self.enable_controls) + self.transfer_manager.set_ui_callback('show_error', self.report_error) + self.transfer_manager.set_ui_callback('show_stats', self._show_transfer_stats) + self.transfer_manager.set_ui_callback('show_reminder', self._show_debugging_reminder) def _update_status(self, message: str): - """Update status label (thread-safe).""" - def update_ui(): - # Check for transfer progress updates - if message.startswith("TRANSFER_PROGRESS:"): - parts = message.split(":") - if len(parts) == 3: - current = int(parts[1]) - total = int(parts[2]) - self.transfer_file_progress['current'] = current - self.transfer_file_progress['total'] = total - self.transfer_file_progress['active'] = True - # Don't update the UI here - let the animation handle it - return - - # Check if we should start or switch animations based on message content - if self.transfer_animation_job is not None: - # Animation is already running - if "Scanning for duplicates" in message and not hasattr(self, '_scanning_active'): - # Already scanning, ignore duplicate "Scanning for duplicates..." messages - return - elif "Starting transfer" in message or "Transferring" in message: - # Switch from scanning to transfer animation - if hasattr(self, '_scanning_active'): - delattr(self, '_scanning_active') - self._stop_transfer_animation() - self._start_transfer_animation() - return - elif ("Duplicate scan complete" in message or - "No duplicates found" in message or - "All files already exist" in message): - # Allow these messages to show briefly before transfer starts - self._stop_transfer_animation() - self.status_label.config(text=message) - self.update_idletasks() - return - else: - # Don't update status if transfer animation is running (except for specific cases) - return - elif "Scanning for duplicates" in message: - # Start scanning animation - self._scanning_active = True - self._start_scanning_animation() - return - - # Normal status update - self.status_label.config(text=message) - self.update_idletasks() + """Update the status label from any thread. - self.after(0, update_ui) - - def _on_window_configure(self, event): - """Handle window resize events to update label wrapping.""" - # Only handle configure events for the main window, not child widgets - if event.widget == self: - # Calculate available width for the status label - # Account for padding (10px on each side) and some margin - available_width = self.winfo_width() - 40 - if available_width > 100: # Minimum reasonable width - self.status_label.config(wraplength=available_width) + Args: + message: Status message to display + """ + try: + # Ensure UI updates happen on main thread + if threading.current_thread() == threading.main_thread(): + self.status_label.set_text(message) + self.update_idletasks() + else: + self.after(0, lambda: self._update_status(message)) + except Exception as e: + # If there's an error updating the UI, print to console + print(f"Error updating status: {e}") def _validate_paths_and_update_button(self): - """Check if both paths are selected and update button state accordingly.""" - remote_path = self.remote_path_var.get().strip() - local_path = self.local_path_var.get().strip() + """Validate selected paths and update button state accordingly.""" + android_path_valid = self.android_path_selector.is_path_selected() + computer_path_valid = self.computer_path_selector.is_path_selected() - # Check if both paths are set and not the default placeholder text - remote_selected = remote_path and remote_path != "Please select file or folder ->" - local_selected = local_path and local_path != "Please select file or folder ->" - - # Only manage state if button is in "Start Transfer" mode - if self.start_btn.cget("text") == "Start Transfer": - if remote_selected and local_selected: - self.start_btn.config(state="normal") - else: - self.start_btn.config(state="disabled") + if android_path_valid and computer_path_valid and self.device_connected: + self.transfer_button.set_transfer_mode(self.start_transfer, enabled=True) + else: + self.transfer_button.set_transfer_mode(self.start_transfer, enabled=False) def _clear_paths_and_disable_button(self): - """Clear path selections and disable the transfer button.""" - self.remote_path_var.set("Please select file or folder ->") - self.local_path_var.set("Please select file or folder ->") - # Only disable if button is in "Start Transfer" mode - if self.start_btn.cget("text") == "Start Transfer": - self.start_btn.config(state="disabled") + """Clear all path selections and disable the start button.""" + self.android_path_selector.clear_path() + self.computer_path_selector.clear_path() + self.transfer_button.disable() def _enable_browse_buttons(self): - """Enable both browse buttons when device is connected.""" - self.remote_browse_btn.config(state="normal") - self.local_browse_btn.config(state="normal") + """Enable path browse buttons when device is connected.""" + self.android_path_selector.enable_browse() + self.computer_path_selector.enable_browse() def _disable_browse_buttons(self): - """Disable both browse buttons when no device is detected.""" - self.remote_browse_btn.config(state="disabled") - self.local_browse_btn.config(state="disabled") - - def _start_transfer_animation(self): - """Start the 'Transferring...' animation.""" - self.transfer_dots = 0 - self.transfer_animation_job = True # Mark as active before starting - self._animate_transfer_text() - - def _start_scanning_animation(self): - """Start the 'Scanning for duplicates...' animation.""" - self.transfer_dots = 0 - self.transfer_animation_job = True # Mark as active before starting - self._animate_scanning_text() - - def _animate_scanning_text(self): - """Animate the scanning text with dots.""" - if self.transfer_animation_job is not None: - dots = "." * (self.transfer_dots + 1) - status_text = f"Scanning for duplicates{dots}" - self.status_label.config(text=status_text) - self.update_idletasks() # Force immediate UI update - self.transfer_dots = (self.transfer_dots + 1) % 5 # Cycle 0-4 dots - # Schedule next update in 500ms - self.transfer_animation_job = self.after(500, self._animate_scanning_text) - - def _animate_transfer_text(self): - """Animate the transfer text with dots.""" - if self.transfer_animation_job is not None: - dots = "." * (self.transfer_dots + 1) - - # Show file progress if available - if (self.transfer_file_progress['active'] and - self.transfer_file_progress['total'] > 0): - current = self.transfer_file_progress['current'] - total = self.transfer_file_progress['total'] - status_text = f"Transferring {current} of {total} files{dots}" - else: - status_text = f"Transferring{dots}" - - self.status_label.config(text=status_text) - self.update_idletasks() # Force immediate UI update - self.transfer_dots = (self.transfer_dots + 1) % 5 # Cycle 0-4 dots - # Schedule next update in 500ms - self.transfer_animation_job = self.after(500, self._animate_transfer_text) - - def _stop_transfer_animation(self): - """Stop the transfer animation.""" - if self.transfer_animation_job is not None: - if isinstance(self.transfer_animation_job, str): # It's an after job ID - self.after_cancel(self.transfer_animation_job) - self.transfer_animation_job = None - - # Clear scanning flag if it exists - if hasattr(self, '_scanning_active'): - delattr(self, '_scanning_active') - - # Reset file progress tracking - self.transfer_file_progress = { - 'current': 0, - 'total': 0, - 'active': False - } + """Disable path browse buttons when device is not connected.""" + self.android_path_selector.disable_browse() + self.computer_path_selector.disable_browse() def _initialize_app(self): - """Initialize the application - check ADB and device.""" - # Check adb availability - if not is_adb_available(): - # Show welcome message when platform-tools need to be downloaded - messagebox.showinfo( - "Welcome to Android File Transfer!", - "Welcome to Android File Transfer! This application simplifies " - "and speeds up file transfers over USB between computers and Android devices. " - "Please do not delete or move the platform-tools folder that will be " - "downloaded. These are tools written by Google " - "and they are required for this application to function properly." - ) - - self.disable_controls() - self._update_status("ADB not found locally. Downloading...") - self.update() - self.adb_manager.download_and_extract_adb() - success = is_adb_available() - if success: - self._update_status("ADB downloaded and ready.") - self.enable_controls() - else: - self._update_status( - "Failed to download/access Android Debug Bridge tools. Please check your internet and for any blocking security pop-ups and restart." - ) - messagebox.showerror("Error", "Failed to download ADB tools. Exiting.") - self.quit() - return - - # Check device connected - self.check_device_connection() - - def check_device_connection(self): - """Check for device connection and update UI accordingly.""" - self._update_status("Checking for connected device...") - self.update() - device = self.adb_manager.check_device() - if not device: - self.device_connected = False - self.disable_controls() - self._disable_browse_buttons() - self._clear_paths_and_disable_button() - self._update_status( - "No Android devices detected. Please check the USB connection at both ends is securely inserted, USB debugging is enabled, and that File Transfer mode is turned on." - ) - self._switch_to_recheck_mode() - self.show_enable_debugging_instructions() - else: - self.device_connected = True + """Initialize the application and check device connection.""" + # Initialize ADB and check device connection + self.device_manager.initialize_adb() + self.device_connected = self.device_manager.check_device_connection() + + if self.device_connected: + self._update_status("Status: Android device detected. Ready for file transfer.") self._enable_browse_buttons() - self._update_status(f"Device detected: {device}") - self._switch_to_transfer_mode() - self.enable_controls() - - def browse_remote_folder(self): - """Browse remote Android files and folders.""" - # Pass the current direction to the browser - direction = self.direction_var.get() - self.browser.show_browser(direction) - - def show_file_folder_selection_notice(self): - """Show instructions for file and folder selection in a custom dialog.""" - # Create custom dialog window - dialog = tk.Toplevel(self) - dialog.title("File/Folder Selection Notice") - dialog.geometry("600x250") - dialog.minsize(600, 250) - dialog.resizable(True, True) - dialog.transient(self) - dialog.grab_set() - - # Center the dialog on the parent window - dialog.update_idletasks() - x = self.winfo_x() + (self.winfo_width() // 2) - (600 // 2) - y = self.winfo_y() + (self.winfo_height() // 2) - (250 // 2) - dialog.geometry(f"600x250+{x}+{y}") - - # Track if OK was clicked - self.dialog_confirmed = False - - # Create main frame - main_frame = tk.Frame(dialog) - main_frame.pack(fill="both", expand=True, padx=20, pady=20) - - # Selection notice text - notice_text = ( - "This is just a quick note about how file and folder selection works in this application.\n\n" - "If you want to select a file for transfer, simply click on the file to select it.\n\n" - "If you want to select a folder for transfer, the folder that you navigate to (the current directory you are viewing, not a highlighted folder) will be selected for transfer.\n\n" - "You can only transfer one file or one folder at a time." - ) - - # Create responsive text label - text_label = tk.Label( - main_frame, - text=notice_text, - justify="left", - anchor="nw", - wraplength=0, # Will be set dynamically - font=("Arial", 10) - ) - text_label.pack(fill="both", expand=True, pady=(0, 20)) - - # OK button with confirmation callback - def on_ok_clicked(): - self.dialog_confirmed = True - dialog.destroy() - - ok_button = tk.Button( - main_frame, - text="OK", - command=on_ok_clicked, - width=10, - font=("Arial", 10) - ) - ok_button.pack(pady=10) - - # Configure text wrapping on dialog resize - def on_dialog_configure(event): - if event.widget == dialog: - # Calculate available width for text (account for padding and margins) - available_width = dialog.winfo_width() - 60 # 20px padding * 2 + some margin - if available_width > 200: # Minimum reasonable width - text_label.config(wraplength=available_width) - - dialog.bind("", on_dialog_configure) - - # Set initial wrap length - dialog.after(10, lambda: on_dialog_configure(type('Event', (), {'widget': dialog})())) - - # Handle window close (X button) - treat as cancel - def on_dialog_close(): - self.dialog_confirmed = False - dialog.destroy() - - dialog.protocol("WM_DELETE_WINDOW", on_dialog_close) - - # Wait for dialog to close - dialog.wait_window() - - def browse_local_folder(self): - """Browse for local folder or file based on direction.""" - import tkinter.filedialog as fd - import os - from tkinter import messagebox - - # Show helpful notification about folder selection behavior - self.show_file_folder_selection_notice() - - # Only proceed if user clicked OK - if not getattr(self, 'dialog_confirmed', False): - return # User closed dialog without clicking OK - - direction = self.direction_var.get() - - # Set initial directory to user's home directory - initial_dir = os.path.expanduser("~") - - if direction == "push": - # For push, show file selection first, then folder selection if cancelled - - # First try file selection - selected_path = fd.askopenfilename( - title="Select file to push to Android device", - initialdir=initial_dir, - filetypes=[("All files", "*.*")] - ) - - # If no file was selected, offer folder selection as an alternative - if not selected_path: - selected_path = fd.askdirectory( - title="Select folder to push to Android device", - initialdir=initial_dir - ) - - # Only set the path if something was actually selected - # If user cancels both dialogs, selected_path will be empty and nothing happens - if selected_path: - self.local_path_var.set(selected_path) - self._validate_paths_and_update_button() - - else: # pull direction - # For pull, only allow folder selection (destination) - folder = filedialog.askdirectory( - title="Select destination folder for pulled files", - initialdir=initial_dir - ) - if folder: - self.local_path_var.set(folder) - self._validate_paths_and_update_button() - - def disable_controls(self): - """Disable UI controls during operations.""" - # Path displays are already read-only (Labels), no need to disable them - # Don't disable start_btn here - it will be handled by button mode switching - pass - - def enable_controls(self): - """Enable UI controls after operations (thread-safe).""" - # Schedule UI update on main thread - self.after(0, self._enable_controls_ui) - - def _enable_controls_ui(self): - """Internal method to enable controls on main thread.""" - # Path displays are already read-only (Labels), no need to enable them - # Button state is handled by mode switching methods - pass + self._validate_paths_and_update_button() + else: + self._update_status("Status: No Android device detected. Please connect your device and enable USB debugging.") + self.transfer_button.set_recheck_mode(self.recheck_device) def _on_direction_change(self): - """Handle radio button direction change.""" + """Handle transfer direction change.""" self._arrange_path_sections() - + def _arrange_path_sections(self): - """Arrange the Android and Computer path sections based on direction.""" - # Remove both frames from container - self.android_frame.pack_forget() - self.computer_frame.pack_forget() + """Arrange path sections based on transfer direction.""" + # Remove both sections first + self.android_path_selector.pack_forget() + self.computer_path_selector.pack_forget() - direction = self.direction_var.get() - if direction == "pull": - # Pull: Android → Computer (Android on top) - self.android_frame.pack(fill="x", pady=(0, 5)) - self.computer_frame.pack(fill="x") - else: # push - # Push: Computer → Android (Computer on top) - self.computer_frame.pack(fill="x", pady=(0, 5)) - self.android_frame.pack(fill="x") - - def _switch_to_recheck_mode(self): - """Switch button to recheck device mode.""" - self.start_btn.config( - text="Recheck for connected Android device", - command=self.recheck_device, - state="normal", - ) - - def _switch_to_transfer_mode(self): - """Switch button to transfer mode.""" - self.start_btn.config( - text="Start Transfer", command=self.start_transfer - ) - # Only enable if both paths are selected, otherwise keep disabled - self._validate_paths_and_update_button() - - def _switch_to_cancel_mode(self): - """Switch button to cancel transfer mode.""" - self.start_btn.config( - text="Cancel Transfer", command=self.cancel_transfer, state="normal" - ) - - def handle_button_click(self): - """Handle button click - delegates to appropriate method based on device state.""" - if self.device_connected: - self.start_transfer() + if self.direction_selector.get_direction() == "pull": + # Pull: Android (source) on top, Computer (destination) on bottom + self.android_path_selector.pack(fill="x", pady=(0, 0)) + self.computer_path_selector.pack(fill="x", pady=(0, 0)) else: - self.recheck_device() + # Push: Computer (source) on top, Android (destination) on bottom + self.computer_path_selector.pack(fill="x", pady=(0, 0)) + self.android_path_selector.pack(fill="x", pady=(0, 0)) + + def browse_remote_folder(self): + """Open the Android file browser for remote path selection.""" + def on_path_selected(path): + self.android_path_selector.set_path(path) + self._validate_paths_and_update_button() + + browser = AndroidFileBrowser(self, self.adb_manager, on_path_selected) + + def browse_local_folder(self): + """Browse for local file or folder selection.""" + def on_file_selected(): + filename = filedialog.askopenfilename( + title="Select a file to transfer", + initialdir=os.path.expanduser("~") + ) + if filename: + self.computer_path_selector.set_path(filename) + self._validate_paths_and_update_button() + + def on_folder_selected(): + foldername = filedialog.askdirectory( + title="Select a folder to transfer", + initialdir=os.path.expanduser("~") + ) + if foldername: + self.computer_path_selector.set_path(foldername) + self._validate_paths_and_update_button() + + # Show file/folder selection dialog + self.dialog_manager.show_file_folder_choice(on_file_selected, on_folder_selected) def recheck_device(self): """Recheck for connected Android device.""" - # Temporarily disable the button during recheck - self.start_btn.config(state="disabled") - self.start_btn.config(text="Checking...") + self.transfer_button.set_checking_mode() + self.animation_handler.start_scanning_animation("Status: Scanning") + + def perform_recheck(): + self.device_connected = self.device_manager.check_device_connection() + self.after(0, self._handle_device_recheck_result) + + threading.Thread(target=perform_recheck, daemon=True).start() - # Use after() to allow UI to update before blocking operation - self.after(100, self._perform_device_recheck) - - def _perform_device_recheck(self): - """Perform the actual device recheck.""" - self.check_device_connection() - - def cancel_transfer(self): - """Cancel the current transfer.""" + def _handle_device_recheck_result(self): + """Handle the result of device recheck.""" + self.animation_handler.stop_animation() - # Cancel the actual ADB process - cancelled = self.adb_manager.cancel_transfer() - - # Increment transfer ID to invalidate current transfer - self.current_transfer_id += 1 - - # Stop animation and restore UI - self._stop_transfer_animation() - - # Use after() to ensure status update happens after animation stops - def update_cancelled_status(): - if cancelled: - self.status_label.config(text="Transfer cancelled by user.") - else: - self.status_label.config(text="Transfer cancellation failed.") - self.update_idletasks() - - self.after(0, update_cancelled_status) - self.enable_controls() - - # Restore proper button state if self.device_connected: - self._switch_to_transfer_mode() + self._update_status("Status: Android device detected. Ready for file transfer.") + self._enable_browse_buttons() + self._validate_paths_and_update_button() else: - self._switch_to_recheck_mode() - - def show_enable_debugging_instructions(self): - """Show instructions to connect device, enable file transfer, and to enable USB debugging.""" - # Create custom dialog window - dialog = tk.Toplevel(self) - dialog.title("Check device connection and enable USB Debugging") - dialog.geometry("700x800") # Increased height for better button spacing - dialog.minsize(700, 800) # Set minimum size to ensure all content is visible - dialog.resizable(True, True) - dialog.transient(self) - dialog.grab_set() - - # Center the dialog on the parent window - dialog.update_idletasks() - x = self.winfo_x() + (self.winfo_width() // 2) - (700 // 2) - y = self.winfo_y() + (self.winfo_height() // 2) - (450 // 2) - dialog.geometry(f"700x450+{x}+{y}") - - # Create main frame - main_frame = tk.Frame(dialog) - main_frame.pack(fill="both", expand=True, padx=20, pady=20) - - # Create responsive text label - text_label = tk.Label( - main_frame, - text=self.troubleshooting_steps, - justify="left", - anchor="nw", - wraplength=0, # Will be set dynamically - font=("Arial", 10) - ) - text_label.pack(fill="both", expand=True, pady=(0, 20)) - - # OK button - ok_button = tk.Button( - main_frame, - text="OK", - command=dialog.destroy, - width=10, - font=("Arial", 10) - ) - ok_button.pack(pady=10) - - # Configure text wrapping on dialog resize - def on_dialog_configure(event): - if event.widget == dialog: - # Calculate available width for text (account for padding and margins) - available_width = dialog.winfo_width() - 60 # 20px padding * 2 + some margin - if available_width > 200: # Minimum reasonable width - text_label.config(wraplength=available_width) - - dialog.bind("", on_dialog_configure) - - # Set initial wrap length - dialog.after(10, lambda: on_dialog_configure(type('Event', (), {'widget': dialog})())) - - # After user clicks OK, ensure recheck button is enabled - dialog.protocol("WM_DELETE_WINDOW", lambda: [dialog.destroy(), self.after(0, self._enable_recheck_after_dialog)]) - ok_button.config(command=lambda: [dialog.destroy(), self.after(0, self._enable_recheck_after_dialog)]) - - def _enable_recheck_after_dialog(self): - """Re-enable recheck button after user dismisses the dialog.""" - if not self.device_connected: - self._switch_to_recheck_mode() - - def show_disable_debugging_reminder(self): - """Show reminder to disable USB debugging after transfer (thread-safe).""" - # Schedule UI update on main thread - self.after(0, self._show_debugging_reminder_ui) - - def _show_debugging_reminder_ui(self): - """Internal method to show debugging reminder on main thread.""" - msg = ( - "Transfer completed!\n\n" - "For security, please disable USB debugging when done:\n" - "Settings → Developer Options → disable 'USB debugging'." - ) - messagebox.showinfo("Disable USB Debugging", msg) + self._update_status("Status: No Android device detected. Please connect your device and enable USB debugging.") + self.transfer_button.set_recheck_mode(self.recheck_device) def start_transfer(self): - """Start the file transfer operation.""" - # Recheck device connectivity status before starting transfer - device = self.adb_manager.check_device() - if not device: - # Device is no longer connected, update status and reset to recheck mode - self.device_connected = False - self._disable_browse_buttons() - self._clear_paths_and_disable_button() - self._update_status( - "No Android devices detected. Please check the USB connection at both ends is securely inserted, USB debugging is enabled, and that File Transfer mode is turned on." - ) - self._switch_to_recheck_mode() - self.show_enable_debugging_instructions() - return - else: - # Device is still connected, update status if needed - if not self.device_connected: - self.device_connected = True - self._enable_browse_buttons() - self._update_status(f"Device detected: {device}") - - # Double-check device is still connected before starting transfer - if not self.device_connected: - msg = ( - self.troubleshooting_steps + "\n\n" - ) - messagebox.showerror("No Android Device Detected", msg) - self._switch_to_recheck_mode() - return - - remote_path = self.remote_path_var.get().strip() - local_path = self.local_path_var.get().strip() - direction = self.direction_var.get() - - # Validate inputs - if not remote_path: - messagebox.showerror("Input Error", "Remote path cannot be empty.") - return - - # For push operations, validate local path - if direction == "push": - if not local_path or (not os.path.isfile(local_path) and not os.path.isdir(local_path)): - messagebox.showerror( - "Input Error", "Please select a valid local file or folder." - ) + """Start the file transfer process.""" + try: + # Get current values + direction = self.direction_selector.get_direction() + remote_path = self.android_path_selector.get_path() + local_path = self.computer_path_selector.get_path() + + # Validate paths + if not self.android_path_selector.is_path_selected(): + messagebox.showerror("Error", "Please select an Android device path.") return - else: # pull operations - if not local_path or not os.path.isdir(local_path): - messagebox.showerror( - "Input Error", "Please select a valid local destination folder for pulled files." - ) + + if not self.computer_path_selector.is_path_selected(): + messagebox.showerror("Error", "Please select a computer path.") return - - # Final device check before transfer - device = self.adb_manager.check_device() - if not device: - self.device_connected = False - messagebox.showerror( - "Device Disconnected", - "Android device was disconnected. Please reconnect and try again.", + + # Check device connection + if not self.device_manager.check_device_connection(): + messagebox.showerror("Error", "Android device not connected. Please check your connection and try again.") + self._handle_device_disconnection() + return + + # Switch to cancel mode and start transfer + self.transfer_button.set_cancel_mode(self.cancel_transfer) + self.animation_handler.start_transfer_animation("Status: Transferring") + + # Start transfer using transfer manager + self.transfer_manager.start_transfer( + direction, + remote_path, + local_path, + self._on_transfer_complete ) - self._switch_to_recheck_mode() - return + + except Exception as e: + self.report_error(f"Error starting transfer: {str(e)}") - # Start transfer - self.current_transfer_id += 1 - transfer_id = self.current_transfer_id - - self.disable_controls() - # Switch button to cancel mode during transfer - self._switch_to_cancel_mode() - # Start with scanning animation for folder transfers - # File transfers will skip directly to transfer animation - - # Determine if we're dealing with files or folders - if direction == "pull": - # For pull, check if remote path is a file - is_file = self._is_remote_file(remote_path) - if not is_file: - # Start scanning animation for folder transfers - self._start_scanning_animation() - else: - # Start transfer animation for file transfers - self._start_transfer_animation() - threading.Thread( - target=self._transfer_thread, - args=(direction, remote_path, local_path, transfer_id, is_file), - daemon=True, - ).start() - elif direction == "push": - # For push, check if local path is a file - is_file = os.path.isfile(local_path) - if not is_file: - # Start scanning animation for folder transfers - self._start_scanning_animation() - else: - # Start transfer animation for file transfers - self._start_transfer_animation() - threading.Thread( - target=self._transfer_thread, - args=(direction, local_path, remote_path, transfer_id, is_file), - daemon=True, - ).start() - else: - self.report_error("Invalid transfer direction selected.") + def cancel_transfer(self): + """Cancel ongoing file transfer.""" + try: + self.transfer_manager.cancel_transfer() + self.animation_handler.stop_animation() + self._update_status("Status: Transfer cancelled by user.") self.enable_controls() - - def _transfer_thread(self, direction: str, source_path: str, dest_path: str, transfer_id: int, is_file: bool): - """Unified thread function for all transfer operations.""" - try: - # Check if this transfer is still current - if self.current_transfer_id != transfer_id: - return - - # Recheck device connectivity before proceeding with transfer - device = self.adb_manager.check_device() - if not device: - # Device disconnected during transfer setup - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.after(0, self._handle_device_disconnection) - return - - # Get the appropriate transfer method - transfer_method, transfer_type = self._get_file_transfer_methods(direction, is_file) + self._validate_paths_and_update_button() - # For pull operations with files, we need to construct the full destination path - if direction == "pull" and is_file: - # Extract filename from source and append to destination directory - filename = source_path.split("/")[-1] - full_dest_path = os.path.join(dest_path, filename) - success = transfer_method(source_path, full_dest_path) - stats = None # File transfers don't return stats - else: - # For folder transfers, the method returns (success, stats) - if transfer_type == "folder": - success, stats = transfer_method(source_path, dest_path) - else: - success = transfer_method(source_path, dest_path) - stats = None - - if success and self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - transfer_desc = "File" if is_file else "Folder" - self._update_status(f"{transfer_desc} transfer completed successfully. To start another transfer, please select another file or folder.") - self.show_disable_debugging_reminder() - - # Show transfer statistics for folder transfers - if stats is not None: - self.after(0, lambda: self._show_transfer_stats(stats, direction.capitalize())) - - # Clear paths and disable button for next transfer - self.after(0, self._clear_paths_and_disable_button) except Exception as e: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - transfer_desc = "file" if is_file else "folder" - self.report_error(f"{direction.capitalize()} {transfer_desc} operation failed: {e}") - finally: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.enable_controls() - # Restore proper button state after transfer - self.after(0, self._restore_button_state) + self.report_error(f"Error cancelling transfer: {str(e)}") - def _pull_thread(self, remote_path: str, local_path: str, transfer_id: int): - """Thread function for pull operations.""" - try: - # Check if this transfer is still current - if self.current_transfer_id != transfer_id: - return - - success, stats = self.adb_manager.pull_folder_with_dedup(remote_path, local_path) - if success and self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self._update_status("Transfer completed successfully. To start another transfer, please select another file or folder.") - self.show_disable_debugging_reminder() - # Show transfer statistics - self.after(0, lambda: self._show_transfer_stats(stats, "Pull")) - # Clear paths and disable button for next transfer - self.after(0, self._clear_paths_and_disable_button) - except Exception as e: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.report_error(f"Pull operation failed: {e}") - finally: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.enable_controls() - # Restore proper button state after transfer - self.after(0, self._restore_button_state) - - def _push_thread(self, local_path: str, remote_path: str, transfer_id: int): - """Thread function for push operations.""" - try: - # Check if this transfer is still current - if self.current_transfer_id != transfer_id: - return - - success, stats = self.adb_manager.push_folder_with_dedup(local_path, remote_path) - if success and self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self._update_status("Transfer completed successfully.") - self.show_disable_debugging_reminder() - # Show transfer statistics - self.after(0, lambda: self._show_transfer_stats(stats, "Push")) - # Clear paths and disable button for next transfer - self.after(0, self._clear_paths_and_disable_button) - except Exception as e: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.report_error(f"Push operation failed: {e}") - finally: - if self.current_transfer_id == transfer_id: - self._stop_transfer_animation() - self.enable_controls() - # Restore proper button state after transfer - self.after(0, self._restore_button_state) - - def _restore_button_state(self): - """Restore the correct button state based on device connection.""" - if self.device_connected: - self._switch_to_transfer_mode() + def _on_transfer_complete(self, success: bool, stats: dict, operation: str): + """Handle transfer completion.""" + self.animation_handler.stop_animation() + + if success: + self._update_status(f"Status: Transfer complete! Successfully {operation}.") + if stats: + self._show_transfer_stats(stats, operation) + self._show_debugging_reminder() else: - self._switch_to_recheck_mode() + self._update_status("Status: Transfer failed. Please check your connection and try again.") + messagebox.showerror("Transfer Failed", "The file transfer was not successful. Please check your device connection and try again.") + + self.enable_controls() + self._validate_paths_and_update_button() def _handle_device_disconnection(self): - """Handle device disconnection during transfers.""" + """Handle when device gets disconnected.""" self.device_connected = False - self._disable_browse_buttons() self._clear_paths_and_disable_button() - self._update_status( - "No Android devices detected. Please check the USB connection at both ends is securely inserted, USB debugging is enabled, and that File Transfer mode is turned on." - ) - self._switch_to_recheck_mode() - self.enable_controls() - self.show_enable_debugging_instructions() + self._disable_browse_buttons() + self.transfer_button.set_recheck_mode(self.recheck_device) + self._update_status("Status: Device disconnected. Please reconnect and enable USB debugging.") - def _is_remote_file(self, remote_path: str) -> bool: - """Check if the remote path points to a file (not a directory).""" - try: - # Use ls -la to check if it's a file - result = self.adb_manager.run_adb_command( - ["shell", "ls", "-la", remote_path] - ) - if isinstance(result, tuple) and len(result) == 3: - stdout, stderr, returncode = result - if returncode == 0 and stdout: - # If the output starts with '-', it's a regular file - return stdout.strip().startswith('-') - return False - except Exception: - return False + def disable_controls(self): + """Disable UI controls during transfer.""" + if threading.current_thread() != threading.main_thread(): + self.after(0, self._disable_controls_ui) + else: + self._disable_controls_ui() - def _get_file_transfer_methods(self, direction: str, is_file: bool): - """Get the appropriate transfer methods based on direction and type.""" - if direction == "pull": - if is_file: - return self.adb_manager.pull_file, "file" - else: - return self.adb_manager.pull_folder_with_dedup, "folder" - else: # push - if is_file: - return self.adb_manager.push_file, "file" - else: - return self.adb_manager.push_folder_with_dedup, "folder" + def enable_controls(self): + """Enable UI controls after transfer.""" + if threading.current_thread() != threading.main_thread(): + self.after(0, self._enable_controls_ui) + else: + self._enable_controls_ui() + + def _disable_controls_ui(self): + """Disable controls on UI thread.""" + self._disable_browse_buttons() + + def _enable_controls_ui(self): + """Enable controls on UI thread.""" + self._enable_browse_buttons() def report_error(self, message: str): - """Report an error to the user (thread-safe).""" - # Schedule UI update on main thread - self.after(0, lambda: self._report_error_ui(message)) + """Report an error message to the user.""" + if threading.current_thread() != threading.main_thread(): + self.after(0, lambda: self._report_error_ui(message)) + else: + self._report_error_ui(message) def _report_error_ui(self, message: str): - """Internal method to report error on main thread.""" - self.status_label.config(text=f"Error: {message}") + """Show error message on UI thread.""" + print(f"Error: {message}") + self._update_status(f"Status: Error - {message}") messagebox.showerror("Error", message) - self.enable_controls() def _show_transfer_stats(self, stats: dict, operation: str): - """Show transfer statistics dialog. - - Args: - stats: Dictionary containing transfer statistics - operation: Type of operation ("Pull" or "Push") - """ - # Format bytes saved - bytes_saved_str = "" - if stats['bytes_saved'] > 0: - # Use the deduplicator's format_bytes method - if hasattr(self.adb_manager, 'deduplicator'): - bytes_saved_str = f" ({self.adb_manager.deduplicator.format_bytes(stats['bytes_saved'])} saved)" - else: - bytes_saved_str = f" ({stats['bytes_saved']} bytes saved)" - - # Build message - title = f"{operation} Transfer Complete" - - if stats['total_files'] == 0: - message = "No files were found to transfer." - else: - message_parts = [ - f"Transfer completed successfully!\n", - f"Files found: {stats['total_files']}", - f"Files transferred: {stats['transferred']}", - f"Duplicate files skipped: {stats['skipped']}{bytes_saved_str}" - ] - - if stats['skipped'] > 0: - message_parts.append(f"\nDuplicate detection helped avoid unnecessary transfers!") - - message = "\n".join(message_parts) - - # Show dialog - messagebox.showinfo(title, message) + """Show transfer statistics in a dialog.""" + self.dialog_manager.show_transfer_stats(stats, operation) + + def _show_debugging_reminder(self): + """Show reminder about disabling USB debugging after transfer.""" + self.after(100, lambda: self.dialog_manager.show_disable_debugging_reminder()) def on_close(self): """Handle window close event.""" - self.destroy() + try: + # Cancel any ongoing transfers + if hasattr(self, 'transfer_manager'): + self.transfer_manager.cancel_transfer() + + # Stop any animations + if hasattr(self, 'animation_handler'): + self.animation_handler.stop_animation() + + self.destroy() + except Exception as e: + print(f"Error during close: {e}") + self.destroy() def main(): - """Main function to run the application.""" - if sys.platform not in ["win32", "linux"]: - root = tk.Tk() - root.withdraw() - messagebox.showerror( - "Unsupported OS", "This application only supports Windows or Linux." - ) - sys.exit(1) - + """Main entry point for the application.""" app = AndroidFileHandlerGUI() - # If running in CI mode, auto-close after 10 seconds - if os.getenv('CI_MODE') == 'true': - def auto_close(): - time.sleep(10) - app.quit() - - threading.Thread(target=auto_close, daemon=True).start() - print("CI Mode: Application will auto-close in 10 seconds") app.mainloop() if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/src/gui/transfer_manager.py b/src/gui/transfer_manager.py new file mode 100644 index 0000000..a65982a --- /dev/null +++ b/src/gui/transfer_manager.py @@ -0,0 +1,394 @@ +""" +Transfer Manager Module +Handles file transfer operations and coordination between GUI and ADB manager. +""" + +import threading +import os +from typing import Optional, Callable, Tuple, Dict, Any + +try: + from .device_manager import DeviceManager + from .animation_handler import AnimationHandler + from .dialog_manager import DialogManager +except ImportError: + from device_manager import DeviceManager + from animation_handler import AnimationHandler + from dialog_manager import DialogManager + + +class TransferManager: + """Manages file transfer operations and coordination.""" + + def __init__(self, parent_window, device_manager: DeviceManager, + animation_handler: AnimationHandler, dialog_manager: DialogManager): + """Initialize the transfer manager. + + Args: + parent_window: The main window instance + device_manager: Device manager instance + animation_handler: Animation handler instance + dialog_manager: Dialog manager instance + """ + self.parent = parent_window + self.device_manager = device_manager + self.animation_handler = animation_handler + self.dialog_manager = dialog_manager + + # Transfer tracking + self.current_transfer_id = 0 + + # UI callbacks + self.ui_callbacks = {} + + def set_ui_callback(self, name: str, callback: Callable) -> None: + """Set a UI callback function. + + Args: + name: Name of the callback + callback: Function to call + """ + self.ui_callbacks[name] = callback + + def set_status_callback(self, callback: Callable[[str], None]) -> None: + """Set the status update callback. + + Args: + callback: Function to call for status updates + """ + self.ui_callbacks['status'] = callback + self.status_callback = callback + + def set_controls_callback(self, callback: Callable[[], None]) -> None: + """Set the controls update callback. + + Args: + callback: Function to call for enabling/disabling controls + """ + self.controls_callback = callback + + def start_transfer(self, direction: str, source_path: str, dest_path: str, + completion_callback: Optional[Callable] = None) -> bool: + """Start a file transfer operation. + + Args: + direction: Transfer direction ('pull' or 'push') + source_path: Source file or folder path + dest_path: Destination path + completion_callback: Callback to call when transfer completes + + Returns: + True if transfer was started successfully, False otherwise + """ + # Increment transfer ID for cancellation support + self.current_transfer_id += 1 + transfer_id = self.current_transfer_id + + # Determine if transferring a file or folder + if direction == "pull": + is_file = self._is_remote_file(source_path) + else: + is_file = os.path.isfile(source_path) + + # Disable controls during transfer + if 'disable_controls' in self.ui_callbacks: + self.ui_callbacks['disable_controls']() + + # Start transfer in background thread + transfer_thread = threading.Thread( + target=self._transfer_thread, + args=(direction, source_path, dest_path, transfer_id, is_file, completion_callback), + daemon=True + ) + transfer_thread.start() + return True + + def cancel_transfer(self) -> None: + """Cancel the current transfer operation.""" + # Increment transfer ID to invalidate current transfer + self.current_transfer_id += 1 + + # Cancel ADB operation + self.device_manager.cancel_current_operation() + + def _is_remote_file(self, remote_path: str) -> bool: + """Check if a remote path is a file. + + Args: + remote_path: Path on Android device + + Returns: + True if path is a file, False if it's a folder + """ + try: + # Use device manager to check if path is a file + return self.device_manager.is_remote_file(remote_path) + except Exception: + # If we can't determine, assume it's a folder for safety + return False + + def _transfer_thread(self, direction: str, source_path: str, dest_path: str, + transfer_id: int, is_file: bool, completion_callback: Optional[Callable]): + """Handle file transfer in background thread. + + Args: + direction: Transfer direction ('pull' or 'push') + source_path: Source file or folder path + dest_path: Destination path + transfer_id: Transfer ID for cancellation + is_file: True if transferring a file + completion_callback: Callback for completion + """ + try: + # Check if transfer is still valid + if transfer_id != self.current_transfer_id: + return + + # Get ADB manager from device manager + adb_manager = self.device_manager.adb_manager + + # Perform the transfer + if direction == "pull": + if is_file: + success, stats = adb_manager.pull_file(source_path, dest_path) + else: + success, stats = adb_manager.pull_folder(source_path, dest_path) + operation = "pulled from Android device" + else: # push + if is_file: + success, stats = adb_manager.push_file(source_path, dest_path) + else: + success, stats = adb_manager.push_folder(source_path, dest_path) + operation = "pushed to Android device" + + # Check if transfer was cancelled + if transfer_id != self.current_transfer_id: + return + + # Call completion callback on main thread + if completion_callback: + self.parent.after(0, lambda: completion_callback(success, stats, operation)) + + except Exception as e: + # Handle errors on main thread + if transfer_id == self.current_transfer_id and 'show_error' in self.ui_callbacks: + self.parent.after(0, lambda: self.ui_callbacks["show_error"](f"Transfer error: {str(e)}")) + + # Start transfer in background thread + self.current_transfer_id += 1 + transfer_id = self.current_transfer_id + + threading.Thread( + target=self._transfer_thread, + args=(direction, source_path, dest_path, transfer_id, is_file), + daemon=True + ).start() + + return True + + def cancel_transfer(self) -> bool: + """Cancel the current transfer operation. + + Returns: + True if transfer was cancelled successfully, False otherwise + """ + # Cancel the actual ADB process + cancelled = self.device_manager.cancel_transfer() + + # Increment transfer ID to invalidate current transfer + self.current_transfer_id += 1 + + # Stop animation and restore UI + self.animation_handler.stop_animation() + + # Update status + status = "Transfer cancelled by user." if cancelled else "Transfer cancellation failed." + self._update_status(status) + + return cancelled + + def _validate_transfer_paths(self, direction: str, source_path: str, dest_path: str) -> bool: + """Validate transfer paths based on direction. + + Args: + direction: Transfer direction ('pull' or 'push') + source_path: Source path + dest_path: Destination path + + Returns: + True if paths are valid, False otherwise + """ + if not source_path or not dest_path: + self.dialog_manager.show_error("Input Error", "Both source and destination paths are required.") + return False + + if direction == "push": + # For push operations, validate local source path exists + if not os.path.exists(source_path): + self.dialog_manager.show_error( + "Input Error", + f"Source path does not exist: {source_path}" + ) + return False + else: # pull operations + # For pull operations, validate local destination is a directory + if not os.path.isdir(dest_path): + self.dialog_manager.show_error( + "Input Error", + "Destination must be a valid directory for pulled files." + ) + return False + + return True + + def _transfer_thread(self, direction: str, source_path: str, dest_path: str, + transfer_id: int, is_file: bool) -> None: + """Background thread function for transfer operations. + + Args: + direction: Transfer direction ('pull' or 'push') + source_path: Source path + dest_path: Destination path + transfer_id: Transfer ID for thread safety + is_file: True if transferring a file, False for folder + """ + try: + # Check if this transfer is still current + if self.current_transfer_id != transfer_id: + return + + # Recheck device connectivity before proceeding + device = self.device_manager.adb_manager.check_device() + if not device: + if self.current_transfer_id == transfer_id: + self.animation_handler.stop_animation() + self.parent.after(0, self._handle_device_disconnection) + return + + # Get the appropriate transfer method + transfer_method, transfer_type = self.device_manager.get_file_transfer_methods(direction, is_file) + + # Perform the transfer + success, stats = self._execute_transfer( + direction, source_path, dest_path, transfer_method, transfer_type, is_file + ) + + if success and self.current_transfer_id == transfer_id: + self.animation_handler.stop_animation() + transfer_desc = "File" if is_file else "Folder" + self._update_status( + f"{transfer_desc} transfer completed successfully. " + "To start another transfer, please select another file or folder." + ) + + # Show debugging reminder and transfer statistics + self.parent.after(0, self.dialog_manager.show_disable_debugging_reminder) + + if stats is not None: + self.parent.after(0, lambda: self.dialog_manager.show_transfer_stats( + stats, direction.capitalize(), self.device_manager.adb_manager.deduplicator + )) + + except Exception as e: + if self.current_transfer_id == transfer_id: + self.animation_handler.stop_animation() + transfer_desc = "file" if is_file else "folder" + error_msg = f"{direction.capitalize()} {transfer_desc} operation failed: {e}" + self.parent.after(0, lambda: self.dialog_manager.show_error("Transfer Error", error_msg)) + finally: + if self.current_transfer_id == transfer_id: + self.animation_handler.stop_animation() + if self.controls_callback: + self.parent.after(0, self.controls_callback) + + def _execute_transfer(self, direction: str, source_path: str, dest_path: str, + transfer_method: Callable, transfer_type: str, is_file: bool) -> Tuple[bool, Optional[Dict[str, Any]]]: + """Execute the actual transfer operation. + + Args: + direction: Transfer direction + source_path: Source path + dest_path: Destination path + transfer_method: Transfer method to call + transfer_type: Type of transfer ('file' or 'folder') + is_file: True if transferring a file + + Returns: + Tuple of (success, stats_dict or None) + """ + if direction == "pull" and is_file: + # For pull operations with files, construct the full destination path + filename = source_path.split("/")[-1] + full_dest_path = os.path.join(dest_path, filename) + success = transfer_method(source_path, full_dest_path) + return success, None + else: + # For folder transfers or push operations + if transfer_type == "folder": + # Folder transfers return (success, stats) + return transfer_method(source_path, dest_path) + else: + # File transfers return just success + success = transfer_method(source_path, dest_path) + return success, None + + def _handle_device_disconnection(self) -> None: + """Handle device disconnection during transfers.""" + self.device_manager.device_connected = False + self._update_status( + "No Android devices detected. Please check the USB connection at both ends is " + "securely inserted, USB debugging is enabled, and that File Transfer mode is turned on." + ) + self.dialog_manager.show_enable_debugging_instructions() + + def _update_status(self, message: str) -> None: + """Update status through callback if available. + + Args: + message: Status message to display + """ + if self.status_callback: + self.status_callback(message) + + def handle_adb_status_update(self, message: str) -> None: + """Handle status updates from ADB operations. + + Args: + message: Status message from ADB + """ + # Check for transfer progress updates + if message.startswith("TRANSFER_PROGRESS:"): + parts = message.split(":") + if len(parts) == 3: + current = int(parts[1]) + total = int(parts[2]) + self.animation_handler.update_transfer_progress(current, total) + return + + # Check for animation state changes + if self.animation_handler.is_animation_running(): + if "Scanning for duplicates" in message and not self.animation_handler.is_scanning(): + # Already scanning, ignore duplicate messages + return + elif "Starting transfer" in message or "Transferring" in message: + # Switch from scanning to transfer animation + self.animation_handler.stop_animation() + self.animation_handler.start_transfer_animation() + return + elif ("Duplicate scan complete" in message or + "No duplicates found" in message or + "All files already exist" in message): + # Allow these messages to show briefly before transfer starts + self.animation_handler.stop_animation() + self._update_status(message) + return + else: + # Don't update status if animation is running (except for specific cases) + return + elif "Scanning for duplicates" in message: + # Start scanning animation + self.animation_handler.start_scanning_animation() + return + + # Normal status update + self._update_status(message) \ No newline at end of file diff --git a/src/gui/ui_components.py b/src/gui/ui_components.py new file mode 100644 index 0000000..600354a --- /dev/null +++ b/src/gui/ui_components.py @@ -0,0 +1,319 @@ +""" +UI Components Module +Contains reusable UI components for the Android file transfer application. +""" + +import tkinter as tk +from tkinter import messagebox +from typing import Callable, Optional + +try: + from .license_agreement import LicenseAgreementFrame, check_license_agreement +except ImportError: + from license_agreement import LicenseAgreementFrame, check_license_agreement + + +class PathSelectorFrame: + """Frame component for path selection with browse button.""" + + def __init__(self, parent: tk.Widget, label_text: str, browse_command: Callable): + """Initialize the path selector frame. + + Args: + parent: Parent widget + label_text: Text for the label + browse_command: Command to execute when browse button is clicked + """ + self.frame = tk.Frame(parent) + self.label = tk.Label(self.frame, text=label_text) + self.label.pack(anchor="w") + + # Path frame with label and button + path_frame = tk.Frame(self.frame) + path_frame.pack(fill="x", pady=(0, 10)) + + self.path_var = tk.StringVar(value="Please select file or folder ->") + self.path_display = tk.Label( + path_frame, + textvariable=self.path_var, + anchor="w" + ) + self.path_display.pack(side="left", fill="x", expand=True) + + self.browse_btn = tk.Button( + path_frame, + text="Browse...", + command=browse_command + ) + self.browse_btn.pack(side="right", padx=(5, 0)) + + def pack(self, **kwargs): + """Pack the frame.""" + self.frame.pack(**kwargs) + + def pack_forget(self): + """Remove the frame from packing.""" + self.frame.pack_forget() + + def set_path(self, path: str) -> None: + """Set the displayed path. + + Args: + path: Path to display + """ + self.path_var.set(path) + + def get_path(self) -> str: + """Get the current path. + + Returns: + Current path string + """ + return self.path_var.get().strip() + + def is_path_selected(self) -> bool: + """Check if a valid path is selected. + + Returns: + True if path is selected and not the default placeholder + """ + path = self.get_path() + return path and path != "Please select file or folder ->" + + def clear_path(self) -> None: + """Clear the path selection.""" + self.path_var.set("Please select file or folder ->") + + def enable_browse(self) -> None: + """Enable the browse button.""" + self.browse_btn.config(state="normal") + + def disable_browse(self) -> None: + """Disable the browse button.""" + self.browse_btn.config(state="disabled") + + +class DirectionSelector: + """Component for selecting transfer direction (pull/push).""" + + def __init__(self, parent: tk.Widget, on_change_command: Optional[Callable] = None): + """Initialize the direction selector. + + Args: + parent: Parent widget + on_change_command: Command to execute when direction changes + """ + self.direction_var = tk.StringVar(value="pull") + + self.frame = tk.Frame(parent) + self.frame.pack(anchor="w", padx=10, pady=(10, 0)) + + tk.Radiobutton( + self.frame, + text="Pull (Android → Computer)", + variable=self.direction_var, + value="pull", + command=on_change_command, + ).pack(side="left") + + tk.Radiobutton( + self.frame, + text="Push (Computer → Android)", + variable=self.direction_var, + value="push", + command=on_change_command, + ).pack(side="left", padx=(20, 0)) + + def get_direction(self) -> str: + """Get the current direction. + + Returns: + Current direction ('pull' or 'push') + """ + return self.direction_var.get() + + +class StatusLabel: + """Responsive status label with word wrapping.""" + + def __init__(self, parent: tk.Widget, initial_text: str = "Status: Idle"): + """Initialize the status label. + + Args: + parent: Parent widget + initial_text: Initial status text + """ + self.label = tk.Label( + parent, + text=initial_text, + wraplength=0, # Will be set dynamically + justify="center", + anchor="center" + ) + self.label.pack(padx=10, fill="x", pady=(20, 5)) + + # Bind parent window resize to update wrapping + parent.bind("", self._on_window_configure) + + def set_text(self, text: str) -> None: + """Set the status text. + + Args: + text: Text to display + """ + self.label.config(text=text) + + def get_text(self) -> str: + """Get the current status text. + + Returns: + Current status text + """ + return self.label.cget("text") + + def _on_window_configure(self, event) -> None: + """Handle window resize events to update label wrapping. + + Args: + event: Configure event + """ + # Only handle configure events for the main window, not child widgets + if hasattr(event.widget, 'winfo_toplevel') and event.widget == event.widget.winfo_toplevel(): + # Calculate available width for the status label + # Account for padding (10px on each side) and some margin + available_width = event.widget.winfo_width() - 40 + if available_width > 100: # Minimum reasonable width + self.label.config(wraplength=available_width) + + +class TransferButton: + """Multi-mode transfer button that changes text and behavior based on state.""" + + def __init__(self, parent: tk.Widget): + """Initialize the transfer button. + + Args: + parent: Parent widget + """ + self.button = tk.Button( + parent, + text="Start Transfer", + state="disabled" + ) + self.button.pack(pady=10) + + self.current_mode = "transfer" # transfer, recheck, cancel + + def set_transfer_mode(self, command: Callable, enabled: bool = True) -> None: + """Set button to transfer mode. + + Args: + command: Command to execute on button click + enabled: Whether button should be enabled + """ + self.current_mode = "transfer" + self.button.config( + text="Start Transfer", + command=command, + state="normal" if enabled else "disabled" + ) + + def set_recheck_mode(self, command: Callable) -> None: + """Set button to recheck device mode. + + Args: + command: Command to execute on button click + """ + self.current_mode = "recheck" + self.button.config( + text="Recheck for connected Android device", + command=command, + state="normal" + ) + + def set_cancel_mode(self, command: Callable) -> None: + """Set button to cancel transfer mode. + + Args: + command: Command to execute on button click + """ + self.current_mode = "cancel" + self.button.config( + text="Cancel Transfer", + command=command, + state="normal" + ) + + def set_checking_mode(self) -> None: + """Set button to temporary checking state.""" + self.button.config( + text="Checking...", + state="disabled" + ) + + def enable(self) -> None: + """Enable the button.""" + self.button.config(state="normal") + + def disable(self) -> None: + """Disable the button.""" + self.button.config(state="disabled") + + def get_mode(self) -> str: + """Get the current button mode. + + Returns: + Current mode string ('transfer', 'recheck', 'cancel') + """ + return self.current_mode + + +class LicenseManager: + """Manages license agreement display and main interface switching.""" + + def __init__(self, parent_window: tk.Tk): + """Initialize the license manager. + + Args: + parent_window: The main window instance + """ + self.parent = parent_window + self.license_agreed = check_license_agreement() + self.license_frame = None + self.on_agreed_callback: Optional[Callable] = None + + def needs_license_agreement(self) -> bool: + """Check if license agreement is needed. + + Returns: + True if license agreement needs to be shown, False otherwise + """ + return not self.license_agreed + + def show_license_agreement(self, on_agreed_callback: Callable) -> None: + """Show the license agreement interface. + + Args: + on_agreed_callback: Callback to execute when license is agreed + """ + self.on_agreed_callback = on_agreed_callback + + # Adjust window size for license agreement + self.parent.geometry("700x600") + self.parent.minsize(700, 600) + + # Create license agreement frame + self.license_frame = LicenseAgreementFrame(self.parent, self._on_license_agreed) + + def _on_license_agreed(self) -> None: + """Handle when user agrees to license.""" + self.license_agreed = True + + # Remove license frame + if self.license_frame: + self.license_frame.destroy() + self.license_frame = None + + # Execute callback + if self.on_agreed_callback: + self.on_agreed_callback() \ No newline at end of file