bioimageio.spec.utils

626def get_reader(
627    source: Union[PermissiveFileSource, FileDescr, ZipPath],
628    /,
629    progressbar: Union[Progressbar, Callable[[], Progressbar], bool, None] = None,
630    **kwargs: Unpack[HashKwargs],
631) -> BytesReader:
632    """Open a file `source` (download if needed)"""
633    if isinstance(source, FileDescr):
634        if "sha256" not in kwargs:
635            kwargs["sha256"] = source.sha256
636
637        source = source.source
638    elif isinstance(source, str):
639        source = interprete_file_source(source)
640
641    if isinstance(source, RelativeFilePath):
642        source = source.absolute()
643    elif isinstance(source, pydantic.AnyUrl):
644        with get_validation_context().replace(perform_io_checks=False):
645            source = HttpUrl(source)
646
647    if isinstance(source, HttpUrl):
648        return _open_url(source, progressbar=progressbar, **kwargs)
649
650    if isinstance(source, ZipPath):
651        if not source.exists():
652            raise FileNotFoundError(source)
653
654        f = source.open(mode="rb")
655        assert not isinstance(f, TextIOWrapper)
656        root = source.root
657    elif isinstance(source, Path):
658        if source.is_dir():
659            raise FileNotFoundError(f"{source} is a directory, not a file")
660
661        if not source.exists():
662            raise FileNotFoundError(source)
663
664        f = source.open("rb")
665        root = source.parent
666    else:
667        assert_never(source)
668
669    expected_sha = kwargs.get("sha256")
670    if expected_sha is None:
671        sha = None
672    else:
673        sha = get_sha256(f)
674        _ = f.seek(0)
675        if sha != expected_sha:
676            raise ValueError(
677                f"SHA256 mismatch for {source}. Expected {expected_sha}, got {sha}."
678            )
679
680    return BytesReader(
681        f,
682        sha256=sha,
683        suffix=source.suffix,
684        original_file_name=source.name,
685        original_root=root,
686        is_zipfile=None,
687    )

277def ensure_description_is_dataset(
278    rd: Union[InvalidDescr, ResourceDescr],
279) -> AnyDatasetDescr:
280    if isinstance(rd, InvalidDescr):
281        rd.validation_summary.display()
282        raise ValueError(f"Invalid {rd.type} description.")
283
284    if rd.type != "dataset":
285        rd.validation_summary.display()
286        raise ValueError(
287            f"Expected a dataset resource, but got resource type '{rd.type}'"
288        )
289
290    assert not isinstance(
291        rd,
292        (
293            GenericDescr_v0_2,
294            GenericDescr_v0_3,
295        ),
296    )
297
298    return rd

249def ensure_description_is_model(
250    rd: Union[InvalidDescr, ResourceDescr],
251) -> AnyModelDescr:
252    """
253    Raises:
254        ValueError: for invalid or non-model resources
255    """
256    if isinstance(rd, InvalidDescr):
257        rd.validation_summary.display()
258        raise ValueError(f"Invalid {rd.type} description")
259
260    if rd.type != "model":
261        rd.validation_summary.display()
262        raise ValueError(
263            f"Expected a model resource, but got resource type '{rd.type}'"
264        )
265
266    assert not isinstance(
267        rd,
268        (
269            GenericDescr_v0_2,
270            GenericDescr_v0_3,
271        ),
272    )
273
274    return rd

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

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

626def get_reader(
627    source: Union[PermissiveFileSource, FileDescr, ZipPath],
628    /,
629    progressbar: Union[Progressbar, Callable[[], Progressbar], bool, None] = None,
630    **kwargs: Unpack[HashKwargs],
631) -> BytesReader:
632    """Open a file `source` (download if needed)"""
633    if isinstance(source, FileDescr):
634        if "sha256" not in kwargs:
635            kwargs["sha256"] = source.sha256
636
637        source = source.source
638    elif isinstance(source, str):
639        source = interprete_file_source(source)
640
641    if isinstance(source, RelativeFilePath):
642        source = source.absolute()
643    elif isinstance(source, pydantic.AnyUrl):
644        with get_validation_context().replace(perform_io_checks=False):
645            source = HttpUrl(source)
646
647    if isinstance(source, HttpUrl):
648        return _open_url(source, progressbar=progressbar, **kwargs)
649
650    if isinstance(source, ZipPath):
651        if not source.exists():
652            raise FileNotFoundError(source)
653
654        f = source.open(mode="rb")
655        assert not isinstance(f, TextIOWrapper)
656        root = source.root
657    elif isinstance(source, Path):
658        if source.is_dir():
659            raise FileNotFoundError(f"{source} is a directory, not a file")
660
661        if not source.exists():
662            raise FileNotFoundError(source)
663
664        f = source.open("rb")
665        root = source.parent
666    else:
667        assert_never(source)
668
669    expected_sha = kwargs.get("sha256")
670    if expected_sha is None:
671        sha = None
672    else:
673        sha = get_sha256(f)
674        _ = f.seek(0)
675        if sha != expected_sha:
676            raise ValueError(
677                f"SHA256 mismatch for {source}. Expected {expected_sha}, got {sha}."
678            )
679
680    return BytesReader(
681        f,
682        sha256=sha,
683        suffix=source.suffix,
684        original_file_name=source.name,
685        original_root=root,
686        is_zipfile=None,
687    )

150def get_sha256(source: Union[BytesReaderP, BytesReaderIntoP, Path]) -> Sha256:
151    chunksize = 128 * 1024
152    h = hashlib.sha256()
153
154    if isinstance(source, BytesReaderIntoP):
155        b = bytearray(chunksize)
156        mv = memoryview(b)
157        for n in iter(lambda: source.readinto(mv), 0):
158            h.update(mv[:n])
159    else:
160        if isinstance(source, Path):
161            read_ctxt = source.open(mode="rb")
162        else:
163            read_ctxt = nullcontext(source)
164
165        with read_ctxt as r:
166            for chunk in iter(partial(r.read, chunksize), b""):
167                h.update(chunk)
168
169    sha = h.hexdigest()
170    return Sha256(sha)

61def get_spdx_licenses() -> SpdxLicenses:
62    """get details of the SPDX licenses known to bioimageio.spec"""
63    with (
64        files("bioimageio.spec")
65        .joinpath("static/spdx_licenses.json")
66        .open("r", encoding="utf-8")
67    ) as f:
68        return json.load(f)

361def identify_bioimageio_yaml_file_name(file_names: Iterable[FileName]) -> FileName:
362    file_names = sorted(file_names)
363    for bioimageio_name in ALL_BIOIMAGEIO_YAML_NAMES:
364        for file_name in file_names:
365            if file_name == bioimageio_name or file_name.endswith(
366                "." + bioimageio_name
367            ):
368                return file_name
369
370    raise ValueError(
371        f"No {BIOIMAGEIO_YAML} found in {file_names}. (Looking for '{BIOIMAGEIO_YAML}'"
372        + " or or any of the alterntive file names:"
373        + f" {ALTERNATIVE_BIOIMAGEIO_YAML_NAMES}, or any file with an extension of"
374        + f"  those, e.g. 'anything.{BIOIMAGEIO_YAML}')."
375    )

353def is_valid_bioimageio_yaml_name(file_name: FileName) -> bool:
354    for bioimageio_name in ALL_BIOIMAGEIO_YAML_NAMES:
355        if file_name == bioimageio_name or file_name.endswith("." + bioimageio_name):
356            return True
357
358    return False

343def load_array(source: Union[FileSource, FileDescr, ZipPath]) -> NDArray[Any]:
344    reader = get_reader(source)
345    if settings.allow_pickle:
346        logger.warning("Loading numpy array with `allow_pickle=True`.")
347
348    return numpy.load(reader, allow_pickle=settings.allow_pickle)

142def open_bioimageio_yaml(
143    source: Union[PermissiveFileSource, ZipFile, ZipPath],
144    /,
145    **kwargs: Unpack[HashKwargs],
146) -> OpenedBioimageioYaml:
147    if isinstance(source, RelativeFilePath):
148        source = source.absolute()
149
150    if isinstance(source, ZipFile):
151        return _open_bioimageio_zip(source, original_source_name=str(source))
152    elif isinstance(source, ZipPath):
153        return _open_bioimageio_rdf_in_zip(
154            source, original_root=source.root, original_source_name=str(source)
155        )
156
157    try:
158        if isinstance(source, (Path, str)) and (source_dir := Path(source)).is_dir():
159            # open bioimageio yaml from a folder
160            src = source_dir / find_bioimageio_yaml_file_name(source_dir)
161        else:
162            src = interprete_file_source(source)
163
164        reader = get_reader(src, **kwargs)
165
166    except Exception:
167        # check if `source` is a collection id
168        if (
169            not isinstance(source, str)
170            or not isinstance(settings.id_map, str)
171            or "/" not in settings.id_map
172        ):
173            raise
174
175        if settings.collection_http_pattern:
176            with ValidationContext(perform_io_checks=False):
177                url = HttpUrl(
178                    settings.collection_http_pattern.format(bioimageio_id=source)
179                )
180
181            try:
182                r = httpx.get(url, follow_redirects=True)
183                _ = r.raise_for_status()
184                unparsed_content = r.content.decode(encoding="utf-8")
185                content = _sanitize_bioimageio_yaml(read_yaml(unparsed_content))
186            except Exception as e:
187                logger.warning("Failed to get bioimageio.yaml from {}: {}", url, e)
188            else:
189                original_file_name = (
190                    "rdf.yaml" if url.path is None else url.path.split("/")[-1]
191                )
192                return OpenedBioimageioYaml(
193                    content=content,
194                    original_root=url.parent,
195                    original_file_name=original_file_name,
196                    original_source_name=source,
197                    unparsed_content=unparsed_content,
198                )
199
200        id_map = get_id_map()
201        if id_map and source not in id_map:
202            close_matches = get_close_matches(source, id_map)
203            if len(close_matches) == 0:
204                raise
205
206            if len(close_matches) == 1:
207                did_you_mean = f" Did you mean '{close_matches[0]}'?"
208            else:
209                did_you_mean = f" Did you mean any of {close_matches}?"
210
211            raise FileNotFoundError(f"'{source}' not found.{did_you_mean}")
212
213        entry = id_map[source]
214        logger.info("loading {} from {}", source, entry.source)
215        reader = entry.get_reader()
216        with get_validation_context().replace(perform_io_checks=False):
217            src = HttpUrl(entry.source)
218
219    if reader.is_zipfile:
220        return _open_bioimageio_zip(ZipFile(reader), original_source_name=str(src))
221
222    unparsed_content = reader.read().decode(encoding="utf-8")
223    content = _sanitize_bioimageio_yaml(read_yaml(unparsed_content))
224
225    if isinstance(src, RelativeFilePath):
226        src = src.absolute()
227
228    if isinstance(src, ZipPath):
229        root = src.root
230    else:
231        root = src.parent
232
233    return OpenedBioimageioYaml(
234        content,
235        original_root=root,
236        original_source_name=str(src),
237        original_file_name=extract_file_name(src),
238        unparsed_content=unparsed_content,
239    )

61def read_yaml(
62    file: Union[FilePath, ZipPath, IO[str], IO[bytes], BytesReader, str],
63) -> YamlValue:
64    if isinstance(file, (ZipPath, Path)):
65        data = file.read_text(encoding="utf-8")
66    else:
67        data = file
68
69    content: YamlValue = _yaml_load.load(data)
70    return content

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

46class SpdxLicenseEntry(TypedDict):
47    isDeprecatedLicenseId: bool
48    isKnownByZenodo: bool
49    isOsiApproved: bool
50    licenseId: str
51    name: str
52    reference: str

55class SpdxLicenses(TypedDict):
56    licenseListVersion: str
57    licenses: List[SpdxLicenseEntry]
58    releaseDate: str

73def write_yaml(
74    content: Union[YamlValue, BioimageioYamlContentView, BaseModel],
75    /,
76    file: Union[NewPath, FilePath, IO[str], IO[bytes], ZipPath],
77):
78    if isinstance(file, Path):
79        cm = file.open("w", encoding="utf-8")
80    else:
81        cm = nullcontext(file)
82
83    if isinstance(content, BaseModel):
84        content = content.model_dump(mode="json")
85
86    with cm as f:
87        _yaml_dump.dump(content, f)