Coverage for src/bioimageio/spec/_internal/common_nodes.py: 87%
205 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 09:17 +0000
1from __future__ import annotations
3from abc import ABC
4from inspect import signature
5from io import BytesIO
6from pathlib import Path
7from types import MappingProxyType
8from typing import (
9 IO,
10 TYPE_CHECKING,
11 Any,
12 Callable,
13 ClassVar,
14 Iterable,
15 Literal,
16 Mapping,
17 Protocol,
18 TypeVar,
19)
20from zipfile import ZipFile
22import pydantic
23from pydantic import DirectoryPath, PrivateAttr, model_validator
24from pydantic_core import PydanticUndefined
25from typing_extensions import ParamSpec, Self
27from ..summary import (
28 WARNING_LEVEL_TO_NAME,
29 ErrorEntry,
30 ValidationDetail,
31 ValidationSummary,
32 WarningEntry,
33)
34from .field_warning import issue_warning
35from .io import (
36 BioimageioYamlContent,
37 FileDescr,
38 IncompleteDescr,
39 IncompleteDescrView,
40 deepcopy_incomplete_descr,
41 extract_file_descrs,
42 populate_cache,
43)
44from .io_basics import BIOIMAGEIO_YAML, FileName
45from .io_utils import write_content_to_zip
46from .node import Node
47from .packaging_context import PackagingContext
48from .root_url import RootHttpUrl
49from .type_guards import is_dict
50from .utils import get_format_version_tuple
51from .validation_context import ValidationContext, get_validation_context
52from .warning_levels import ALERT, ERROR, INFO
55class NodeWithExplicitlySetFields(Node):
56 _fields_to_set_explicitly: ClassVar[Mapping[str, Any]]
58 @classmethod
59 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
60 explict_fields: dict[str, Any] = {}
61 for attr in dir(cls):
62 if attr.startswith("implemented_"):
63 field_name = attr.replace("implemented_", "")
64 if field_name not in cls.model_fields:
65 continue
67 assert (
68 cls.model_fields[field_name].get_default() is PydanticUndefined
69 ), field_name
70 default = getattr(cls, attr)
71 explict_fields[field_name] = default
73 cls._fields_to_set_explicitly = MappingProxyType(explict_fields)
74 return super().__pydantic_init_subclass__(**kwargs)
76 @model_validator(mode="before")
77 @classmethod
78 def _set_fields_explicitly(cls, data: Any | dict[str, Any]) -> Any | dict[str, Any]:
79 if isinstance(data, dict):
80 for name, default in cls._fields_to_set_explicitly.items():
81 if name not in data:
82 data[name] = default
84 return data # pyright: ignore[reportUnknownVariableType]
87if TYPE_CHECKING:
89 class _ResourceDescrBaseAbstractFieldsProtocol(Protocol):
90 """workaround to add "abstract" fields to ResourceDescrBase"""
92 # TODO: implement as proper abstract fields of ResourceDescrBase
94 type: Any # should be LiteralString
95 format_version: Any # should be LiteralString
96 implemented_type: ClassVar[Any]
97 implemented_format_version: ClassVar[Any]
99else:
101 class _ResourceDescrBaseAbstractFieldsProtocol:
102 pass
105P = ParamSpec("P")
106T = TypeVar("T")
109class ResourceDescrBase(
110 NodeWithExplicitlySetFields, ABC, _ResourceDescrBaseAbstractFieldsProtocol
111):
112 """base class for all resource descriptions"""
114 _validation_summary: ValidationSummary | None = None
116 implemented_format_version_tuple: ClassVar[tuple[int, int, int]]
118 # @field_validator("format_version", mode="before", check_fields=False)
119 # field_validator on "format_version" is not possible, because we want to use
120 # "format_version" in a descriminated Union higher up
121 # (PydanticUserError: Cannot use a mode='before' validator in the discriminator
122 # field 'format_version' of Model 'CollectionDescr')
123 @model_validator(mode="before")
124 @classmethod
125 def _ignore_future_patch(cls, data: Any, /) -> Any:
126 if (
127 cls.implemented_format_version == "unknown"
128 or not is_dict(data)
129 or "format_version" not in data
130 ):
131 return data
133 value = data["format_version"]
134 fv = get_format_version_tuple(value)
135 if fv is None:
136 return data
137 if (
138 fv[0] == cls.implemented_format_version_tuple[0]
139 and fv[1:] > cls.implemented_format_version_tuple[1:]
140 ):
141 issue_warning(
142 "future format_version '{value}' treated as '{implemented}'",
143 value=value,
144 msg_context={"implemented": cls.implemented_format_version},
145 severity=ALERT,
146 )
147 data["format_version"] = cls.implemented_format_version
149 return data
151 @model_validator(mode="after")
152 def _set_init_validation_summary(self) -> Self:
153 context = get_validation_context()
155 self._validation_summary = ValidationSummary(
156 name="bioimageio format validation",
157 source_name=context.source_name,
158 id=getattr(self, "id", None),
159 version=getattr(self, "version", None),
160 type=self.type,
161 format_version=self.format_version,
162 status="failed" if isinstance(self, InvalidDescr) else "valid-format",
163 metadata_completeness=self._get_metadata_completeness(),
164 details=(
165 []
166 if isinstance(self, InvalidDescr)
167 else [
168 ValidationDetail(
169 name=f"Successfully created `{self.__class__.__name__}` instance.",
170 status="passed",
171 context=context.summary,
172 )
173 ]
174 ),
175 )
176 return self
178 @property
179 def validation_summary(self) -> ValidationSummary:
180 assert self._validation_summary is not None, "access only after initialization"
181 return self._validation_summary
183 _root: RootHttpUrl | DirectoryPath | ZipFile = PrivateAttr(
184 default_factory=lambda: get_validation_context().root
185 )
187 _file_name: FileName | None = PrivateAttr(
188 default_factory=lambda: get_validation_context().file_name
189 )
191 @property
192 def root(self) -> RootHttpUrl | DirectoryPath | ZipFile:
193 """The URL/Path prefix to resolve any relative paths with."""
194 return self._root
196 @property
197 def file_name(self) -> FileName | None:
198 """File name of the bioimageio.yaml file the description was loaded from."""
199 return self._file_name
201 @classmethod
202 def __pydantic_init_subclass__(cls, **kwargs: Any):
203 super().__pydantic_init_subclass__(**kwargs)
204 # set classvar implemented_format_version_tuple
205 if "format_version" in cls.model_fields:
206 if "." not in cls.implemented_format_version:
207 cls.implemented_format_version_tuple = (0, 0, 0)
208 else:
209 fv_tuple = get_format_version_tuple(cls.implemented_format_version)
210 assert fv_tuple is not None, (
211 f"failed to cast '{cls.implemented_format_version}' to tuple"
212 )
213 cls.implemented_format_version_tuple = fv_tuple
215 @classmethod
216 def load_from_kwargs(
217 cls: Callable[P, T],
218 context: ValidationContext | None = None,
219 *args: P.args,
220 **kwargs: P.kwargs,
221 ) -> T | InvalidDescr:
222 sig = signature(cls)
223 bound = sig.bind_partial(*args, **kwargs)
224 return cls.load(dict(bound.arguments), context=context) # pyright: ignore[reportFunctionMemberAccess]
226 @classmethod
227 def load(
228 cls,
229 data: IncompleteDescrView,
230 context: ValidationContext | None = None,
231 ) -> Self | InvalidDescr:
232 """factory method to create a resource description object"""
234 context = context or get_validation_context()
235 if context.perform_io_checks:
236 file_descrs = extract_file_descrs(data)
237 populate_cache(file_descrs) # TODO: add progress bar
239 with context.replace(log_warnings=context.warning_level <= INFO):
240 rd, errors, val_warnings = cls._load_impl(deepcopy_incomplete_descr(data))
242 if context.warning_level > INFO:
243 all_warnings_context = context.replace(
244 warning_level=INFO, log_warnings=False, raise_errors=False
245 )
246 # raise all validation warnings by reloading
247 with all_warnings_context:
248 _, _, val_warnings = cls._load_impl(deepcopy_incomplete_descr(data))
250 format_status = "failed" if errors else "passed"
251 rd.validation_summary.add_detail(
252 ValidationDetail(
253 errors=errors,
254 name=(
255 "bioimageio.spec format validation"
256 f" {rd.type} {cls.implemented_format_version}"
257 ),
258 status=format_status,
259 warnings=val_warnings,
260 ),
261 update_status=False, # this special validation detail needs manual format updating below
262 )
263 assert format_status != "failed" or isinstance(rd, InvalidDescr)
265 return rd
267 def _get_metadata_completeness(self) -> float:
268 if isinstance(self, InvalidDescr):
269 return 0.0
271 given = self.model_dump(mode="json", exclude_unset=True, exclude_defaults=False)
272 full = self.model_dump(mode="json", exclude_unset=False, exclude_defaults=False)
274 def extract_flat_keys(d: dict[Any, Any], key: str = "") -> Iterable[str]:
275 for k, v in d.items():
276 if is_dict(v):
277 yield from extract_flat_keys(v, key=f"{key}.{k}" if key else k)
279 yield f"{key}.{k}" if key else k
281 given_keys = set(extract_flat_keys(given))
282 full_keys = set(extract_flat_keys(full))
283 assert len(full_keys) >= len(given_keys)
284 return len(given_keys) / len(full_keys) if full_keys else 0.0
286 @classmethod
287 def _load_impl(
288 cls, data: IncompleteDescr
289 ) -> tuple[Self | InvalidDescr, list[ErrorEntry], list[WarningEntry]]:
290 rd: Self | InvalidDescr | None = None
291 val_errors: list[ErrorEntry] = []
292 val_warnings: list[WarningEntry] = []
294 context = get_validation_context()
295 try:
296 rd = cls.model_validate(data)
297 except pydantic.ValidationError as e:
298 for ee in e.errors(include_url=False):
299 if (severity := ee.get("ctx", {}).get("severity", ERROR)) < ERROR:
300 val_warnings.append(
301 WarningEntry(
302 loc=ee["loc"],
303 msg=ee["msg"],
304 type=ee["type"],
305 severity=severity,
306 )
307 )
308 elif context.raise_errors:
309 raise
310 else:
311 val_errors.append(
312 ErrorEntry(loc=ee["loc"], msg=ee["msg"], type=ee["type"])
313 )
315 if len(val_errors) == 0: # FIXME is this reduntant?
316 val_errors.append(
317 ErrorEntry(
318 loc=(),
319 msg=(
320 f"Encountered {len(val_warnings)} more severe than warning"
321 " level "
322 f"'{WARNING_LEVEL_TO_NAME[context.warning_level]}'"
323 ),
324 type="severe_warnings",
325 )
326 )
327 except Exception as e:
328 if context.raise_errors:
329 raise
331 try:
332 msg = str(e)
333 except Exception:
334 msg = e.__class__.__name__ + " encountered"
336 val_errors.append(
337 ErrorEntry(
338 loc=(),
339 msg=msg,
340 type=type(e).__name__,
341 with_traceback=True,
342 )
343 )
345 if rd is None:
346 try:
347 rd = InvalidDescr.model_validate(data)
348 except Exception:
349 if context.raise_errors:
350 raise
351 resource_type = cls.model_fields["type"].default
352 format_version = cls.implemented_format_version
353 rd = InvalidDescr(type=resource_type, format_version=format_version)
354 if context.raise_errors:
355 raise ValueError(rd)
357 return rd, val_errors, val_warnings
359 def package(
360 self,
361 dest: ZipFile | IO[bytes] | Path | str | None = None,
362 /,
363 local_files_only: bool = False,
364 ) -> ZipFile:
365 """package the described resource as a zip archive
367 Args:
368 dest: (path/bytes stream of) destination zipfile
369 """
370 if dest is None:
371 dest = BytesIO()
373 if isinstance(dest, ZipFile):
374 zip = dest
375 if "r" in zip.mode:
376 raise ValueError(
377 f"zip file {dest} opened in '{zip.mode}' mode,"
378 + " but write access is needed for packaging."
379 )
380 else:
381 zip = ZipFile(dest, mode="w")
383 if zip.filename is None:
384 zip.filename = (
385 str(getattr(self, "id", getattr(self, "name", "bioimageio"))) + ".zip"
386 )
388 content = self.get_package_content(local_files_only=local_files_only)
389 write_content_to_zip(content, zip)
390 return zip
392 def get_package_content(
393 self, local_files_only: bool = False
394 ) -> dict[FileName, FileDescr | BioimageioYamlContent]:
395 """Returns package content without creating the package."""
396 content: dict[FileName, FileDescr] = {}
397 with PackagingContext(
398 bioimageio_yaml_file_name=BIOIMAGEIO_YAML,
399 file_sources=content,
400 local_files_only=local_files_only,
401 ):
402 rdf_content: BioimageioYamlContent = self.model_dump(
403 mode="json", exclude_unset=True
404 )
406 _ = rdf_content.pop("rdf_source", None)
408 return {**content, BIOIMAGEIO_YAML: rdf_content}
411class InvalidDescr(
412 ResourceDescrBase,
413 extra="allow",
414 title="An invalid resource description",
415):
416 """A representation of an invalid resource description"""
418 implemented_type: ClassVar[Literal["unknown"]] = "unknown"
419 if TYPE_CHECKING: # see NodeWithExplicitlySetFields
420 type: Any = "unknown"
421 else:
422 type: Any
424 implemented_format_version: ClassVar[Literal["unknown"]] = "unknown"
425 if TYPE_CHECKING: # see NodeWithExplicitlySetFields
426 format_version: Any = "unknown"
427 else:
428 format_version: Any
430 def get_reason(self) -> str | None:
431 """Get the reason why the description is invalid, if available."""
432 reasons: list[str] = []
433 if self.validation_summary and self.validation_summary.details:
434 for detail in self.validation_summary.details:
435 if detail.status == "failed" and detail.errors:
436 reasons.extend(
437 f"{loc}: {msg}"
438 for loc, msg in (
439 (
440 ".".join(map(str, error.loc)),
441 error.msg.replace("\n", " "),
442 )
443 for error in detail.errors
444 )
445 )
447 return "\n- ".join(reasons) if reasons else None
450class KwargsNode(Node):
451 def get(self, item: str, default: Any = None) -> Any:
452 return self[item] if item in self else default # ruff: ignore[SIM401]
454 def __getitem__(self, item: str) -> Any:
455 if item in self.__class__.model_fields:
456 return getattr(self, item)
457 else:
458 raise KeyError(item)
460 def __contains__(self, item: str) -> bool:
461 return item in self.__class__.model_fields