bioimageio.spec.utils

668def resolve(
669    source: Union[PermissiveFileSource, FileDescr, ZipPath],
670    /,
671    progressbar: Union[Progressbar, bool, None] = None,
672    **kwargs: Unpack[HashKwargs],
673) -> Union[LocalFile, FileInZip]:
674    """Resolve file `source` (download if needed)"""
675
676    if isinstance(source, str):
677        source = interprete_file_source(source)
678
679    if isinstance(source, RelativeFilePath):
680        source = source.absolute()
681        if isinstance(source, ZipPath):
682            return FileInZip(source, source.root, extract_file_name(source))
683
684    if isinstance(source, pydantic.AnyUrl):
685        with get_validation_context().replace(perform_io_checks=False):
686            source = HttpUrl(source)
687
688    if isinstance(source, FileDescr):
689        return source.download()
690    elif isinstance(source, ZipPath):
691        zip_root = source.root
692        assert isinstance(zip_root, ZipFile)
693        return FileInZip(
694            source,
695            zip_root,
696            extract_file_name(source),
697        )
698    elif isinstance(source, Path):
699        if source.is_dir():
700            raise FileNotFoundError(f"{source} is a directory, not a file")
701
702        if not source.exists():
703            raise FileNotFoundError(source)
704        local_source = source
705        root: Union[RootHttpUrl, DirectoryPath] = source.parent
706    elif isinstance(source, HttpUrl):
707        if source.scheme not in ("http", "https"):
708            raise NotImplementedError(source.scheme)
709
710        if settings.CI:
711            headers = {"User-Agent": "ci"}
712            if progressbar is None:
713                progressbar = False
714        else:
715            headers = {}
716            if progressbar is None:
717                progressbar = True
718
719        if settings.user_agent is not None:
720            headers["User-Agent"] = settings.user_agent
721
722        downloader = pooch.HTTPDownloader(
723            headers=headers,
724            progressbar=progressbar,  # pyright: ignore[reportArgumentType]
725        )
726        fname = _get_unique_file_name(source)
727        _ls: Any = pooch.retrieve(
728            url=str(source),
729            known_hash=_get_known_hash(kwargs),
730            downloader=downloader,
731            fname=fname,
732            path=settings.cache_path,
733        )
734        local_source = Path(_ls).absolute()
735        root = source.parent
736    else:
737        assert_never(source)
738
739    return LocalFile(
740        local_source,
741        root,
742        extract_file_name(source),
743    )

263def ensure_description_is_dataset(
264    rd: Union[InvalidDescr, ResourceDescr],
265) -> AnyDatasetDescr:
266    if isinstance(rd, InvalidDescr):
267        rd.validation_summary.display()
268        raise ValueError(f"Invalid {rd.type} description.")
269
270    if rd.type != "dataset":
271        rd.validation_summary.display()
272        raise ValueError(
273            f"Expected a dataset resource, but got resource type '{rd.type}'"
274        )
275
276    assert not isinstance(
277        rd,
278        (
279            GenericDescr_v0_2,
280            GenericDescr_v0_3,
281        ),
282    )
283
284    return rd

235def ensure_description_is_model(
236    rd: Union[InvalidDescr, ResourceDescr],
237) -> AnyModelDescr:
238    """
239    Raises:
240        ValueError: for invalid or non-model resources
241    """
242    if isinstance(rd, InvalidDescr):
243        rd.validation_summary.display()
244        raise ValueError(f"Invalid {rd.type} description")
245
246    if rd.type != "model":
247        rd.validation_summary.display()
248        raise ValueError(
249            f"Expected a model resource, but got resource type '{rd.type}'"
250        )
251
252    assert not isinstance(
253        rd,
254        (
255            GenericDescr_v0_2,
256            GenericDescr_v0_3,
257        ),
258    )
259
260    return rd

828def extract_file_name(
829    src: Union[pydantic.HttpUrl, HttpUrl, PurePath, RelativeFilePath, ZipPath],
830) -> FileName:
831    if isinstance(src, ZipPath):
832        return src.name or src.root.filename or "bioimageio.zip"
833    elif isinstance(src, RelativeFilePath):
834        return src.path.name
835    elif isinstance(src, PurePath):
836        return src.name
837    else:
838        url = urlparse(str(src))
839        if (
840            url.scheme == "https"
841            and url.hostname == "zenodo.org"
842            and url.path.startswith("/api/records/")
843            and url.path.endswith("/content")
844        ):
845            return url.path.split("/")[-2]
846        else:
847            return url.path.split("/")[-1]

850def get_sha256(path: Union[Path, ZipPath]) -> Sha256:
851    """from https://stackoverflow.com/a/44873382"""
852    if isinstance(path, ZipPath):
853        # no buffered reading available
854        zf = path.root
855        assert isinstance(zf, ZipFile)
856        data = path.read_bytes()
857        assert isinstance(data, bytes)
858        h = hashlib.sha256(data)
859    else:
860        h = hashlib.sha256()
861        chunksize = 128 * 1024
862        b = bytearray(chunksize)
863        mv = memoryview(b)
864        with open(path, "rb", buffering=0) as f:
865            for n in iter(lambda: f.readinto(mv), 0):
866                h.update(mv[:n])
867
868    sha = h.hexdigest()
869    assert len(sha) == 64
870    return Sha256(sha)

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

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

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

284def load_array(source: Union[FileSource, FileDescr, ZipPath]) -> NDArray[Any]:
285    path = resolve(source).path
286    with path.open(mode="rb") as f:
287        assert not isinstance(f, io.TextIOWrapper)
288        return numpy.load(f, allow_pickle=False)

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

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

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

65def write_yaml(
66    content: Union[YamlValue, BioimageioYamlContent],
67    /,
68    file: Union[NewPath, FilePath, IO[str], IO[bytes], ZipPath],
69):
70    if isinstance(file, Path):
71        cm = file.open("w", encoding="utf-8")
72    else:
73        cm = nullcontext(file)
74
75    with cm as f:
76        _yaml_dump.dump(content, f)