Rewrite device handling: udisksctl mount/unmount, Nano 7 database integration
- Replace libimobiledevice dependency with udisksctl (works without root) - Detect block devices (/dev/sdX) for iPod partitions - Add mount_device() using udisksctl mount -b - Add unmount_device() using udisksctl unmount + power-off - Update transfer_file() to use Nano7Database.add_track() - Handle duplicate skips (add_track returns None) - Fall back to simple file copy if Nano 7 DB unavailable
This commit is contained in:
Maksim Totmin
parent
11c40c8a99
commit
762e9dffe8
@ -1,11 +1,12 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
iPod Device Handler Module for iPod Nano Transfer Tool
|
||||
Handles device detection and file transfer to iPod Nano
|
||||
Handles device detection, mounting, and file transfer to iPod Nano
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import logging
|
||||
import platform
|
||||
import shutil
|
||||
@ -31,37 +32,9 @@ class IPodDevice:
|
||||
self.mount_point = mount_point
|
||||
self.system = platform.system()
|
||||
self.device_info = {}
|
||||
self.logger = logger # Add this line
|
||||
|
||||
# Check if libimobiledevice is installed
|
||||
if not self._check_dependencies():
|
||||
self.logger.warning("Some dependencies are missing. Device detection may not work properly.")
|
||||
|
||||
def _check_dependencies(self) -> bool:
|
||||
"""
|
||||
Check if required dependencies are installed
|
||||
|
||||
Returns:
|
||||
True if all dependencies are installed, False otherwise
|
||||
"""
|
||||
try:
|
||||
if self.system == "Linux":
|
||||
# Check for ifuse and libimobiledevice
|
||||
subprocess.run(["which", "ifuse"], check=True, capture_output=True)
|
||||
subprocess.run(["which", "idevice_id"], check=True, capture_output=True)
|
||||
return True
|
||||
elif self.system == "Darwin": # macOS
|
||||
# Check for libimobiledevice
|
||||
subprocess.run(["which", "idevice_id"], check=True, capture_output=True)
|
||||
return True
|
||||
elif self.system == "Windows":
|
||||
# Check for libimobiledevice-win32 (simplified check)
|
||||
# In reality, this would need a more complex check
|
||||
return os.path.exists("C:\\Program Files\\libimobiledevice")
|
||||
|
||||
return False
|
||||
except subprocess.SubprocessError:
|
||||
return False
|
||||
self._block_device = None # /dev/sdd1 etc.
|
||||
self._parent_drive = None # /dev/sdd (for power-off)
|
||||
self.logger = logger
|
||||
|
||||
def detect_devices(self) -> List[Dict[str, str]]:
|
||||
"""
|
||||
@ -71,234 +44,223 @@ class IPodDevice:
|
||||
List of dictionaries with device information
|
||||
"""
|
||||
devices = []
|
||||
|
||||
# Check platform
|
||||
system = platform.system()
|
||||
|
||||
try:
|
||||
if system == "Darwin": # macOS
|
||||
# Try using libimobiledevice if available
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["idevice_id", "-l"],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
# Parse device IDs
|
||||
device_ids = [line.strip() for line in result.stdout.splitlines() if line.strip()]
|
||||
|
||||
for device_id in device_ids:
|
||||
# Get device info
|
||||
info_result = subprocess.run(
|
||||
["ideviceinfo", "-u", device_id],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True
|
||||
)
|
||||
|
||||
if info_result.returncode == 0:
|
||||
# Parse device info
|
||||
device_info = {}
|
||||
for line in info_result.stdout.splitlines():
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
device_info[key.strip()] = value.strip()
|
||||
|
||||
if "ProductType" in device_info and "iPod" in device_info["ProductType"]:
|
||||
devices.append({
|
||||
"id": device_id,
|
||||
"name": device_info.get("DeviceName", "iPod"),
|
||||
"model": device_info.get("ProductType", "Unknown"),
|
||||
"mount_point": self._get_ipod_mount_point(device_id)
|
||||
})
|
||||
except FileNotFoundError:
|
||||
# libimobiledevice not installed, fall back to disk utility
|
||||
self.logger.warning("libimobiledevice not found, falling back to disk utility")
|
||||
|
||||
# Fall back to diskutil list
|
||||
result = subprocess.run(
|
||||
["diskutil", "list"],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
# Look for iPod in the output
|
||||
for line in result.stdout.splitlines():
|
||||
if "iPod" in line:
|
||||
# Extract disk identifier
|
||||
parts = line.split()
|
||||
for i, part in enumerate(parts):
|
||||
if part.startswith("/dev/"):
|
||||
disk_id = part
|
||||
|
||||
# Get mount point
|
||||
mount_result = subprocess.run(
|
||||
["diskutil", "info", disk_id],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True
|
||||
)
|
||||
|
||||
if mount_result.returncode == 0:
|
||||
mount_point = None
|
||||
for info_line in mount_result.stdout.splitlines():
|
||||
if "Mount Point" in info_line:
|
||||
mount_point = info_line.split(":", 1)[1].strip()
|
||||
break
|
||||
|
||||
if mount_point and os.path.exists(mount_point):
|
||||
devices.append({
|
||||
"id": disk_id,
|
||||
"name": "iPod",
|
||||
"model": "Unknown",
|
||||
"mount_point": mount_point
|
||||
})
|
||||
|
||||
# If no devices found, check for mounted volumes with iPod in the name
|
||||
if not devices:
|
||||
volumes_dir = "/Volumes"
|
||||
if os.path.exists(volumes_dir):
|
||||
for volume in os.listdir(volumes_dir):
|
||||
volume_path = os.path.join(volumes_dir, volume)
|
||||
if "iPod" in volume and os.path.isdir(volume_path):
|
||||
devices.append({
|
||||
"id": "unknown",
|
||||
"name": volume,
|
||||
"model": "Unknown",
|
||||
"mount_point": volume_path
|
||||
})
|
||||
|
||||
if system == "Linux":
|
||||
devices = self._detect_linux()
|
||||
elif system == "Darwin":
|
||||
devices = self._detect_macos()
|
||||
elif system == "Windows":
|
||||
# On Windows, check for mounted drives with iPod in the label
|
||||
import win32api
|
||||
import win32con
|
||||
|
||||
try:
|
||||
drives = win32api.GetLogicalDriveStrings().split('\0')[:-1]
|
||||
for drive in drives:
|
||||
try:
|
||||
volume_name = win32api.GetVolumeInformation(drive)[0]
|
||||
if volume_name and "iPod" in volume_name:
|
||||
devices.append({
|
||||
"id": drive[0],
|
||||
"name": volume_name,
|
||||
"model": "Unknown",
|
||||
"mount_point": drive
|
||||
})
|
||||
except:
|
||||
pass
|
||||
except ImportError:
|
||||
self.logger.warning("win32api module not available, cannot detect iPod on Windows")
|
||||
|
||||
# Fallback to checking drive letters
|
||||
import string
|
||||
for letter in string.ascii_uppercase:
|
||||
drive = f"{letter}:\\"
|
||||
if os.path.exists(drive):
|
||||
# Check if this looks like an iPod
|
||||
if os.path.exists(os.path.join(drive, "iPod_Control")):
|
||||
devices.append({
|
||||
"id": letter,
|
||||
"name": "iPod",
|
||||
"model": "Unknown",
|
||||
"mount_point": drive
|
||||
})
|
||||
|
||||
elif system == "Linux":
|
||||
# On Linux, check mounted devices
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["lsblk", "-o", "NAME,LABEL,MOUNTPOINT", "-J"],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
import json
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
for device in data.get("blockdevices", []):
|
||||
if "children" in device:
|
||||
for child in device["children"]:
|
||||
label = child.get("label", "")
|
||||
mountpoint = child.get("mountpoint")
|
||||
if mountpoint and label and "ipod" in label.lower():
|
||||
devices.append({
|
||||
"id": child.get("name", "unknown"),
|
||||
"name": label,
|
||||
"model": "Unknown",
|
||||
"mount_point": mountpoint
|
||||
})
|
||||
except json.JSONDecodeError:
|
||||
self.logger.warning("Failed to parse lsblk output as JSON")
|
||||
except FileNotFoundError:
|
||||
self.logger.warning("lsblk command not found")
|
||||
|
||||
# Fallback to checking /media and /mnt directories
|
||||
if not devices:
|
||||
for mount_dir in ["/media", "/mnt", "/run/media"]:
|
||||
if os.path.exists(mount_dir):
|
||||
for user_dir in os.listdir(mount_dir):
|
||||
user_path = os.path.join(mount_dir, user_dir)
|
||||
if os.path.isdir(user_path):
|
||||
for device_dir in os.listdir(user_path):
|
||||
if "ipod" in device_dir.lower():
|
||||
device_path = os.path.join(user_path, device_dir)
|
||||
if os.path.isdir(device_path):
|
||||
devices.append({
|
||||
"id": "unknown",
|
||||
"name": device_dir,
|
||||
"model": "Unknown",
|
||||
"mount_point": device_path
|
||||
})
|
||||
devices = self._detect_windows()
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error detecting devices: {e}")
|
||||
|
||||
# If no devices found through system methods, check for manual mount point
|
||||
# If no devices found, check for manual mount point
|
||||
if not devices and self.mount_point and os.path.exists(self.mount_point):
|
||||
devices.append({
|
||||
"id": "manual",
|
||||
"name": os.path.basename(self.mount_point),
|
||||
"model": "Unknown",
|
||||
"mount_point": self.mount_point
|
||||
"mount_point": self.mount_point,
|
||||
"mounted": True,
|
||||
})
|
||||
|
||||
return devices
|
||||
|
||||
def _get_ipod_mount_point(self, device_id: str) -> Optional[str]:
|
||||
"""
|
||||
Get the mount point for an iPod device
|
||||
def _detect_linux(self) -> List[Dict[str, str]]:
|
||||
"""Detect iPod on Linux using lsblk and udisksctl"""
|
||||
devices = []
|
||||
|
||||
Args:
|
||||
device_id: Device ID
|
||||
# Method 1: lsblk JSON output - find devices with IPOD label
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["lsblk", "-o", "NAME,LABEL,MOUNTPOINT,TYPE", "-J"],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
Returns:
|
||||
Mount point if found, None otherwise
|
||||
"""
|
||||
# This method is not implemented in the provided code
|
||||
# You may need to implement it according to your requirements
|
||||
pass
|
||||
if result.returncode == 0:
|
||||
data = json.loads(result.stdout)
|
||||
for device in data.get("blockdevices", []):
|
||||
if device.get("type") != "disk":
|
||||
continue
|
||||
parent_name = device.get("name", "")
|
||||
for child in device.get("children", []):
|
||||
label = child.get("label", "")
|
||||
mountpoint = child.get("mountpoint")
|
||||
child_name = child.get("name", "")
|
||||
|
||||
if label and "ipod" in label.lower():
|
||||
dev_path = f"/dev/{child_name}"
|
||||
parent_dev = f"/dev/{parent_name}"
|
||||
|
||||
device_entry = {
|
||||
"id": dev_path,
|
||||
"name": label,
|
||||
"model": "iPod Nano 7",
|
||||
"mount_point": mountpoint,
|
||||
"mounted": mountpoint is not None,
|
||||
"_parent_drive": parent_dev,
|
||||
}
|
||||
devices.append(device_entry)
|
||||
except (FileNotFoundError, json.JSONDecodeError) as e:
|
||||
self.logger.warning(f"lsblk JSON failed: {e}")
|
||||
|
||||
# Method 2: udisksctl dump - find iPod drives
|
||||
if not devices:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["udisksctl", "dump"],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
current_block = None
|
||||
current_drive = None
|
||||
current_label = None
|
||||
current_mount = None
|
||||
|
||||
for line in result.stdout.splitlines():
|
||||
if "/org/freedesktop/UDisks2/block_devices/" in line:
|
||||
# Save previous if it was an iPod
|
||||
if current_block and current_label and "ipod" in current_label.lower():
|
||||
devices.append({
|
||||
"id": current_block,
|
||||
"name": current_label,
|
||||
"model": "iPod",
|
||||
"mount_point": current_mount if current_mount else None,
|
||||
"mounted": current_mount is not None,
|
||||
"_parent_drive": self._get_parent_drive_from_udisks(current_block),
|
||||
})
|
||||
current_block = line.split("/block_devices/")[-1].strip().rstrip(":")
|
||||
current_drive = None
|
||||
current_label = None
|
||||
current_mount = None
|
||||
|
||||
if "IdLabel:" in line and current_block:
|
||||
val = line.split(":", 1)[1].strip()
|
||||
if val and val != "''":
|
||||
current_label = val.strip("'")
|
||||
if "MountPoints:" in line and current_block:
|
||||
val = line.split(":", 1)[1].strip()
|
||||
if val and val != "''" and val != "b''":
|
||||
# MountPoints is a byte array like b'/run/media/mat/IPOD'
|
||||
import ast
|
||||
try:
|
||||
decoded = ast.literal_eval(val)
|
||||
if isinstance(decoded, bytes):
|
||||
current_mount = decoded.decode("utf-8", errors="replace")
|
||||
elif isinstance(decoded, str):
|
||||
current_mount = decoded
|
||||
except (ValueError, SyntaxError):
|
||||
current_mount = val.strip("'\"")
|
||||
|
||||
# Don't forget the last entry
|
||||
if current_block and current_label and "ipod" in current_label.lower():
|
||||
devices.append({
|
||||
"id": current_block,
|
||||
"name": current_label,
|
||||
"model": "iPod",
|
||||
"mount_point": current_mount if current_mount else None,
|
||||
"mounted": current_mount is not None,
|
||||
"_parent_drive": self._get_parent_drive_from_udisks(current_block),
|
||||
})
|
||||
except FileNotFoundError:
|
||||
self.logger.warning("udisksctl not found")
|
||||
|
||||
# Method 3: Check common mount directories for already-mounted iPods
|
||||
for device in devices:
|
||||
if not device["mounted"]:
|
||||
for base in ["/run/media", "/media", "/mnt"]:
|
||||
if os.path.exists(base):
|
||||
for user_or_device in os.listdir(base):
|
||||
user_path = os.path.join(base, user_or_device)
|
||||
if os.path.isdir(user_path):
|
||||
# Could be /run/media/mat/IPOD or /media/IPOD
|
||||
check_paths = [user_path]
|
||||
if os.path.isdir(user_path):
|
||||
for sub in os.listdir(user_path):
|
||||
sub_path = os.path.join(user_path, sub)
|
||||
if "ipod" in sub.lower() and os.path.isdir(sub_path):
|
||||
check_paths.append(sub_path)
|
||||
|
||||
for check_path in check_paths:
|
||||
if os.path.exists(os.path.join(check_path, "iPod_Control")):
|
||||
device["mount_point"] = check_path
|
||||
device["mounted"] = True
|
||||
break
|
||||
if device["mounted"]:
|
||||
break
|
||||
if device["mounted"]:
|
||||
break
|
||||
|
||||
return devices
|
||||
|
||||
def _get_parent_drive_from_udisks(self, block_name: str) -> Optional[str]:
|
||||
"""Get parent drive device from udisksctl for a partition"""
|
||||
try:
|
||||
# /dev/sdd1 -> /dev/sdd
|
||||
if block_name and block_name.startswith("/dev/"):
|
||||
name = block_name[5:] # sdd1
|
||||
# Strip trailing partition number
|
||||
base = name.rstrip("0123456789")
|
||||
if base:
|
||||
return f"/dev/{base}"
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def _detect_macos(self) -> List[Dict[str, str]]:
|
||||
"""Detect iPod on macOS"""
|
||||
devices = []
|
||||
|
||||
# Check /Volumes for iPod
|
||||
volumes_dir = "/Volumes"
|
||||
if os.path.exists(volumes_dir):
|
||||
for volume in os.listdir(volumes_dir):
|
||||
volume_path = os.path.join(volumes_dir, volume)
|
||||
if "iPod" in volume and os.path.isdir(volume_path):
|
||||
if os.path.exists(os.path.join(volume_path, "iPod_Control")):
|
||||
devices.append({
|
||||
"id": "unknown",
|
||||
"name": volume,
|
||||
"model": "Unknown",
|
||||
"mount_point": volume_path,
|
||||
"mounted": True,
|
||||
})
|
||||
|
||||
return devices
|
||||
|
||||
def _detect_windows(self) -> List[Dict[str, str]]:
|
||||
"""Detect iPod on Windows"""
|
||||
devices = []
|
||||
import string
|
||||
for letter in string.ascii_uppercase:
|
||||
drive = f"{letter}:\\"
|
||||
if os.path.exists(drive):
|
||||
if os.path.exists(os.path.join(drive, "iPod_Control")):
|
||||
devices.append({
|
||||
"id": letter,
|
||||
"name": "iPod",
|
||||
"model": "Unknown",
|
||||
"mount_point": drive,
|
||||
"mounted": True,
|
||||
})
|
||||
return devices
|
||||
|
||||
def mount_device(self, device_id: Optional[str] = None, detected_mount: Optional[str] = None) -> Optional[str]:
|
||||
"""
|
||||
Mount iPod device
|
||||
Mount iPod device using udisksctl (Linux) or verify existing mount
|
||||
|
||||
Args:
|
||||
device_id: Device ID (optional)
|
||||
detected_mount: Already detected mount point (optional)
|
||||
device_id: Device path like /dev/sdd1
|
||||
detected_mount: Already detected mount point
|
||||
|
||||
Returns:
|
||||
Mount point if successful, None otherwise
|
||||
"""
|
||||
# If device is already mounted at detected path, just use it
|
||||
if detected_mount and os.path.exists(detected_mount):
|
||||
# Verify it looks like an iPod
|
||||
if os.path.exists(os.path.join(detected_mount, "iPod_Control")):
|
||||
self.mount_point = detected_mount
|
||||
self.logger.info(f"Using already mounted device at: {detected_mount}")
|
||||
@ -310,106 +272,203 @@ class IPodDevice:
|
||||
|
||||
try:
|
||||
if self.system == "Linux":
|
||||
# Check common mount points for already-mounted USB storage
|
||||
for base in ["/run/media", "/media", "/mnt"]:
|
||||
if os.path.exists(base):
|
||||
for user_dir in os.listdir(base):
|
||||
user_path = os.path.join(base, user_dir)
|
||||
if os.path.isdir(user_path):
|
||||
if os.path.exists(os.path.join(user_path, "iPod_Control")):
|
||||
self.mount_point = user_path
|
||||
self.logger.info(f"Found already mounted iPod at: {user_path}")
|
||||
return user_path
|
||||
|
||||
# Try to mount with ifuse if available
|
||||
if shutil.which("ifuse"):
|
||||
# Create mount point
|
||||
mount_point = "/tmp/ipod_mount"
|
||||
os.makedirs(mount_point, exist_ok=True)
|
||||
|
||||
# Mount with ifuse
|
||||
cmd = ["ifuse", mount_point]
|
||||
if device_id:
|
||||
cmd.extend(["-u", device_id])
|
||||
|
||||
subprocess.run(cmd, check=True)
|
||||
|
||||
self.mount_point = mount_point
|
||||
self.logger.info(f"Device mounted at: {mount_point}")
|
||||
return mount_point
|
||||
else:
|
||||
self.logger.error("ifuse not installed and device not already mounted")
|
||||
return None
|
||||
|
||||
return self._mount_linux(device_id)
|
||||
elif self.system == "Darwin":
|
||||
# On macOS, devices are automatically mounted
|
||||
# Find the iPod mount point
|
||||
result = subprocess.run(
|
||||
["df", "-h"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
for line in result.stdout.strip().split('\n'):
|
||||
if "iPod" in line:
|
||||
# Extract disk identifier
|
||||
parts = line.split()
|
||||
if len(parts) >= 9: # macOS df output format
|
||||
mount_point = parts[8]
|
||||
self.mount_point = mount_point
|
||||
self.logger.info(f"Device found at: {mount_point}")
|
||||
return mount_point
|
||||
|
||||
self.logger.warning("iPod mount point not found")
|
||||
return None
|
||||
|
||||
return self._mount_macos()
|
||||
elif self.system == "Windows":
|
||||
# Windows implementation would go here
|
||||
# This is a placeholder for future implementation
|
||||
self.logger.warning("Windows mounting not fully implemented")
|
||||
return None
|
||||
|
||||
return self._mount_windows()
|
||||
except subprocess.SubprocessError as e:
|
||||
self.logger.error(f"Failed to mount device: {e}")
|
||||
return None
|
||||
|
||||
return None
|
||||
|
||||
def _mount_linux(self, device_id: Optional[str]) -> Optional[str]:
|
||||
"""Mount on Linux using udisksctl"""
|
||||
# device_id should be like /dev/sdd1
|
||||
if not device_id or not device_id.startswith("/dev/"):
|
||||
self.logger.error("No valid device path provided for mounting")
|
||||
return None
|
||||
|
||||
if not os.path.exists(device_id):
|
||||
self.logger.error(f"Device {device_id} does not exist")
|
||||
return None
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["udisksctl", "mount", "-b", device_id],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
self.logger.error(f"udisksctl mount failed: {result.stderr}")
|
||||
return None
|
||||
|
||||
# Parse mount point from output: "Mounted /dev/sdd1 at /run/media/mat/IPOD.\n"
|
||||
output = result.stdout.strip()
|
||||
mount_point = None
|
||||
for line in output.splitlines():
|
||||
if "Mounted" in line and "at" in line:
|
||||
# "Mounted /dev/sdd1 at /run/media/mat/IPOD."
|
||||
parts = line.split(" at ")
|
||||
if len(parts) >= 2:
|
||||
mp = parts[1].strip().rstrip(".")
|
||||
if os.path.exists(mp):
|
||||
mount_point = mp
|
||||
|
||||
if not mount_point:
|
||||
# Fallback: check /run/media for the device
|
||||
for user_dir in os.listdir("/run/media"):
|
||||
user_path = os.path.join("/run/media", user_dir)
|
||||
if os.path.isdir(user_path):
|
||||
for d in os.listdir(user_path):
|
||||
d_path = os.path.join(user_path, d)
|
||||
if os.path.exists(os.path.join(d_path, "iPod_Control")):
|
||||
mount_point = d_path
|
||||
break
|
||||
if mount_point:
|
||||
break
|
||||
|
||||
if mount_point:
|
||||
self.mount_point = mount_point
|
||||
self._block_device = device_id
|
||||
# Extract parent drive: /dev/sdd1 -> /dev/sdd
|
||||
name = device_id[5:]
|
||||
self._parent_drive = f"/dev/{name.rstrip('0123456789')}"
|
||||
self.logger.info(f"Device mounted at: {mount_point}")
|
||||
return mount_point
|
||||
|
||||
self.logger.error("Could not determine mount point")
|
||||
return None
|
||||
|
||||
except FileNotFoundError:
|
||||
self.logger.error("udisksctl not found. Please install udisks2.")
|
||||
return None
|
||||
|
||||
def _mount_macos(self) -> Optional[str]:
|
||||
"""Mount on macOS - devices auto-mount, just find the mount point"""
|
||||
result = subprocess.run(
|
||||
["diskutil", "list"],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
for line in result.stdout.splitlines():
|
||||
if "iPod" in line:
|
||||
df_result = subprocess.run(
|
||||
["df", "-h"],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
for df_line in df_result.stdout.splitlines():
|
||||
if "iPod" in df_line:
|
||||
parts = df_line.split()
|
||||
if len(parts) >= 9:
|
||||
mount_point = parts[8]
|
||||
self.mount_point = mount_point
|
||||
return mount_point
|
||||
|
||||
return None
|
||||
|
||||
def _mount_windows(self) -> Optional[str]:
|
||||
"""Mount on Windows"""
|
||||
self.logger.warning("Windows mounting not fully implemented")
|
||||
return None
|
||||
|
||||
def unmount_device(self) -> bool:
|
||||
"""
|
||||
Unmount iPod device
|
||||
Unmount and safely eject iPod device
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
if not self.mount_point:
|
||||
if not self.mount_point and not self._block_device:
|
||||
self.logger.warning("No device mounted")
|
||||
return False
|
||||
|
||||
try:
|
||||
if self.system == "Linux":
|
||||
# Unmount with fusermount
|
||||
subprocess.run(["fusermount", "-u", self.mount_point], check=True)
|
||||
self.logger.info(f"Device unmounted: {self.mount_point}")
|
||||
self.mount_point = None
|
||||
return True
|
||||
|
||||
return self._unmount_linux()
|
||||
elif self.system == "Darwin":
|
||||
# On macOS, use diskutil
|
||||
subprocess.run(["diskutil", "eject", self.mount_point], check=True)
|
||||
self.logger.info(f"Device ejected: {self.mount_point}")
|
||||
self.mount_point = None
|
||||
return True
|
||||
|
||||
return self._unmount_macos()
|
||||
elif self.system == "Windows":
|
||||
# Windows implementation would go here
|
||||
# This is a placeholder for future implementation
|
||||
self.logger.warning("Windows unmounting not fully implemented")
|
||||
return False
|
||||
|
||||
return self._unmount_windows()
|
||||
except subprocess.SubprocessError as e:
|
||||
self.logger.error(f"Failed to unmount device: {e}")
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def _unmount_linux(self) -> bool:
|
||||
"""Unmount and power-off on Linux using udisksctl"""
|
||||
success = True
|
||||
|
||||
# Step 1: Unmount
|
||||
if self.mount_point or self._block_device:
|
||||
device = self._block_device
|
||||
if not device:
|
||||
# Try to find the device from mount point
|
||||
result = subprocess.run(
|
||||
["findmnt", "-n", "-o", "SOURCE", self.mount_point],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
device = result.stdout.strip()
|
||||
|
||||
if device:
|
||||
result = subprocess.run(
|
||||
["udisksctl", "unmount", "-b", device],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
self.logger.info(f"Device unmounted: {device}")
|
||||
else:
|
||||
self.logger.warning(f"Unmount warning: {result.stderr}")
|
||||
# Not a fatal error - device might already be unmounted
|
||||
|
||||
# Step 2: Power off the drive (safely remove)
|
||||
drive = self._parent_drive
|
||||
if not drive and self._block_device:
|
||||
name = self._block_device[5:]
|
||||
drive = f"/dev/{name.rstrip('0123456789')}"
|
||||
|
||||
if drive:
|
||||
result = subprocess.run(
|
||||
["udisksctl", "power-off", "-b", drive],
|
||||
capture_output=True, check=False, text=True
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
self.logger.info(f"Device powered off: {drive}")
|
||||
else:
|
||||
self.logger.warning(f"Power-off warning: {result.stderr}")
|
||||
# Not fatal - USB may have been unplugged already
|
||||
|
||||
self.mount_point = None
|
||||
self._block_device = None
|
||||
self._parent_drive = None
|
||||
return success
|
||||
|
||||
def _unmount_macos(self) -> bool:
|
||||
"""Eject on macOS"""
|
||||
if self.mount_point:
|
||||
subprocess.run(["diskutil", "eject", self.mount_point], check=True)
|
||||
self.logger.info(f"Device ejected: {self.mount_point}")
|
||||
self.mount_point = None
|
||||
return True
|
||||
return False
|
||||
|
||||
def _unmount_windows(self) -> bool:
|
||||
"""Unmount on Windows"""
|
||||
self.logger.warning("Windows unmounting not fully implemented")
|
||||
return False
|
||||
|
||||
def is_mounted(self) -> bool:
|
||||
"""Check if iPod is currently mounted"""
|
||||
if self.mount_point and os.path.exists(self.mount_point):
|
||||
return os.path.exists(os.path.join(self.mount_point, "iPod_Control"))
|
||||
return False
|
||||
|
||||
def get_device_info(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Get information about the mounted device
|
||||
@ -443,18 +502,6 @@ class IPodDevice:
|
||||
info["has_nano7_db"] = True
|
||||
except Exception:
|
||||
info["has_nano7_db"] = False
|
||||
# Fallback: count files in Music directory
|
||||
music_dir = os.path.join(self.mount_point, "Music")
|
||||
if os.path.exists(music_dir):
|
||||
info["has_music_dir"] = True
|
||||
music_files = []
|
||||
for root, _, files in os.walk(music_dir):
|
||||
for file in files:
|
||||
if file.lower().endswith(('.m4a', '.mp3', '.aac')):
|
||||
music_files.append(os.path.join(root, file))
|
||||
info["music_file_count"] = len(music_files)
|
||||
else:
|
||||
info["has_music_dir"] = False
|
||||
|
||||
return info
|
||||
|
||||
@ -487,6 +534,10 @@ class IPodDevice:
|
||||
from ipod_nano7_db import Nano7Database
|
||||
nano7_db = Nano7Database(self.mount_point)
|
||||
result = nano7_db.add_track(source_file)
|
||||
if result is None:
|
||||
# Track already exists (duplicate)
|
||||
self.logger.info(f"Track already on iPod, skipping: {os.path.basename(source_file)}")
|
||||
return True # Not an error - track is already there
|
||||
self.logger.info(f"Track added to iPod database: {result['title']}")
|
||||
return True
|
||||
except ImportError:
|
||||
@ -541,24 +592,18 @@ class IPodDevice:
|
||||
if dest_dir:
|
||||
full_dest_dir = os.path.join(self.mount_point, dest_dir)
|
||||
else:
|
||||
# Default to Music directory
|
||||
full_dest_dir = os.path.join(self.mount_point, "Music")
|
||||
|
||||
# Create destination directory if it doesn't exist
|
||||
os.makedirs(full_dest_dir, exist_ok=True)
|
||||
|
||||
# Copy directory contents
|
||||
for root, dirs, files in os.walk(source_dir):
|
||||
# Create relative path
|
||||
rel_path = os.path.relpath(root, source_dir)
|
||||
if rel_path == '.':
|
||||
rel_path = ''
|
||||
|
||||
# Create destination subdirectory
|
||||
dest_subdir = os.path.join(full_dest_dir, rel_path)
|
||||
os.makedirs(dest_subdir, exist_ok=True)
|
||||
|
||||
# Copy files
|
||||
for file in files:
|
||||
if file.lower().endswith(('.m4a', '.mp3', '.aac', '.mp4')):
|
||||
src_file = os.path.join(root, file)
|
||||
@ -575,24 +620,30 @@ class IPodDevice:
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
ipod = IPodDevice()
|
||||
|
||||
# Detect devices
|
||||
devices = ipod.detect_devices()
|
||||
for device in devices:
|
||||
print(f"Found device: {device['name']} ({device['model']})")
|
||||
status = "mounted" if device.get("mounted") else "NOT mounted"
|
||||
print(f"Found device: {device['name']} ({device['model']}) - {status}")
|
||||
if device.get("mount_point"):
|
||||
print(f" Mount point: {device['mount_point']}")
|
||||
print(f" Device: {device['id']}")
|
||||
|
||||
# Mount device
|
||||
# Mount and eject if device found
|
||||
if devices:
|
||||
mount_point = ipod.mount_device(devices[0]['id'])
|
||||
if mount_point:
|
||||
print(f"Device mounted at: {mount_point}")
|
||||
device = devices[0]
|
||||
if not device.get("mounted"):
|
||||
mount_point = ipod.mount_device(device["id"])
|
||||
if mount_point:
|
||||
print(f"Mounted at: {mount_point}")
|
||||
|
||||
# Get device info
|
||||
info = ipod.get_device_info()
|
||||
print(f"Total space: {info.get('total_space', 0) / (1024**3):.2f} GB")
|
||||
print(f"Free space: {info.get('free_space', 0) / (1024**3):.2f} GB")
|
||||
info = ipod.get_device_info()
|
||||
print(f"Free space: {info.get('free_space', 0) / 1024**2:.1f} MB")
|
||||
|
||||
# Unmount device
|
||||
ipod.unmount_device()
|
||||
# Unmount
|
||||
ipod.unmount_device()
|
||||
print("Ejected.")
|
||||
else:
|
||||
print(f"Already mounted at: {device['mount_point']}")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user