Faster iProX/ProteomeXchange downloads: expose -w parallel files, wire download-px-raw-files, opt-in iProX Aspera

docs/usage.md

@@ -239,7 +239,59 @@ pridepy download-px-raw-files \
 | --- | --- | --- |
 | `-a, --accession` | ProteomeXchange accession (e.g. `PXD039236`). `--px` is a deprecated alias | required |
 | `-o, --output-folder` | Destination directory | required |
+| `-p, --protocol` | Transfer protocol: `ftp`, `aspera`, `globus`, `s3` (FTP-first with fallback) | `ftp` |
+| `-w, --parallel-files` | Download 1–32 files concurrently (across-file concurrency) | `1` |
+| `-t, --threads` | Parallel HTTP Range threads per file (1–32) for fast per-file downloads | `1` |
 | `--skip-if-downloaded-already` | Skip files already present locally | off |
+| `--preserve-structure` | Recreate the dataset's subdirectory layout under the output folder | off |
+| `--iprox-user` | iProX account username (with `--protocol aspera`; env fallback: `IPROX_USER`) | — |
+
+The iProX Aspera password is never accepted as a command-line flag. Set the
+`IPROX_ASPERA_PASSWORD` environment variable, or omit it and `pridepy` will
+prompt for it securely (hidden input) when `--protocol aspera` is used.
+
+### Fast downloads: parallel files and per-file segments
+
+Combine `-w` (files in parallel) and `-t` (Range segments per file) for fast bulk downloads.
+The total concurrent connections is approximately `parallel_files × threads`.
+
+**Parallel across files (recommended for most users, no account required):**
+
+```bash
+# Download up to 8 files concurrently from ProteomeXchange
+pridepy download-px-raw-files \
+  -a PXD077178 \
+  -o ./PXD077178 \
+  -w 8
+```
+
+**Combine parallel files with per-file segments:**
+
+```bash
+# Download 8 files in parallel, each split into 4 Range segments
+pridepy download-px-raw-files \
+  -a PXD077178 \
+  -o ./out \
+  -w 8 \
+  -t 4
+```
+
+### Fast downloads with iProX Aspera (account required)
+
+iProX offers Aspera for very large bulk transfers. Aspera is faster than HTTP
+on high-bandwidth connections but requires an iProX account. Combine
+`--protocol aspera` with `--iprox-user`; the password is read from
+`IPROX_ASPERA_PASSWORD` or prompted for securely (never passed as a flag):
+
+```bash
+# Download via iProX Aspera with 8-file parallelism
+IPROX_ASPERA_PASSWORD=your_password pridepy download-px-raw-files \
+  -a PXD077178 \
+  -o ./out \
+  --protocol aspera \
+  --iprox-user your_username \
+  -w 8
+```
 
 ### Go directly to the hosting repository (native MassIVE / JPOST / iProX accessions)
 
@@ -268,7 +320,7 @@ How each repository is enumerated:
 
 - **MassIVE** walks the FTPS tree at `massive-ftp.ucsd.edu` (the server requires TLS). MassIVE distributes datasets across versioned root directories (`/v01`–`/vNN`); `pridepy` discovers the correct root automatically. If FTP/FTPS is blocked by the network, `pridepy` falls back to HTTPS: it lists the dataset from the GNPS2 file index (`datasetcache.gnps2.org`) and downloads each file from the ProteoSAFe endpoint at `massive.ucsd.edu` (byte-identical to the FTPS copy).
 - **JPOST** lists files through the JSON PROXI endpoint at `https://repository.jpostdb.org/proxi/datasets/<JPSTxxxxxx>` and downloads from `ftp.jpostdb.org` over plain FTP. The PROXI listing avoids the source-IP connection limit JPOST enforces on FTP.
-- **iProX** fetches the dataset's ProteomeXchange XML from `http://download.iprox.org/<accession>/PX_<accession>.xml`, then downloads each referenced file from the same host over anonymous HTTP (with `Range` support for resume). iProX also exposes Aspera (`faspe://`) with username/password for very large bulk transfers; `pridepy` uses the public HTTP endpoint so no iProX credentials are required.
+- **iProX** fetches the dataset's ProteomeXchange XML from `http://download.iprox.org/<accession>/PX_<accession>.xml`, then downloads each referenced file from the same host over anonymous HTTP (with `Range` support for resume). iProX also exposes Aspera with username/password for very large bulk transfers; `pridepy` uses the public HTTP endpoint so no iProX credentials are required.
 
 `download-all-public-raw-files` retrieves the files stored under the dataset's
 `raw/` collection. These direct downloads support resume (REST for FTP,

pridepy/download/client.py

@@ -318,8 +318,21 @@ def download_px_raw_files(
         output_folder: str,
         skip_if_downloaded_already: bool = True,
         flatten: bool = True,
+        parallel_files: int = 1,
+        download_threads: int = 1,
+        protocol: str = "ftp",
+        iprox_user: Optional[str] = None,
+        iprox_password: Optional[str] = None,
     ) -> None:
         """Delegate to :meth:`ProteomeXchangeProvider.download_from_accession_or_url`."""
         return ProteomeXchangeProvider().download_from_accession_or_url(
-            px_id_or_url, output_folder, skip_if_downloaded_already, flatten=flatten
+            px_id_or_url,
+            output_folder,
+            skip_if_downloaded_already,
+            flatten=flatten,
+            parallel_files=parallel_files,
+            download_threads=download_threads,
+            protocol=protocol,
+            iprox_user=iprox_user,
+            iprox_password=iprox_password,
         )

pridepy/download/iprox.py

@@ -14,6 +14,8 @@
 import logging
 import os
 import re
+import subprocess
+from concurrent.futures import ThreadPoolExecutor, as_completed
 import defusedxml.ElementTree as ET
 from typing import ClassVar, Dict, List, Optional
 from urllib.parse import urlparse
@@ -23,6 +25,7 @@
 from pridepy.download import registry
 from pridepy.download.base import Provider
 from pridepy.download.jpost import JpostProvider
+from pridepy.download.transport import _safe_join
 
 
 @registry.register
@@ -34,6 +37,9 @@ class IproxProvider(Provider):
     PX_XML_URL_TEMPLATE: ClassVar[str] = (
         "http://download.iprox.org/{accession}/PX_{accession}.xml"
     )
+    ASPERA_HOST: ClassVar[str] = "download.iprox.org"
+    ASPERA_PORT: ClassVar[str] = "33001"
+    ASPERA_ROOT: ClassVar[str] = "/data/iprox"
     # iProX PX XML uses the same PSI-MS cvParam "name" values as JPOST PROXI,
     # so we reuse JpostProvider's category map.
     PX_CATEGORY_MAP: ClassVar[Dict[str, str]] = JpostProvider.PROXI_CATEGORY_MAP
@@ -45,6 +51,134 @@ def matches(accession: str) -> bool:
             return False
         return bool(re.fullmatch(r"IPX\d{7,10}", accession.upper()))
 
+    @staticmethod
+    def _ascp_binary() -> str:
+        # Reuse PRIDE's bundled ascp binary resolution.
+        from pridepy.download.pride import PrideProvider
+        return PrideProvider.get_ascp_binary()
+
+    @classmethod
+    def _aspera_download_one(
+        cls,
+        ascp: str,
+        url: str,
+        relpath: Optional[str],
+        output_folder: str,
+        user: str,
+        password: str,
+        maximum_bandwidth: str,
+        skip_if_downloaded_already: bool,
+        env: Dict[str, str],
+    ) -> Optional[str]:
+        """Download a single URL via ascp. Returns ``url`` on failure, else None."""
+        path = urlparse(url).path.lstrip("/")  # e.g. IPX.../.../a.raw
+        source = f"{user}@{cls.ASPERA_HOST}:{cls.ASPERA_ROOT}/{path}"
+        if relpath:
+            dest = _safe_join(output_folder, relpath)
+        else:
+            dest = os.path.join(output_folder, os.path.basename(urlparse(url).path))
+        dest_parent = os.path.dirname(dest) or output_folder
+        os.makedirs(dest_parent, exist_ok=True)
+        if (
+            skip_if_downloaded_already
+            and os.path.isfile(dest)
+            and os.path.getsize(dest) > 0
+        ):
+            logging.info(f"Skipping download as file already exists: {dest}")
+            return None
+        argv = [
+            ascp, "-QT", "-P", cls.ASPERA_PORT, "-l", maximum_bandwidth,
+            "-k", "2", source, dest,
+        ]
+        logging.info(
+            "Aspera: %s -> %s", source.replace(password, "***"), dest
+        )
+        try:
+            subprocess.run(argv, check=True, env=env)
+            return None
+        except subprocess.CalledProcessError as e:
+            logging.error(f"iProX Aspera failed for {url}: {e}")
+            return url
+
+    @classmethod
+    def aspera_download(
+        cls,
+        urls: List[str],
+        output_folder: str,
+        relative_paths: List[Optional[str]],
+        user: Optional[str],
+        password: Optional[str],
+        maximum_bandwidth: str = "100M",
+        skip_if_downloaded_already: bool = False,
+        parallel_files: int = 1,
+    ) -> None:
+        """Download iProX-hosted URLs via ascp on port 33001.
+
+        Requires iProX account credentials; the password is passed to the
+        subprocess through ASPERA_SCP_PASS (never argv). When
+        ``parallel_files`` > 1, transfers run concurrently: each ``ascp``
+        invocation is its own subprocess writing its own destination file, so
+        this is safe.
+        """
+        if not user or not password:
+            raise ValueError(
+                "iProX Aspera requires credentials: pass --iprox-user and set "
+                "IPROX_ASPERA_PASSWORD (or answer the password prompt), or use "
+                "the default parallel HTTP transport instead."
+            )
+        ascp = cls._ascp_binary()
+        env = dict(os.environ)
+        env["ASPERA_SCP_PASS"] = password
+        os.makedirs(output_folder, exist_ok=True)
+        failed: List[str] = []
+        workers = max(1, min(parallel_files, len(urls)))
+        if workers > 1:
+            with ThreadPoolExecutor(max_workers=workers) as executor:
+                future_to_url = {
+                    executor.submit(
+                        cls._aspera_download_one,
+                        ascp,
+                        url,
+                        relative_paths[idx] if idx < len(relative_paths) else None,
+                        output_folder,
+                        user,
+                        password,
+                        maximum_bandwidth,
+                        skip_if_downloaded_already,
+                        env,
+                    ): url
+                    for idx, url in enumerate(urls)
+                }
+                for future in as_completed(future_to_url):
+                    url = future_to_url[future]
+                    try:
+                        result = future.result()
+                        if result is not None:
+                            failed.append(result)
+                    except Exception as e:
+                        logging.error(f"iProX Aspera failed for {url}: {e}")
+                        failed.append(url)
+        else:
+            for idx, url in enumerate(urls):
+                relpath = relative_paths[idx] if idx < len(relative_paths) else None
+                result = cls._aspera_download_one(
+                    ascp,
+                    url,
+                    relpath,
+                    output_folder,
+                    user,
+                    password,
+                    maximum_bandwidth,
+                    skip_if_downloaded_already,
+                    env,
+                )
+                if result is not None:
+                    failed.append(result)
+        if failed:
+            raise RuntimeError(
+                f"iProX Aspera download failed for {len(failed)} file(s): {failed}"
+            )
+
     @staticmethod
     def _get_public_root(accession: str) -> str:
         return f"/{accession.upper()}"

pridepy/download/proteomexchange.py

@@ -26,10 +26,11 @@
 import posixpath
 import re
 import defusedxml.ElementTree as ET
-from typing import ClassVar, Dict, List
+from typing import ClassVar, Dict, List, Optional
 from urllib.parse import urlparse
 
 from pridepy.download.base import Provider
+from pridepy.download.util import flatten_relative_paths
 from pridepy.util.api_handling import Util
 
 
@@ -170,23 +171,79 @@ def download_from_accession_or_url(
         output_folder: str,
         skip_if_downloaded_already: bool = True,
         flatten: bool = True,
+        parallel_files: int = 1,
+        download_threads: int = 1,
+        protocol: str = "ftp",
+        iprox_user: Optional[str] = None,
+        iprox_password: Optional[str] = None,
     ) -> None:
         """End-to-end: resolve XML, list files, partition by scheme, download.
 
         Convenience for the ``download-px-raw-files`` CLI command — combines
         :meth:`list_files` and :meth:`download_files` with the original
         ``download_px_raw_files`` defaults (skip-if-downloaded-already
-        defaults to ``True``, no parallel workers).
+        defaults to ``True``). ``parallel_files`` controls across-file
+        concurrency and ``download_threads`` controls per-file HTTP Range
+        segments; ``protocol`` flows into :meth:`download_files` (ftp/http(s)
+        are handled directly today).
+
+        When ``protocol == "aspera"``, iProX-hosted files are routed through
+        :meth:`IproxProvider.aspera_download` instead of the HTTP/FTP path
+        (opt-in, requires ``iprox_user``/``iprox_password``).
         """
         records = self.list_files(px_id_or_url)
         if not records:
             logging.info("No Associated raw file URIs found in PX XML")
             return
+
+        if protocol.lower() == "aspera":
+            from pridepy.download.iprox import IproxProvider
+            iprox_urls, rels = [], []
+            for r in records:
+                loc = self.get_download_url(r, protocol)
+                host = (urlparse(loc).hostname or "").lower()
+                if host == "download.iprox.org":
+                    iprox_urls.append(loc)
+                    rels.append(r.get("relativePath"))
+            if not iprox_urls:
+                raise ValueError(
+                    "Aspera requested but no iProX-hosted files found in this dataset."
+                )
+            if len(iprox_urls) < len(records):
+                logging.warning(
+                    "%d of %d file(s) are NOT hosted on iProX and were NOT "
+                    "downloaded: --protocol aspera only handles iProX-hosted "
+                    "files. Use the default HTTP transport (omit --protocol, "
+                    "or pass --protocol ftp) to download the full set.",
+                    len(records) - len(iprox_urls),
+                    len(records),
+                )
+            if flatten:
+                sources = [
+                    rel if rel else urlparse(url).path
+                    for url, rel in zip(iprox_urls, rels)
+                ]
+                dest_rels: List[Optional[str]] = flatten_relative_paths(sources)
+            else:
+                dest_rels = rels
+            IproxProvider.aspera_download(
+                urls=iprox_urls,
+                output_folder=output_folder,
+                relative_paths=dest_rels,
+                user=iprox_user,
+                password=iprox_password,
+                skip_if_downloaded_already=skip_if_downloaded_already,
+                parallel_files=parallel_files,
+            )
+            return
+
         self.download_files(
             accession=px_id_or_url,
             records=records,
             output_folder=output_folder,
             skip_if_downloaded_already=skip_if_downloaded_already,
-            protocol="ftp",
+            protocol=protocol,
             flatten=flatten,
+            parallel_files=parallel_files,
+            download_threads=download_threads,
         )

pridepy/download/transport.py

@@ -19,6 +19,11 @@
 
 from pridepy.util.api_handling import Util
 
+# Combined cap on parallel_files (-w) x download_threads (-t): each factor is
+# independently clamped to 32, but nested they can reach 1024 concurrent HTTP
+# connections. Clamp the product to keep peak connections reasonable.
+MAX_TOTAL_HTTP_CONNECTIONS = 64
+
 
 def _safe_join(output_folder: str, relative_path: str) -> str:
     """Join ``output_folder`` with a dataset-relative path.
@@ -845,6 +850,19 @@ def _rel(idx: int) -> Optional[str]:
 
     failed: List[str] = []
     workers = max(1, min(parallel_files, len(http_urls)))
+    if workers * download_threads > MAX_TOTAL_HTTP_CONNECTIONS:
+        clamped_threads = max(1, MAX_TOTAL_HTTP_CONNECTIONS // workers)
+        logging.warning(
+            "parallel_files (%d) x download_threads (%d) = %d exceeds the "
+            "combined connection cap of %d; reducing download_threads to %d "
+            "to keep peak HTTP connections bounded.",
+            workers,
+            download_threads,
+            workers * download_threads,
+            MAX_TOTAL_HTTP_CONNECTIONS,
+            clamped_threads,
+        )
+        download_threads = clamped_threads
     if workers > 1:
         logging.info(
             f"Downloading {len(http_urls)} HTTP(S) file(s) with {workers} parallel workers"