Downloaders

downloaders

HuggingFaceDownloader

HuggingFaceDownloader(repo_id: str)

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, repo_id: str) -> None:
    self.repo_id = repo_id

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from Hugging Face. Here instance_ids is a list of filenames to download.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Download from Hugging Face. Here instance_ids is a list of filenames to download."""

    if instance_ids is None:
        logger.info(
            f"Downloading all instances from Hugging Face repository {self.repo_id}"
        )
        files_to_download = self.members
    else:
        logger.info(
            f"Downloading {len(instance_ids)} instances from Hugging Face repository {self.repo_id}"
        )
        remaining = set(instance_ids)
        files_to_download = []

        for file_name in self.members:
            if file_name in remaining:
                files_to_download.append(file_name)
                remaining.remove(file_name)
                continue

            if RemoteArchive.is_supported_archive(file_name) and remaining:
                archive_url = hf_hub_url(
                    repo_id=self.repo_id,
                    filename=file_name,
                    repo_type="dataset",
                )
                archive = RemoteArchive(
                    archive_url,
                    RemoteArchive.require_extension(file_name),
                )
                extracted = archive.extract(
                    filenames=sorted(remaining),
                    output_path=output_path,
                )
                remaining -= set(extracted)

        if remaining:
            msg = f"Instance IDs {sorted(remaining)} not found in Hugging Face repository {self.repo_id}"
            logger.warning(msg)

    if len(files_to_download) > 0:
        with tqdm_logging_redirect():                    
            snapshot_download(
                repo_id=self.repo_id,
                local_dir=output_path,
                tqdm_class=_tqdm,
                allow_patterns=files_to_download,
                repo_type="dataset",
                **kwargs,
            )

ZenodoDownloader

ZenodoDownloader(record_id: str)

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, record_id: str) -> None:
    self.record_id = record_id
    self.url = "https://zenodo.org/api/records"

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download files from Zenodo. Here instance_ids is a list of filenames to download.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Download files from Zenodo. Here instance_ids is a list of filenames to download."""
    output_path.mkdir(parents=True, exist_ok=True)
    files = self.files_info

    if instance_ids is None:
        files_to_download = files
        logger.info(
            f"Downloading all files from Zenodo record {self.record_id}"
        )
    else:
        logger.info(
            f"Downloading {len(instance_ids)} files from Zenodo record {self.record_id}"
        )
        remaining = set(instance_ids)
        files_to_download = []

        for file_info in files:
            file_name = file_info["key"]
            if file_name in remaining:
                files_to_download.append(file_info)
                remaining.remove(file_name)
                continue

            if RemoteArchive.is_supported_archive(file_name) and remaining:
                archive = RemoteArchive(
                    file_info["links"]["self"],
                    RemoteArchive.require_extension(file_name),
                )
                extracted = archive.extract(
                    filenames=sorted(remaining),
                    output_path=output_path,
                )
                remaining -= set(extracted)

        if remaining:
            msg = f"Instance IDs {sorted(remaining)} not found in Zenodo record {self.record_id}"
            logger.warning(msg)

    for file_info in files_to_download:
        _download_http_file(
            url=file_info["links"]["self"],
            out_file=output_path / file_info["key"],
            desc=file_info["key"],
            size=file_info["size"],
        )

DropboxDownloader

DropboxDownloader(url: str)

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, url: str) -> None:
    self.url = url.replace("dl=0", "dl=1").replace(
        "www.dropbox.com", "dl.dropboxusercontent.com"
    )

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from Dropbox. Supports selecting specific instance_ids from archives.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Download from Dropbox. Supports selecting specific instance_ids from archives."""
    output_path.mkdir(parents=True, exist_ok=True)
    file_name = Path(urlparse(self.url).path).name

    if instance_ids is None:
        logger.info(
            f"Downloading all files from Dropbox source {self.url}"
        )
        _download_http_file(
            url=self.url,
            out_file=output_path / file_name,
            desc=file_name,
        )
        return None

    remaining = set(instance_ids)
    logger.info(
        f"Downloading {len(remaining)} files from Dropbox source {self.url}"
    )
    if file_name in remaining:
        _download_http_file(
            url=self.url,
            out_file=output_path / file_name,
            desc=file_name,
        )
        remaining.remove(file_name)

    if RemoteArchive.is_supported_archive(file_name) and remaining:
        archive = RemoteArchive(
            self.url, RemoteArchive.require_extension(file_name)
        )
        extracted = archive.extract(
            filenames=sorted(remaining), output_path=output_path
        )
        remaining -= set(extracted)

    if remaining:
        msg = (
            f"Instance IDs {sorted(remaining)} not found in Dropbox source"
        )
        logger.warning(msg)

HttpDownloader

HttpDownloader(url: str)

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, url: str) -> None:
    self.url = url

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from a direct HTTP(S) URL. Supports selecting instance_ids from archives.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Download from a direct HTTP(S) URL. Supports selecting instance_ids from archives."""
    output_path.mkdir(parents=True, exist_ok=True)
    file_name = Path(urlparse(self.url).path).name

    if instance_ids is None:
        logger.info(f"Downloading all files from HTTP source {self.url}")
        _download_http_file(
            url=self.url,
            out_file=output_path / file_name,
            desc=file_name,
        )
        return None

    remaining = set(instance_ids)
    logger.info(
        f"Downloading {len(remaining)} files from HTTP source {self.url}"
    )
    if file_name in remaining:
        _download_http_file(
            url=self.url,
            out_file=output_path / file_name,
            desc=file_name,
        )
        remaining.remove(file_name)

    if RemoteArchive.is_supported_archive(file_name) and remaining:
        archive = RemoteArchive(
            self.url, RemoteArchive.require_extension(file_name)
        )
        extracted = archive.extract(
            filenames=sorted(remaining), output_path=output_path
        )
        remaining -= set(extracted)

    if remaining:
        msg = f"Instance IDs {sorted(remaining)} not found in HTTP source"
        logger.warning(msg)

IDCDownloader

IDCDownloader(collection_id: str)

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, collection_id: str) -> None:
    self.collection_id = (
        collection_id.lower().replace(" ", "_").replace("-", "_")
    )
    self.client = get_idc_client()

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from IDC. Here instance_ids is a list of series UIDs to download.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ANN401
) -> None:
    """Download from IDC. Here instance_ids is a list of series UIDs to download."""

    if instance_ids is not None:
        if not all(
            instance_id in self.members for instance_id in instance_ids
        ):
            msg = f"Instance IDs {instance_ids} not found in IDC collection {self.collection_id}"
            raise ValueError(msg)
        series_uids = instance_ids
    else:
        logger.warning(
            f"No instance IDs provided, downloading all series from IDC collection {self.collection_id}"
        )
        series_uids = self.members

    output_path.mkdir(parents=True, exist_ok=True)
    with tqdm_logging_redirect():
        self.client.download_dicom_series(
            series_uids,
            output_path,
            dirTemplate="%PatientID/%StudyInstanceUID/%Modality_%SeriesInstanceUID",
        )

GitHubDownloader

GitHubDownloader(repo_id: str)

Bases: imgnet.download.base.BaseDownloader

Initialize GitHub downloader.

Args: repo_id: GitHub repository in format "owner/repo"

Source code in src/imgnet/download/downloaders.py

def __init__(self, repo_id: str) -> None:
    """
    Initialize GitHub downloader.

    Args:
        repo_id: GitHub repository in format "owner/repo"
    """
    self.repo_id = repo_id
    self._api_base = "https://api.github.com/repos"
    self._raw_base = "https://raw.githubusercontent.com"
    self._cache = {}  # Simple cache for API responses

size property

size: float

Return the size of the repository in GB.

members property

members: list[str]

Return all file paths in the repository.

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from GitHub repository.

Source code in src/imgnet/download/downloaders.py

def download(
    self,
    output_path: Path,
    instance_ids: list[str] | None = None,
    **kwargs: Any,
) -> None:
    """Download from GitHub repository."""

    output_path.mkdir(parents=True, exist_ok=True)

    if instance_ids is None:
        logger.info(f"Downloading all files from GitHub repository {self.repo_id}")
        files_to_download = self.members
    else:
        logger.info(
            f"Downloading {len(instance_ids)} instances from GitHub repository {self.repo_id}"
        )
        remaining = set(instance_ids)
        files_to_download = []

        # Get all members first to check for archives
        all_members = self.members

        for file_path in all_members:
            if file_path in remaining:
                files_to_download.append(file_path)
                remaining.remove(file_path)
                continue

            # Check if this is an archive that might contain remaining files
            if RemoteArchive.is_supported_archive(file_path) and remaining:
                # Try to extract and see if it contains any of the remaining files
                archive_url = self._get_raw_url(file_path)
                archive = RemoteArchive(
                    archive_url,
                    RemoteArchive.require_extension(file_path),
                )

                # Create temp dir for extraction
                with tempfile.TemporaryDirectory() as temp_dir:
                    temp_path = Path(temp_dir)
                    extracted = archive.extract(
                        filenames=sorted(remaining),
                        output_path=temp_path,
                    )

                    # If we extracted something, move it to the actual output
                    if extracted:
                        for extracted_file in extracted:
                            src = temp_path / extracted_file
                            dst = output_path / extracted_file
                            dst.parent.mkdir(parents=True, exist_ok=True)
                            shutil.move(str(src), str(dst))
                        remaining -= set(extracted)

        if remaining:
            logger.warning(
                f"Instance IDs {sorted(remaining)} not found in GitHub repository {self.repo_id}"
            )

    # Download all collected files
    if files_to_download:
        self._download_files(files_to_download, output_path)

CompositeDownloader

CompositeDownloader(downloaders: list[imgnet.download.base.BaseDownloader])

Bases: imgnet.download.base.BaseDownloader

Source code in src/imgnet/download/downloaders.py

def __init__(self, downloaders: list[BaseDownloader]):
    self.downloaders = downloaders
    self._size = None
    self._members = None

download

download(output_path: pathlib.Path, instance_ids: list[str] | None = None, **kwargs: typing.Any) -> None

Download from all sources.

If instance_ids is provided, each downloader will only download files that match the requested IDs. IDs not found in any source are ignored.

Source code in src/imgnet/download/downloaders.py

def download(self,
        output_path: Path,
        instance_ids: list[str] | None = None,
        **kwargs: Any
    ) -> None:
    """
    Download from all sources.

    If instance_ids is provided, each downloader will only download files
    that match the requested IDs. IDs not found in any source are ignored.
    """
    output_path.mkdir(parents=True, exist_ok=True)
    if instance_ids is None:
        for downloader in self.downloaders:
            downloader.download(output_path, **kwargs)
    else:
        for downloader in self.downloaders:
            members = downloader.members
            ids = [id for id in instance_ids if id in members]
            if len(ids) > 0:
                downloader.download(output_path, instance_ids=ids, **kwargs)