afc: add progress_bar option when pulling files

pymobiledevice3/services/afc.py

@@ -22,7 +22,7 @@
 from pygments import formatters, highlight, lexers
 from pygnuutils.cli.ls import ls as ls_cli
 from pygnuutils.ls import Ls, LsStub
-from tqdm import trange
+from tqdm.auto import trange
 from xonsh.built_ins import XSH
 from xonsh.cli_utils import Annotated, Arg, ArgParserAlias
 from xonsh.main import main as xonsh_main
@@ -225,7 +225,7 @@ def __init__(self, lockdown: LockdownServiceProvider, service_name: str = None):
         self.packet_num = 0
 
     def pull(self, relative_src: str, dst: str, match: Optional[Pattern] = None, callback: Optional[Callable] = None,
-             src_dir: str = '', ignore_errors: bool = False) -> None:
+             src_dir: str = '', ignore_errors: bool = False, progress_bar: bool = True) -> None:
         src = self.resolve_path(posixpath.join(src_dir, relative_src))
 
         if not self.isdir(src):
@@ -239,7 +239,11 @@ def pull(self, relative_src: str, dst: str, match: Optional[Pattern] = None, cal
                 else:
                     left_size = src_size
                     handle = self.fopen(src)
-                    for _ in trange(src_size // MAXIMUM_READ_SIZE + 1):
+                    if progress_bar:
+                        pb = trange(src_size // MAXIMUM_READ_SIZE + 1)
+                    else:
+                        pb = range(src_size // MAXIMUM_READ_SIZE + 1)
+                    for _ in pb:
                         f.write(self.fread(handle, min(MAXIMUM_READ_SIZE, left_size)))
                         left_size -= MAXIMUM_READ_SIZE
                     self.fclose(handle)
@@ -263,10 +267,12 @@ def pull(self, relative_src: str, dst: str, match: Optional[Pattern] = None, cal
                 try:
                     if self.isdir(src_filename):
                         dst_filename.mkdir(exist_ok=True)
-                        self.pull(src_filename, str(dst_path), callback=callback, ignore_errors=ignore_errors)
+                        self.pull(src_filename, str(dst_path), callback=callback, ignore_errors=ignore_errors,
+                                  progress_bar=progress_bar)
                         continue
 
-                    self.pull(src_filename, str(dst_path), callback=callback, ignore_errors=ignore_errors)
+                    self.pull(src_filename, str(dst_path), callback=callback, ignore_errors=ignore_errors,
+                              progress_bar=progress_bar)
 
                 except Exception as afc_exception:
                     if not ignore_errors:
@@ -870,11 +876,12 @@ def _do_rm(self, file: Annotated[list[str], Arg(nargs='+', completer=path_comple
             self.afc.rm(self.relative_path(filename))
 
     def _do_pull(self, remote_path: Annotated[str, Arg(completer=path_completer)], local_path: str,
-                 ignore_errors: bool = False):
+                 ignore_errors: bool = False, progress_bar: bool = True):
         def log(src, dst):
             print(f'{src} --> {dst}')
 
-        self.afc.pull(remote_path, local_path, callback=log, src_dir=self.cwd, ignore_errors=ignore_errors)
+        self.afc.pull(remote_path, local_path, callback=log, src_dir=self.cwd, ignore_errors=ignore_errors,
+                      progress_bar=progress_bar)
 
     def _do_push(self, local_path: str, remote_path: Annotated[str, Arg(completer=path_completer)]):
         def log(src, dst):

pymobiledevice3/services/crash_reports.py

@@ -79,13 +79,15 @@ def ls(self, path: str = '/', depth: int = 1) -> list[str]:
         """
         return list(self.afc.dirlist(path, depth))[1:]  # skip the root path '/'
 
-    def pull(self, out: str, entry: str = '/', erase: bool = False, match: Optional[str] = None) -> None:
+    def pull(self, out: str, entry: str = '/', erase: bool = False, match: Optional[str] = None,
+             progress_bar: bool = True) -> None:
         """
         Pull crash reports from the device.
         :param out: Directory to pull crash reports to.
         :param entry: File or Folder to pull.
         :param erase: Whether to erase the original file from the CrashReports directory.
         :param match: Regex to match against file and directory names to pull.
+        :param progress_bar: Whether to show a progress bar when pulling large files.
         """
 
         def log(src: str, dst: str) -> None:
@@ -95,7 +97,7 @@ def log(src: str, dst: str) -> None:
                     self.afc.rm_single(src, force=True)
 
         match = None if match is None else re.compile(match)
-        self.afc.pull(entry, out, match, callback=log)
+        self.afc.pull(entry, out, match, callback=log, progress_bar=progress_bar)
 
     def flush(self) -> None:
         """ Trigger com.apple.crashreportmover to flush all products into CrashReports directory """