bioimageio.spec.utils

658def resolve(
659    source: Union[PermissiveFileSource, FileDescr, ZipPath],
660    /,
661    progressbar: Union[Progressbar, bool, None] = None,
662    **kwargs: Unpack[HashKwargs],
663) -> Union[LocalFile, FileInZip]:
664    """Resolve file `source` (download if needed)"""
665    if isinstance(source, FileDescr):
666        return source.download()
667    elif isinstance(source, ZipPath):
668        zip_root = source.root
669        assert isinstance(zip_root, ZipFile)
670        return FileInZip(
671            source,
672            zip_root,
673            extract_file_name(source),
674        )
675
676    strict_source = interprete_file_source(source)
677    if isinstance(strict_source, RelativeFilePath):
678        strict_source = strict_source.absolute()
679        if isinstance(strict_source, ZipPath):
680            return FileInZip(
681                strict_source, strict_source.root, extract_file_name(strict_source)
682            )
683
684    if isinstance(strict_source, PurePath):
685        if not strict_source.exists():
686            raise FileNotFoundError(strict_source)
687        local_source = strict_source
688        root: Union[RootHttpUrl, DirectoryPath] = strict_source.parent
689    else:
690        if strict_source.scheme not in ("http", "https"):
691            raise NotImplementedError(strict_source.scheme)
692
693        if settings.CI:
694            headers = {"User-Agent": "ci"}
695            if progressbar is None:
696                progressbar = False
697        else:
698            headers = {}
699            if progressbar is None:
700                progressbar = True
701
702        if settings.user_agent is not None:
703            headers["User-Agent"] = settings.user_agent
704
705        downloader = pooch.HTTPDownloader(
706            headers=headers,
707            progressbar=progressbar,  # pyright: ignore[reportArgumentType]
708        )
709        fname = _get_unique_file_name(strict_source)
710        _ls: Any = pooch.retrieve(
711            url=str(strict_source),
712            known_hash=_get_known_hash(kwargs),
713            downloader=downloader,
714            fname=fname,
715            path=settings.cache_path,
716        )
717        local_source = Path(_ls).absolute()
718        root = strict_source.parent
719
720    return LocalFile(
721        local_source,
722        root,
723        extract_file_name(strict_source),
724    )

214def ensure_description_is_dataset(
215    rd: Union[InvalidDescr, ResourceDescr],
216) -> AnyDatasetDescr:
217    if isinstance(rd, InvalidDescr):
218        rd.validation_summary.display()
219        raise ValueError("resource description is invalid")
220
221    if rd.type != "dataset":
222        rd.validation_summary.display()
223        raise ValueError(
224            f"expected a dataset resource, but got resource type '{rd.type}'"
225        )
226
227    assert not isinstance(
228        rd,
229        (
230            GenericDescr_v0_2,
231            GenericDescr_v0_3,
232        ),
233    )
234
235    return rd

186def ensure_description_is_model(
187    rd: Union[InvalidDescr, ResourceDescr],
188) -> AnyModelDescr:
189    """
190    Raises:
191        ValueError: for invalid or non-model resources
192    """
193    if isinstance(rd, InvalidDescr):
194        rd.validation_summary.display()
195        raise ValueError("resource description is invalid")
196
197    if rd.type != "model":
198        rd.validation_summary.display()
199        raise ValueError(
200            f"expected a model resource, but got resource type '{rd.type}'"
201        )
202
203    assert not isinstance(
204        rd,
205        (
206            GenericDescr_v0_2,
207            GenericDescr_v0_3,
208        ),
209    )
210
211    return rd

803def extract_file_name(
804    src: Union[pydantic.HttpUrl, HttpUrl, PurePath, RelativeFilePath, ZipPath],
805) -> FileName:
806    if isinstance(src, ZipPath):
807        return src.name or src.root.filename or "bioimageio.zip"
808    elif isinstance(src, RelativeFilePath):
809        return src.path.name
810    elif isinstance(src, PurePath):
811        return src.name
812    else:
813        url = urlparse(str(src))
814        if (
815            url.scheme == "https"
816            and url.hostname == "zenodo.org"
817            and url.path.startswith("/api/records/")
818            and url.path.endswith("/content")
819        ):
820            return url.path.split("/")[-2]
821        else:
822            return url.path.split("/")[-1]

825@lru_cache
826def get_sha256(path: Union[Path, ZipPath]) -> Sha256:
827    """from https://stackoverflow.com/a/44873382"""
828    desc = f"computing SHA256 of {path.name}"
829    if isinstance(path, ZipPath):
830        # no buffered reading available
831        zf = path.root
832        assert isinstance(zf, ZipFile)
833        file_size = zf.NameToInfo[path.at].file_size
834        pbar = tqdm(desc=desc, total=file_size)
835        data = path.read_bytes()
836        assert isinstance(data, bytes)
837        h = hashlib.sha256(data)
838    else:
839        file_size = path.stat().st_size
840        pbar = tqdm(desc=desc, total=file_size)
841        h = hashlib.sha256()
842        chunksize = 128 * 1024
843        b = bytearray(chunksize)
844        mv = memoryview(b)
845        with open(path, "rb", buffering=0) as f:
846            for n in iter(lambda: f.readinto(mv), 0):
847                h.update(mv[:n])
848                _ = pbar.update(n)
849
850    sha = h.hexdigest()
851    pbar.set_description(desc=desc + f" (result: {sha})")
852    pbar.close()
853    assert len(sha) == 64
854    return Sha256(sha)

47def get_spdx_licenses() -> SpdxLicenses:
48    """get details of the SPDX licenses known to bioimageio.spec"""
49    with files("bioimageio.spec").joinpath("static/spdx_licenses.json").open(
50        "r", encoding="utf-8"
51    ) as f:
52        return json.load(f)

429def identify_bioimageio_yaml_file_name(file_names: Iterable[FileName]) -> FileName:
430    file_names = sorted(file_names)
431    for bioimageio_name in ALL_BIOIMAGEIO_YAML_NAMES:
432        for file_name in file_names:
433            if file_name == bioimageio_name or file_name.endswith(
434                "." + bioimageio_name
435            ):
436                return file_name
437
438    raise ValueError(
439        f"No {BIOIMAGEIO_YAML} found in {file_names}. (Looking for '{BIOIMAGEIO_YAML}'"
440        + " or or any of the alterntive file names:"
441        + f" {ALTERNATIVE_BIOIMAGEIO_YAML_NAMES}, or any file with an extension of"
442        + f"  those, e.g. 'anything.{BIOIMAGEIO_YAML}')."
443    )

421def is_valid_bioimageio_yaml_name(file_name: FileName) -> bool:
422    for bioimageio_name in ALL_BIOIMAGEIO_YAML_NAMES:
423        if file_name == bioimageio_name or file_name.endswith("." + bioimageio_name):
424            return True
425
426    return False

235def load_array(source: Union[FileSource, FileDescr, ZipPath]) -> NDArray[Any]:
236    path = download(source).path
237    with path.open(mode="rb") as f:
238        assert not isinstance(f, io.TextIOWrapper)
239        return numpy.load(f, allow_pickle=False)

242def save_array(path: Union[Path, ZipPath], array: NDArray[Any]) -> None:
243    with path.open(mode="wb") as f:
244        assert not isinstance(f, io.TextIOWrapper)
245        return numpy.save(f, array, allow_pickle=False)

32class SpdxLicenseEntry(TypedDict):
33    isDeprecatedLicenseId: bool
34    isKnownByZenodo: bool
35    isOsiApproved: bool
36    licenseId: str
37    name: str
38    reference: str

41class SpdxLicenses(TypedDict):
42    licenseListVersion: str
43    licenses: List[SpdxLicenseEntry]
44    releaseDate: str