Utilities

utils

RemoteArchive

RemoteArchive(url: str, archive_extension: str)

Source code in src/imgnet/utils/archive.py

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

    if archive_extension not in SUPPORTED_EXTENSIONS:
        msg = f"Invalid archive extension: {archive_extension}"
        raise ValueError(msg)
    self.archive_extension = archive_extension

archive_extension_from classmethod

archive_extension_from(path_or_url: str) -> str | None

Return the supported archive extension for a path/URL, or None.

Source code in src/imgnet/utils/archive.py

@classmethod
def archive_extension_from(cls, path_or_url: str) -> str | None:
    """Return the supported archive extension for a path/URL, or None."""
    name = Path(urlparse(path_or_url).path).name.lower()
    for ext in sorted(SUPPORTED_EXTENSIONS, key=len, reverse=True):
        if name.endswith(ext):
            return ext
    return None

require_extension classmethod

require_extension(path_or_url: str) -> str

Return the supported archive extension, or raise if unsupported.

Source code in src/imgnet/utils/archive.py

@classmethod
def require_extension(cls, path_or_url: str) -> str:
    """Return the supported archive extension, or raise if unsupported."""
    ext = cls.archive_extension_from(path_or_url)
    if ext is None:
        msg = f"Unsupported archive: {path_or_url}"
        raise ValueError(msg)
    return ext

archive_stem classmethod

archive_stem(path_or_url: str) -> str

Return the archive filename stem, handling compound suffixes like .tar.gz.

Source code in src/imgnet/utils/archive.py

@classmethod
def archive_stem(cls, path_or_url: str) -> str:
    """Return the archive filename stem, handling compound suffixes like .tar.gz."""
    name = Path(urlparse(path_or_url).path).name
    name_lower = name.lower()
    for ext in sorted(SUPPORTED_EXTENSIONS, key=len, reverse=True):
        if name_lower.endswith(ext):
            return name[: -len(ext)]
    return Path(name).stem

check_tar_filenames

check_tar_filenames(filenames: list[str]) -> list[str]

Check if the filenames roots are the same as the url filename stem. If they are, we can skip the extraction.

Source code in src/imgnet/utils/archive.py

def check_tar_filenames(self, filenames: list[str]) -> list[str]:
    """Check if the filenames roots are the same as the url filename stem.
    If they are, we can skip the extraction.
    """
    return [
        filename
        for filename in filenames
        if self._tar_check_root(filename)
    ]

get_idc_client

get_idc_client() -> idc_index.IDCClient

Return the shared IDC client, creating it on first use. Thread-safe.

Source code in src/imgnet/utils/idc.py

def get_idc_client() -> IDCClient:
    """Return the shared IDC client, creating it on first use. Thread-safe."""
    if _state["client"] is None:
        with _lock:
            if _state["client"] is None:
                _state["client"] = IDCClient()

    if _state["client"] is None:
        raise RuntimeError("Failed to create IDC client")
    return _state["client"]