kbolen/photophetch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Fix iOS helper: async pymobiledevice3 API + auto-discover pipx Python

Browse Source 
The helper script now uses asyncio.run() for all pymobiledevice3 calls
since create_using_usbmux() and AfcService methods are async in newer
versions.

IosConnector: auto-discovers the pipx venv Python at
~/Library/Application Support/pipx/venvs/pymobiledevice3/bin/python3
instead of using system python3 (which doesn't have the module).
Falls back to ~/.local/pipx/... paths too.

AppConfig python3Path acts as an override if auto-discovery fails.
This commit is contained in:
Kyle Bolen
2026-07-28 08:06:13 +00:00
parent 1ae16f0ead
commit 7f11a4b7d6
3 changed files with 81 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/main/resources/scripts/photophetch-ios-helper.py
View File

@@ -2,13 +2,14 @@
"""
photophetch-ios-helper.py
Thin wrapper around pymobiledevice3 AFC for PhotoPhetch iOS connectivity.
Async wrapper around pymobiledevice3 AFC for PhotoPhetch iOS connectivity.
Called by IosConnector as a subprocess.
Usage:
python3 photophetch-ios-helper.py list-devices
python3 photophetch-ios-helper.py list-photos
python3 photophetch-ios-helper.py pull /DCIM/148APPLE/IMG_001.HEIC /tmp/dest.heic
python3 photophetch-ios-helper.py list-devices
python3 photophetch-ios-helper.py delete /DCIM/148APPLE/IMG_001.HEIC
Output is JSON on stdout. Errors go to stderr with exit code 1.
"""
@@ -16,6 +17,7 @@ Output is JSON on stdout. Errors go to stderr with exit code 1.
import sys
import json
import os
import asyncio
from datetime import datetime
MEDIA_EXTENSIONS = {
@@ -30,50 +32,48 @@ VIDEO_EXTENSIONS = {
}
def get_afc_service():
async def get_lockdown():
from pymobiledevice3.lockdown import create_using_usbmux
return await create_using_usbmux()
async def get_afc(lockdown):
from pymobiledevice3.services.afc import AfcService
lockdown = create_using_usbmux()
return AfcService(lockdown), lockdown
return AfcService(lockdown)
def cmd_list_devices():
async def cmd_list_devices():
try:
from pymobiledevice3.usbmux import select_devices_by_connection_type
devices = select_devices_by_connection_type(connection_type='USB')
result = [{'udid': d.serial, 'name': getattr(d, 'label', d.serial)} for d in devices]
lockdown = await get_lockdown()
result = [{
'udid': lockdown.udid,
'name': lockdown.display_name or lockdown.udid,
}]
print(json.dumps(result))
except Exception as e:
# fallback: try lockdown
try:
from pymobiledevice3.lockdown import create_using_usbmux
lockdown = create_using_usbmux()
print(json.dumps([{
'udid': lockdown.udid,
'name': lockdown.display_name or lockdown.udid,
}]))
except Exception as e2:
print(json.dumps([]))
print(f'Error: {e}', file=sys.stderr)
# Return empty list rather than failing — device not connected
print(json.dumps([]))
def cmd_list_photos():
async def cmd_list_photos():
try:
afc, _ = get_afc_service()
lockdown = await get_lockdown()
afc = await get_afc(lockdown)
# List all DCIM subdirectories
try:
dcim_entries = afc.listdir('/DCIM')
except Exception:
dcim_entries = await afc.listdir('/DCIM')
except Exception as e:
print(f'Error listing DCIM: {e}', file=sys.stderr)
print(json.dumps([]))
return
photos = []
current_year = datetime.now().year
for folder in sorted(dcim_entries):
folder_path = f'/DCIM/{folder}'
try:
entries = afc.listdir(folder_path)
entries = await afc.listdir(folder_path)
except Exception:
continue
@@ -84,17 +84,19 @@ def cmd_list_photos():
device_path = f'{folder_path}/{filename}'
# Get file info
# Get file metadata via stat
size_bytes = 0
date_iso = None
try:
info = afc.stat(device_path)
info = await afc.stat(device_path)
size_bytes = int(info.get('st_size', 0))
# st_mtime is nanoseconds since epoch on AFC
mtime_ns = int(info.get('st_mtime', 0))
mtime_s = mtime_ns / 1_000_000_000 if mtime_ns > 1_000_000_000 else mtime_ns
date_iso = datetime.fromtimestamp(mtime_s).isoformat() if mtime_s > 0 else None
# st_mtime may be nanoseconds or seconds depending on version
mtime_s = mtime_ns / 1_000_000_000 if mtime_ns > 1_700_000_000_000 else mtime_ns
if mtime_s > 0:
date_iso = datetime.fromtimestamp(mtime_s).isoformat()
except Exception:
size_bytes = 0
date_iso = None
pass
photos.append({
'devicePath': device_path,
@@ -113,27 +115,23 @@ def cmd_list_photos():
sys.exit(1)
def cmd_pull(device_path: str, local_path: str):
async def cmd_pull(device_path: str, local_path: str):
try:
afc, _ = get_afc_service()
os.makedirs(os.path.dirname(local_path), exist_ok=True)
with afc.open(device_path, 'rb') as remote_f:
with open(local_path, 'wb') as local_f:
while True:
chunk = remote_f.read(64 * 1024)
if not chunk:
break
local_f.write(chunk)
lockdown = await get_lockdown()
afc = await get_afc(lockdown)
os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True)
await afc.pull(device_path, local_path)
print(json.dumps({'ok': True, 'path': local_path}))
except Exception as e:
print(f'Error: {e}', file=sys.stderr)
sys.exit(1)
def cmd_delete(device_path: str):
async def cmd_delete(device_path: str):
try:
afc, _ = get_afc_service()
afc.rm(device_path)
lockdown = await get_lockdown()
afc = await get_afc(lockdown)
await afc.rm(device_path)
print(json.dumps({'ok': True}))
except Exception as e:
print(f'Error: {e}', file=sys.stderr)
@@ -148,13 +146,13 @@ if __name__ == '__main__':
command = sys.argv[1]
if command == 'list-devices':
cmd_list_devices()
asyncio.run(cmd_list_devices())
elif command == 'list-photos':
cmd_list_photos()
asyncio.run(cmd_list_photos())
elif command == 'pull' and len(sys.argv) == 4:
cmd_pull(sys.argv[2], sys.argv[3])
asyncio.run(cmd_pull(sys.argv[2], sys.argv[3]))
elif command == 'delete' and len(sys.argv) == 3:
cmd_delete(sys.argv[2])
asyncio.run(cmd_delete(sys.argv[2]))
else:
print(f'Unknown command: {command}', file=sys.stderr)
sys.exit(1)