en2zmax/neo-pod-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix UI freeze on iPod mount/track-load and add content_hash/sync-metadata

Browse Source 
- Move mount_device, get_device_info, and track scanning to background
  WorkerThread (new 'mount_and_load' task type) so the UI stays responsive
- Fix DeviceMonitor auto-poll to run detect_devices() in a background thread
- Refactor _on_device_detection_finished and _on_mount_clicked to use the
  new worker; remove synchronous _set_device_mounted
- Avoid double scan (get_track_count + get_all_tracks) — scan once in worker
- Add content_hash fingerprint to _scan_ipod_files and library_cache for
  track matching
- Add update_track_metadata / _write_databases_from_tracks with artwork
  override support in Nano7Database
- Add 'Sync Metadata' button and context menu action in LibraryTab
This commit is contained in:
Maksim Totmin 2026-06-01 12:37:49 +07:00
parent 25bb8105ac
commit 3bfb0ebbfe
6 changed files with 437 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/device_monitor.py
View File

@ -1,13 +1,25 @@
import logging
from typing import Set
from PyQt6.QtCore import QObject, QTimer, pyqtSignal
from PyQt6.QtCore import QObject, QTimer, QThread, pyqtSignal
from ipod_device import IPodDevice
logger = logging.getLogger(__name__)
class _DetectWorker(QThread):
finished = pyqtSignal(object)
def run(self):
try:
devices = IPodDevice().detect_devices()
self.finished.emit(devices)
except Exception:
logger.exception("Device detection failed")
self.finished.emit([])
class DeviceMonitor(QObject):
state_changed = pyqtSignal(list)
@ -17,6 +29,7 @@ class DeviceMonitor(QObject):
self.timer = QTimer(self)
self.timer.timeout.connect(self._poll)
self._device_ids: Set[str] = set()
self._poll_worker = None
def start(self, interval_ms: int = 3000):
self._device_ids.clear()
@ -25,14 +38,20 @@ class DeviceMonitor(QObject):
def stop(self):
self.timer.stop()
if self._poll_worker and self._poll_worker.isRunning():
self._poll_worker.quit()
self._poll_worker.wait(3000)
self._device_ids.clear()
def _poll(self):
try:
devices = IPodDevice().detect_devices()
if self._poll_worker and self._poll_worker.isRunning():
return
self._poll_worker = _DetectWorker()
self._poll_worker.finished.connect(self._on_devices_detected)
self._poll_worker.start()
def _on_devices_detected(self, devices):
current_ids = {d["id"] for d in devices}
if current_ids != self._device_ids:
self._device_ids = current_ids
self.state_changed.emit(devices)
except Exception:
logger.exception("DeviceMonitor poll failed")

151
src/ipod_nano7_db.py
View File

@ -132,12 +132,27 @@ MASTER_CONTAINER_PID = 203092939621887772
MUSIC_CONTAINER_PID = 4872745547565090077
_REMOVE_ARTWORK = object()
"""Sentinel: used in artwork_overrides to remove artwork from a track."""
# ── Helpers ──────────────────────────────────────────────────────────
def _to_mac_epoch(ts: int) -> int:
return ts + MAC_EPOCH_OFFSET if ts > 0 else 0
def content_hash(filepath: str, max_bytes: int = 65536) -> str:
"""SHA1 of first max_bytes of audio data (fast content fingerprint)."""
h = hashlib.sha1()
try:
with open(filepath, "rb") as f:
h.update(f.read(max_bytes))
except OSError:
return ""
return h.hexdigest()
def _read_audio_tags(filepath: str) -> dict[str, Any]:
"""Extract title, artist, album, duration, etc. from an audio file."""
info: dict[str, Any] = {}
@ -374,6 +389,7 @@ class Nano7Database:
tags["file_name"] = fname
tags["file_size"] = os.path.getsize(fpath)
tags["pid"] = self._hash_to_pid(fpath)
tags["content_hash"] = content_hash(fpath)
results.append(tags)
return results
@ -471,6 +487,7 @@ class Nano7Database:
"year": s.get("year", 0),
"ipod_path": None,
"file_path": s["file_path"],
"content_hash": s.get("content_hash", ""),
})
return tracks
@ -602,15 +619,127 @@ class Nano7Database:
logger.exception("iTunesDB sync failed: %s", e)
def _sync_itunescdb_impl(self) -> None:
# 1. Scan all audio files on iPod
scanned = self._scan_ipod_files()
if not scanned:
logger.warning("No audio files found on iPod — writing empty databases")
# Write minimal empty databases
self._write_empty_databases()
return
self._write_databases_from_tracks(scanned)
# 2. Prepare artwork for all tracks
def find_track_by_content_hash(self, content_hash: str) -> dict | None:
"""Find an iPod track by its content hash."""
if not content_hash:
return None
for t in self.get_all_tracks():
if t.get("content_hash") == content_hash:
return t
return None
def find_track_by_metadata(
self, title: str, artist: str, album: str = "",
) -> dict | None:
"""Find an iPod track by title+artist+album (case-insensitive, fallback matching)."""
for t in self.get_all_tracks():
if (t["title"].lower() == title.lower() and
t["artist"].lower() == artist.lower() and
(t.get("album") or "").lower() == album.lower()):
return t
return None
def update_track_metadata(
self,
pid: int,
new_metadata: dict,
local_source_path: str | None = None,
) -> bool:
"""Update metadata for a single iPod track without touching audio files.
Only updates the databases (iTunesDB + SQLite + ArtworkDB).
The audio file on the iPod is never modified — new metadata is
written directly into the database records.
Args:
pid: The iPod track pid (from get_all_tracks).
new_metadata: Dict with optional keys: title, artist, album,
album_artist, genre, composer, year, track_number,
disc_number, comment.
local_source_path: If provided, artwork is extracted from this
local file and encoded into iPod-native formats.
Returns:
True if the track was found and updated, False otherwise.
"""
scanned = self._scan_ipod_files()
if not scanned:
logger.warning("No tracks on iPod to update")
return False
target_idx = None
for idx, s in enumerate(scanned):
if s["pid"] == pid:
target_idx = idx
break
if target_idx is None:
logger.warning("Track with pid=%d not found on iPod", pid)
return False
target = scanned[target_idx]
text_fields = {
"title": "title",
"artist": "artist",
"album": "album",
"album_artist": "album_artist",
"genre": "genre",
"composer": "composer",
"comment": "comment",
}
for meta_key, scan_key in text_fields.items():
if meta_key in new_metadata:
target[scan_key] = str(new_metadata[meta_key]) if new_metadata[meta_key] else ""
if "year" in new_metadata:
target["year"] = int(new_metadata["year"] or 0)
if "track_number" in new_metadata:
target["track_number"] = int(new_metadata["track_number"] or 0)
if "disc_number" in new_metadata:
target["disc_number"] = int(new_metadata["disc_number"] or 0)
artwork_overrides: dict[str, object] = {}
if local_source_path and os.path.exists(local_source_path):
try:
art_result = prepare_artwork_for_track(local_source_path)
if art_result is not None:
artwork_overrides[target["file_path"]] = art_result
else:
artwork_overrides[target["file_path"]] = _REMOVE_ARTWORK
except Exception as e:
logger.warning("Failed to extract artwork from %s: %s",
local_source_path, e)
self._write_databases_from_tracks(scanned, artwork_overrides=artwork_overrides)
logger.info("Updated metadata for track: %s", target.get("title", "?"))
return True
def _write_databases_from_tracks(
self,
scanned: list[dict],
artwork_overrides: dict[str, object] | None = None,
) -> None:
"""Build artwork, convert to TrackInfo, write iTunesDB + SQLite.
Args:
scanned: List of track dicts from _scan_ipod_files().
artwork_overrides: Optional dict mapping iPod file_path →
(art_hash, {fmt_id: EncodedFormatPayload}) for tracks whose
artwork should come from a local file instead of the iPod file.
Use the ``_REMOVE_ARTWORK`` sentinel to explicitly remove
artwork for a track. Omit a key to keep existing iPod artwork.
"""
artwork_overrides = artwork_overrides or {}
# 1. Prepare artwork for all tracks
artwork_map: dict[str, tuple[int, int, int]] = {}
art_entries: list[ArtworkEntry] = []
art_hash_to_img_id: dict[str, int] = {}
@ -618,10 +747,16 @@ class Nano7Database:
for idx, s in enumerate(scanned):
fpath = s["file_path"]
override = artwork_overrides.get(fpath)
if override is None:
try:
art_result = prepare_artwork_for_track(fpath)
except Exception:
art_result = None
elif override is _REMOVE_ARTWORK:
art_result = None
else:
art_result = override
if art_result is not None:
art_hash, formats = art_result
art_count = len(formats)
@ -657,11 +792,11 @@ class Nano7Database:
except Exception as e:
logger.warning("ArtworkDB write failed: %s", e)
# 3. Convert to iOpenPod TrackInfo
# 2. Convert to iOpenPod TrackInfo
tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map)
logger.info("Built %d iOpenPod TrackInfo objects", len(tracks))
# 4. Determine capabilities for this device
# 3. Determine capabilities for this device
ipod_device = _import_iop("ipod_device")
capabilities = None
try:
@ -675,7 +810,7 @@ class Nano7Database:
except Exception as e:
logger.debug("Capabilities detection failed: %s", e)
# 5. Generate and write binary iTunesDB first (to get db_pid)
# 4. Generate and write binary iTunesDB first (to get db_pid)
try:
firewire_id = ipod_device.get_firewire_id(self.mountpoint)
except Exception:
@ -695,7 +830,7 @@ class Nano7Database:
return
logger.info("iTunesDB written (%d tracks)", len(tracks))
# 6. Extract db_pid from the binary DB for SQLite cross-reference
# 5. Extract db_pid from the binary DB for SQLite cross-reference
db_pid = 0
try:
ipod_dev = _import_iop("ipod_device")
@ -709,7 +844,7 @@ class Nano7Database:
except Exception as e:
logger.warning("Could not extract db_pid: %s", e)
# 7. Write SQLite databases (Nano 6G/7G require this)
# 6. Write SQLite databases (Nano 6G/7G require this)
try:
write_sqlite_databases = _import_iop("SQLiteDB_Writer").write_sqlite_databases
sqlite_ok = write_sqlite_databases(

26
src/library_cache.py
View File

@ -5,6 +5,7 @@ Caches extracted tag data per file path with mtime-based invalidation.
Stored as JSON in XDG cache directory.
"""
import hashlib
import json
import logging
import os
@ -17,11 +18,23 @@ logger = logging.getLogger(__name__)
_CACHE_FILE = "library_cache.json"
def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
"""SHA1 of first max_bytes of audio data (fast content fingerprint)."""
h = hashlib.sha1()
try:
with open(filepath, "rb") as f:
h.update(f.read(max_bytes))
except OSError:
return ""
return h.hexdigest()
class LibraryCache:
"""JSON-based cache of audio file metadata.
Keys are absolute file paths. Values are dicts of metadata
plus an internal ``_mtime`` field for change detection.
plus internal ``_mtime`` and ``_content_hash`` fields for
change detection and track matching.
Cached entries are returned only when the file still exists
and its ``os.path.getmtime`` matches the stored value.
@ -64,7 +77,15 @@ class LibraryCache:
actual_mtime = os.path.getmtime(file_path)
if actual_mtime != entry.get("_mtime", 0):
return None
return {k: v for k, v in entry.items() if k != "_mtime"}
result = {}
for k, v in entry.items():
if k == "_mtime":
continue
if k == "_content_hash":
result["content_hash"] = v
else:
result[k] = v
return result
def put(self, file_path: str, data: Dict[str, Any]) -> None:
entry = dict(data)
@ -72,6 +93,7 @@ class LibraryCache:
entry["_mtime"] = os.path.getmtime(file_path)
except OSError:
return
entry["_content_hash"] = _content_hash(file_path)
self._data[file_path] = entry
def remove(self, file_path: str) -> None:

83
src/ui/ipod_tab.py
View File

@ -132,21 +132,9 @@ class iPodTab(QWidget):
if self.ipod_devices:
device = self.ipod_devices[0]
device_id = device["id"]
detected_mount = device.get("mount_point")
already_mounted = device.get("mounted", False)
if already_mounted and detected_mount:
self._set_device_mounted(device, detected_mount)
else:
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if mount_point:
device["mount_point"] = mount_point
device["mounted"] = True
self._set_device_mounted(device, mount_point)
else:
self._set_device_unmounted(device)
self._start_mount_and_load(device, detected_mount, already_mounted)
else:
self._set_device_not_found()
else:
@ -154,10 +142,31 @@ class iPodTab(QWidget):
self._cleanup_worker()
def _set_device_mounted(self, device: dict, mount_point: str):
def _start_mount_and_load(self, device: dict, mount_point=None, already_mounted=False):
self.device_status_label.setText("Mounting and loading tracks...")
self.mount_button.setEnabled(False)
self.worker_thread = WorkerThread(
task_type="mount_and_load",
device=device,
mount_point=mount_point,
already_mounted=already_mounted,
)
self.worker_thread.progress_signal.connect(self._on_mount_load_progress)
self.worker_thread.finished_signal.connect(self._on_mount_load_finished)
self.worker_thread.start()
def _on_mount_load_progress(self, progress, status):
self.device_info_label.setText(status)
def _on_mount_load_finished(self, success, message, result):
if success:
device = result["device"]
mount_point = result["mount_point"]
device_info = result["device_info"]
tracks = result["tracks"]
self.current_mount_point = mount_point
ipod = IPodDevice(mount_point=mount_point)
device_info = ipod.get_device_info()
device["info"] = device_info
free_space = device_info.get("free_space", 0)
@ -174,7 +183,26 @@ class iPodTab(QWidget):
self.device_mounted.emit(mount_point)
self._load_ipod_tracks()
self.transferred_list.clear()
for track in tracks:
display = f"{track['artist']} — {track['title']}"
if track.get("album"):
display += f" ({track['album']})"
item = QListWidgetItem(display)
item.setData(Qt.ItemDataRole.UserRole, track)
self.transferred_list.addItem(item)
self.track_count_label.setText(f"Tracks on device: {len(tracks)}")
self.export_button.setEnabled(len(tracks) > 0)
else:
self.device_status_label.setText(f"Status: {message}")
self.mount_button.setEnabled(True)
if self.ipod_devices:
self._set_device_unmounted(self.ipod_devices[0])
else:
self._set_device_not_found()
self._cleanup_worker()
def _set_device_unmounted(self, device: dict):
self.current_mount_point = None
@ -213,27 +241,8 @@ class iPodTab(QWidget):
return
device = self.ipod_devices[0]
device_id = device["id"]
detected_mount = device.get("mount_point")
self.device_status_label.setText("Mounting...")
self.mount_button.setEnabled(False)
try:
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if mount_point:
device["mount_point"] = mount_point
device["mounted"] = True
self._set_device_mounted(device, mount_point)
else:
self.device_status_label.setText("Status: Mount failed")
self.mount_button.setEnabled(True)
QMessageBox.warning(self, "Mount Error", "Failed to mount iPod. Check permissions and try again.")
except Exception as e:
self.device_status_label.setText("Status: Mount error")
self.mount_button.setEnabled(True)
QMessageBox.warning(self, "Mount Error", f"Error mounting iPod: {e}")
self._start_mount_and_load(device, detected_mount, already_mounted=False)
def _on_eject_clicked(self):
reply = QMessageBox.question(

92
src/ui/library_tab.py
View File

@ -214,6 +214,12 @@ class LibraryTab(QWidget):
self.library_transfer_btn.clicked.connect(self._on_library_transfer_clicked)
toolbar.addWidget(self.library_transfer_btn)
self.library_sync_btn = QPushButton("\uD83D\uDD04 Sync Metadata")
self.library_sync_btn.setEnabled(False)
self.library_sync_btn.setToolTip("Sync metadata and artwork for all tracks on iPod")
self.library_sync_btn.clicked.connect(self._on_sync_metadata_clicked)
toolbar.addWidget(self.library_sync_btn)
self.library_remove_btn = QPushButton("\uD83D\uDDD1\uFE0F Remove")
self.library_remove_btn.setEnabled(False)
self.library_remove_btn.clicked.connect(self._on_library_remove_selected)
@ -233,6 +239,7 @@ class LibraryTab(QWidget):
self.library_add_btn,
self.library_convert_btn,
self.library_transfer_btn,
self.library_sync_btn,
self.library_remove_btn,
self.library_edit_btn,
refresh_btn,
@ -717,7 +724,9 @@ class LibraryTab(QWidget):
has_ready = ready_count > 0
has_source = source_count > 0
self.library_transfer_btn.setEnabled(has_ready and self._current_mount_point is not None)
has_mount = self._current_mount_point is not None
self.library_transfer_btn.setEnabled(has_ready and has_mount)
self.library_sync_btn.setEnabled(has_ready and has_mount)
self.library_convert_btn.setEnabled(has_source)
self.library_remove_btn.setEnabled(len(all_tracks) > 0)
self.library_edit_btn.setEnabled(len(all_tracks) > 0)
@ -984,6 +993,8 @@ class LibraryTab(QWidget):
play_action.triggered.connect(lambda: self._play_track_from_index(row))
if is_ready and self._current_mount_point is not None:
sync_action = menu.addAction("\uD83D\uDD04 Sync Metadata to iPod")
sync_action.triggered.connect(self._on_sync_metadata_clicked)
transfer_action = menu.addAction("\uD83D\uDCE4 Transfer to iPod")
transfer_action.triggered.connect(self._on_library_transfer_clicked)
else:
@ -1129,6 +1140,84 @@ class LibraryTab(QWidget):
self._cleanup_worker()
def _on_sync_metadata_clicked(self):
selected_tracks = self._get_selected_ready_tracks()
if not selected_tracks:
QMessageBox.warning(self, "Error", "No tracks selected.")
return
if not self._current_mount_point:
QMessageBox.warning(self, "Error", "No iPod device mounted.")
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
self.library_sync_btn.setEnabled(False)
self.library_status.setText("Matching tracks on iPod...")
self.library_progress.setValue(0)
from ipod_nano7_db import Nano7Database, content_hash as compute_content_hash
db = Nano7Database(self._current_mount_point)
ipod_tracks = db.get_all_tracks()
ipod_by_hash = {t.get("content_hash", ""): t for t in ipod_tracks if t.get("content_hash")}
sync_batch = []
errors = []
for track_info, local_path in selected_tracks:
ch = compute_content_hash(local_path)
ipod_track = ipod_by_hash.get(ch)
if ipod_track is None:
ipod_track = db.find_track_by_metadata(
track_info.title, track_info.artist, track_info.album or "",
)
if ipod_track is None:
errors.append(f"'{track_info.title}' — not found on iPod")
continue
sync_batch.append({
"ipod_pid": ipod_track["pid"],
"local_path": local_path,
"new_metadata": {
"title": track_info.title,
"artist": track_info.artist,
"album": track_info.album or "",
"track_number": track_info.track_number or 0,
"genre": track_info.genre or "",
},
})
if not sync_batch:
msg = "No matching tracks found on iPod"
if errors:
msg += ":\n" + "\n".join(errors[:5])
QMessageBox.warning(self, "Sync Error", msg)
return
if errors:
logger.warning("Sync metadata skipped for %d unmatched track(s)", len(errors))
self.worker_thread = WorkerThread(
task_type="sync_metadata",
sync_batch=sync_batch,
mount_point=self._current_mount_point,
)
self.worker_thread.progress_signal.connect(self._on_library_progress)
self.worker_thread.finished_signal.connect(self._on_sync_metadata_finished)
self.worker_thread.start()
def _on_sync_metadata_finished(self, success, message, result):
self.library_sync_btn.setEnabled(True)
if success:
self.library_status.setText(message)
QMessageBox.information(self, "Sync Complete", message)
else:
self.library_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Sync Error", message)
self._cleanup_worker()
def _cleanup_worker(self):
if self.worker_thread:
self.worker_thread.wait(3000)
@ -1168,6 +1257,7 @@ class LibraryTab(QWidget):
def on_device_unmounted(self):
self._current_mount_point = None
self.library_transfer_btn.setEnabled(False)
self.library_sync_btn.setEnabled(False)
self.library_progress.setVisible(False)
self.library_status.setVisible(False)

88
src/worker.py
View File

@ -39,9 +39,11 @@ class WorkerThread(QThread):
"convert": self._run_convert,
"transfer": self._run_transfer,
"detect_devices": self._run_detect_devices,
"mount_and_load": self._run_mount_and_load,
"export_ipod": self._run_export_ipod,
"delete_tracks": self._run_delete_tracks,
"remove_duplicates": self._run_remove_duplicates,
"sync_metadata": self._run_sync_metadata,
}
handler = handlers.get(self.task_type)
if handler:
@ -168,6 +170,48 @@ class WorkerThread(QThread):
self.finished_signal.emit(True, f"Found {len(devices)} iPod devices", devices)
def _run_mount_and_load(self):
import shutil
device = self.kwargs.get("device", {})
mount_point = self.kwargs.get("mount_point")
already_mounted = self.kwargs.get("already_mounted", False)
if not already_mounted:
device_id = device.get("id")
detected_mount = device.get("mount_point")
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if not mount_point:
self.finished_signal.emit(False, "Failed to mount iPod", None)
return
device["mount_point"] = mount_point
device["mounted"] = True
from ipod_nano7_db import Nano7Database
usage = shutil.disk_usage(mount_point)
device_info = {
"mount_point": mount_point,
"total_space": usage.total,
"free_space": usage.free,
"used_space": usage.used,
}
db = Nano7Database(mount_point)
tracks = db.get_all_tracks()
device_info["track_count"] = len(tracks)
result = {
"device": device,
"mount_point": mount_point,
"device_info": device_info,
"tracks": tracks,
}
self.finished_signal.emit(
True, f"Loaded {len(tracks)} tracks from iPod", result
)
def _run_export_ipod(self):
tracks = self.kwargs.get("tracks", [])
dest_dir = self.kwargs.get("dest_dir", "")
@ -221,6 +265,50 @@ class WorkerThread(QThread):
removed = db.remove_duplicates(progress_callback=progress)
self.finished_signal.emit(True, f"Removed {len(removed)} duplicate(s)", removed)
def _run_sync_metadata(self):
"""Update metadata and artwork for tracks already on iPod (batch)."""
sync_batch = self.kwargs.get("sync_batch", [])
mount_point = self.kwargs.get("mount_point")
if not sync_batch or not mount_point:
self.finished_signal.emit(False, "Missing parameters for sync", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
total = len(sync_batch)
synced = 0
errors = []
for i, item in enumerate(sync_batch):
if not self.is_running:
break
self.progress_signal.emit(
int((i / total) * 100),
f"Syncing metadata: {item['new_metadata'].get('title', '?')}...",
)
try:
success = db.update_track_metadata(
pid=item["ipod_pid"],
new_metadata=item["new_metadata"],
local_source_path=item.get("local_path"),
)
if success:
synced += 1
else:
errors.append(item['new_metadata'].get('title', '?'))
except Exception as e:
logger.exception("Metadata sync failed for track: %s", e)
errors.append(f"{item['new_metadata'].get('title', '?')}: {e}")
msg = f"Synced metadata for {synced}/{total} track(s)"
if errors:
msg += f" ({len(errors)} error(s))"
self.progress_signal.emit(100, msg)
self.finished_signal.emit(synced > 0, msg, {"synced": synced, "errors": errors})
def stop(self):
self.is_running = False
self.wait()