Coverage for src / bioimageio / spec / _internal / common_nodes.py: 87%

1from __future__ import annotations 

2 

3from abc import ABC 

4from io import BytesIO 

5from pathlib import Path 

6from types import MappingProxyType 

7from typing import ( 

8 IO, 

9 TYPE_CHECKING, 

10 Any, 

11 ClassVar, 

12 Dict, 

13 Iterable, 

14 List, 

15 Literal, 

16 Mapping, 

17 Optional, 

18 Protocol, 

19 Tuple, 

20 Union, 

21) 

22from zipfile import ZipFile 

23 

24import pydantic 

25from pydantic import DirectoryPath, PrivateAttr, model_validator 

26from pydantic_core import PydanticUndefined 

27from typing_extensions import Self 

28 

29from ..summary import ( 

30 WARNING_LEVEL_TO_NAME, 

31 ErrorEntry, 

32 ValidationDetail, 

33 ValidationSummary, 

34 WarningEntry, 

35) 

36from .field_warning import issue_warning 

37from .io import ( 

38 BioimageioYamlContent, 

39 BioimageioYamlContentView, 

40 FileDescr, 

41 deepcopy_yaml_value, 

42 extract_file_descrs, 

43 populate_cache, 

44) 

45from .io_basics import BIOIMAGEIO_YAML, FileName 

46from .io_utils import write_content_to_zip 

47from .node import Node 

48from .packaging_context import PackagingContext 

49from .root_url import RootHttpUrl 

50from .type_guards import is_dict 

51from .utils import get_format_version_tuple 

52from .validation_context import ValidationContext, get_validation_context 

53from .warning_levels import ALERT, ERROR, INFO 

54 

55 

56class NodeWithExplicitlySetFields(Node): 

57 _fields_to_set_explicitly: ClassVar[Mapping[str, Any]] 

58 

59 @classmethod 

60 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None: 

61 explict_fields: Dict[str, Any] = {} 

62 for attr in dir(cls): 

63 if attr.startswith("implemented_"): 

64 field_name = attr.replace("implemented_", "") 

65 if field_name not in cls.model_fields: 

66 continue 

67 

68 assert ( 

69 cls.model_fields[field_name].get_default() is PydanticUndefined 

70 ), field_name 

71 default = getattr(cls, attr) 

72 explict_fields[field_name] = default 

73 

74 cls._fields_to_set_explicitly = MappingProxyType(explict_fields) 

75 return super().__pydantic_init_subclass__(**kwargs) 

76 

77 @model_validator(mode="before") 

78 @classmethod 

79 def _set_fields_explicitly( 

80 cls, data: Union[Any, Dict[str, Any]] 

81 ) -> Union[Any, Dict[str, Any]]: 

82 if isinstance(data, dict): 

83 for name, default in cls._fields_to_set_explicitly.items(): 

84 if name not in data: 

85 data[name] = default 

86 

87 return data # pyright: ignore[reportUnknownVariableType] 

88 

89 

90if TYPE_CHECKING: 

91 

92 class _ResourceDescrBaseAbstractFieldsProtocol(Protocol): 

93 """workaround to add "abstract" fields to ResourceDescrBase""" 

94 

95 # TODO: implement as proper abstract fields of ResourceDescrBase 

96 

97 type: Any # should be LiteralString 

98 format_version: Any # should be LiteralString 

99 implemented_type: ClassVar[Any] 

100 implemented_format_version: ClassVar[Any] 

101 

102else: 

103 

104 class _ResourceDescrBaseAbstractFieldsProtocol: 

105 pass 

106 

107 

108class ResourceDescrBase( 

109 NodeWithExplicitlySetFields, ABC, _ResourceDescrBaseAbstractFieldsProtocol 

110): 

111 """base class for all resource descriptions""" 

112 

113 _validation_summary: Optional[ValidationSummary] = None 

114 

115 implemented_format_version_tuple: ClassVar[Tuple[int, int, int]] 

116 

117 # @field_validator("format_version", mode="before", check_fields=False) 

118 # field_validator on "format_version" is not possible, because we want to use 

119 # "format_version" in a descriminated Union higher up 

120 # (PydanticUserError: Cannot use a mode='before' validator in the discriminator 

121 # field 'format_version' of Model 'CollectionDescr') 

122 @model_validator(mode="before") 

123 @classmethod 

124 def _ignore_future_patch(cls, data: Any, /) -> Any: 

125 if ( 

126 cls.implemented_format_version == "unknown" 

127 or not is_dict(data) 

128 or "format_version" not in data 

129 ): 

130 return data 

131 

132 value = data["format_version"] 

133 fv = get_format_version_tuple(value) 

134 if fv is None: 

135 return data 

136 if ( 

137 fv[0] == cls.implemented_format_version_tuple[0] 

138 and fv[1:] > cls.implemented_format_version_tuple[1:] 

139 ): 

140 issue_warning( 

141 "future format_version '{value}' treated as '{implemented}'", 

142 value=value, 

143 msg_context=dict(implemented=cls.implemented_format_version), 

144 severity=ALERT, 

145 ) 

146 data["format_version"] = cls.implemented_format_version 

147 

148 return data 

149 

150 @model_validator(mode="after") 

151 def _set_init_validation_summary(self) -> Self: 

152 context = get_validation_context() 

153 

154 self._validation_summary = ValidationSummary( 

155 name="bioimageio format validation", 

156 source_name=context.source_name, 

157 id=getattr(self, "id", None), 

158 version=getattr(self, "version", None), 

159 type=self.type, 

160 format_version=self.format_version, 

161 status="failed" if isinstance(self, InvalidDescr) else "valid-format", 

162 metadata_completeness=self._get_metadata_completeness(), 

163 details=( 

164 [] 

165 if isinstance(self, InvalidDescr) 

166 else [ 

167 ValidationDetail( 

168 name=f"Successfully created `{self.__class__.__name__}` instance.", 

169 status="passed", 

170 context=context.summary, 

171 ) 

172 ] 

173 ), 

174 ) 

175 return self 

176 

177 @property 

178 def validation_summary(self) -> ValidationSummary: 

179 assert self._validation_summary is not None, "access only after initialization" 

180 return self._validation_summary 

181 

182 _root: Union[RootHttpUrl, DirectoryPath, ZipFile] = PrivateAttr( 

183 default_factory=lambda: get_validation_context().root 

184 ) 

185 

186 _file_name: Optional[FileName] = PrivateAttr( 

187 default_factory=lambda: get_validation_context().file_name 

188 ) 

189 

190 @property 

191 def root(self) -> Union[RootHttpUrl, DirectoryPath, ZipFile]: 

192 """The URL/Path prefix to resolve any relative paths with.""" 

193 return self._root 

194 

195 @property 

196 def file_name(self) -> Optional[FileName]: 

197 """File name of the bioimageio.yaml file the description was loaded from.""" 

198 return self._file_name 

199 

200 @classmethod 

201 def __pydantic_init_subclass__(cls, **kwargs: Any): 

202 super().__pydantic_init_subclass__(**kwargs) 

203 # set classvar implemented_format_version_tuple 

204 if "format_version" in cls.model_fields: 

205 if "." not in cls.implemented_format_version: 

206 cls.implemented_format_version_tuple = (0, 0, 0) 

207 else: 

208 fv_tuple = get_format_version_tuple(cls.implemented_format_version) 

209 assert fv_tuple is not None, ( 

210 f"failed to cast '{cls.implemented_format_version}' to tuple" 

211 ) 

212 cls.implemented_format_version_tuple = fv_tuple 

213 

214 @classmethod 

215 def load( 

216 cls, 

217 data: BioimageioYamlContentView, 

218 context: Optional[ValidationContext] = None, 

219 ) -> Union[Self, InvalidDescr]: 

220 """factory method to create a resource description object""" 

221 context = context or get_validation_context() 

222 if context.perform_io_checks: 

223 file_descrs = extract_file_descrs({k: v for k, v in data.items()}) 

224 populate_cache(file_descrs) # TODO: add progress bar 

225 

226 with context.replace(log_warnings=context.warning_level <= INFO): 

227 rd, errors, val_warnings = cls._load_impl(deepcopy_yaml_value(data)) 

228 

229 if context.warning_level > INFO: 

230 all_warnings_context = context.replace( 

231 warning_level=INFO, log_warnings=False, raise_errors=False 

232 ) 

233 # raise all validation warnings by reloading 

234 with all_warnings_context: 

235 _, _, val_warnings = cls._load_impl(deepcopy_yaml_value(data)) 

236 

237 format_status = "failed" if errors else "passed" 

238 rd.validation_summary.add_detail( 

239 ValidationDetail( 

240 errors=errors, 

241 name=( 

242 "bioimageio.spec format validation" 

243 f" {rd.type} {cls.implemented_format_version}" 

244 ), 

245 status=format_status, 

246 warnings=val_warnings, 

247 ), 

248 update_status=False, # avoid updating status from 'valid-format' to 'passed', but ... 

249 ) 

250 if format_status == "failed": 

251 # ... update status in case of failure 

252 rd.validation_summary.status = "failed" 

253 

254 return rd 

255 

256 def _get_metadata_completeness(self) -> float: 

257 if isinstance(self, InvalidDescr): 

258 return 0.0 

259 

260 given = self.model_dump(mode="json", exclude_unset=True, exclude_defaults=False) 

261 full = self.model_dump(mode="json", exclude_unset=False, exclude_defaults=False) 

262 

263 def extract_flat_keys(d: Dict[Any, Any], key: str = "") -> Iterable[str]: 

264 for k, v in d.items(): 

265 if is_dict(v): 

266 yield from extract_flat_keys(v, key=f"{key}.{k}" if key else k) 

267 

268 yield f"{key}.{k}" if key else k 

269 

270 given_keys = set(extract_flat_keys(given)) 

271 full_keys = set(extract_flat_keys(full)) 

272 assert len(full_keys) >= len(given_keys) 

273 return len(given_keys) / len(full_keys) if full_keys else 0.0 

274 

275 @classmethod 

276 def _load_impl( 

277 cls, data: BioimageioYamlContent 

278 ) -> Tuple[Union[Self, InvalidDescr], List[ErrorEntry], List[WarningEntry]]: 

279 rd: Union[Self, InvalidDescr, None] = None 

280 val_errors: List[ErrorEntry] = [] 

281 val_warnings: List[WarningEntry] = [] 

282 

283 context = get_validation_context() 

284 try: 

285 rd = cls.model_validate(data) 

286 except pydantic.ValidationError as e: 

287 for ee in e.errors(include_url=False): 

288 if (severity := ee.get("ctx", {}).get("severity", ERROR)) < ERROR: 

289 val_warnings.append( 

290 WarningEntry( 

291 loc=ee["loc"], 

292 msg=ee["msg"], 

293 type=ee["type"], 

294 severity=severity, 

295 ) 

296 ) 

297 elif context.raise_errors: 

298 raise e 

299 else: 

300 val_errors.append( 

301 ErrorEntry(loc=ee["loc"], msg=ee["msg"], type=ee["type"]) 

302 ) 

303 

304 if len(val_errors) == 0: # FIXME is this reduntant? 

305 val_errors.append( 

306 ErrorEntry( 

307 loc=(), 

308 msg=( 

309 f"Encountered {len(val_warnings)} more severe than warning" 

310 " level " 

311 f"'{WARNING_LEVEL_TO_NAME[context.warning_level]}'" 

312 ), 

313 type="severe_warnings", 

314 ) 

315 ) 

316 except Exception as e: 

317 if context.raise_errors: 

318 raise e 

319 

320 try: 

321 msg = str(e) 

322 except Exception: 

323 msg = e.__class__.__name__ + " encountered" 

324 

325 val_errors.append( 

326 ErrorEntry( 

327 loc=(), 

328 msg=msg, 

329 type=type(e).__name__, 

330 with_traceback=True, 

331 ) 

332 ) 

333 

334 if rd is None: 

335 try: 

336 rd = InvalidDescr.model_validate(data) 

337 except Exception as e: 

338 if context.raise_errors: 

339 raise e 

340 resource_type = cls.model_fields["type"].default 

341 format_version = cls.implemented_format_version 

342 rd = InvalidDescr(type=resource_type, format_version=format_version) 

343 if context.raise_errors: 

344 raise ValueError(rd) 

345 

346 return rd, val_errors, val_warnings 

347 

348 def package( 

349 self, dest: Optional[Union[ZipFile, IO[bytes], Path, str]] = None, / 

350 ) -> ZipFile: 

351 """package the described resource as a zip archive 

352 

353 Args: 

354 dest: (path/bytes stream of) destination zipfile 

355 """ 

356 if dest is None: 

357 dest = BytesIO() 

358 

359 if isinstance(dest, ZipFile): 

360 zip = dest 

361 if "r" in zip.mode: 

362 raise ValueError( 

363 f"zip file {dest} opened in '{zip.mode}' mode," 

364 + " but write access is needed for packaging." 

365 ) 

366 else: 

367 zip = ZipFile(dest, mode="w") 

368 

369 if zip.filename is None: 

370 zip.filename = ( 

371 str(getattr(self, "id", getattr(self, "name", "bioimageio"))) + ".zip" 

372 ) 

373 

374 content = self.get_package_content() 

375 write_content_to_zip(content, zip) 

376 return zip 

377 

378 def get_package_content( 

379 self, 

380 ) -> Dict[FileName, Union[FileDescr, BioimageioYamlContent]]: 

381 """Returns package content without creating the package.""" 

382 content: Dict[FileName, FileDescr] = {} 

383 with PackagingContext( 

384 bioimageio_yaml_file_name=BIOIMAGEIO_YAML, 

385 file_sources=content, 

386 ): 

387 rdf_content: BioimageioYamlContent = self.model_dump( 

388 mode="json", exclude_unset=True 

389 ) 

390 

391 _ = rdf_content.pop("rdf_source", None) 

392 

393 return {**content, BIOIMAGEIO_YAML: rdf_content} 

394 

395 

396class InvalidDescr( 

397 ResourceDescrBase, 

398 extra="allow", 

399 title="An invalid resource description", 

400): 

401 """A representation of an invalid resource description""" 

402 

403 implemented_type: ClassVar[Literal["unknown"]] = "unknown" 

404 if TYPE_CHECKING: # see NodeWithExplicitlySetFields 

405 type: Any = "unknown" 

406 else: 

407 type: Any 

408 

409 implemented_format_version: ClassVar[Literal["unknown"]] = "unknown" 

410 if TYPE_CHECKING: # see NodeWithExplicitlySetFields 

411 format_version: Any = "unknown" 

412 else: 

413 format_version: Any 

414 

415 

416class KwargsNode(Node): 

417 def get(self, item: str, default: Any = None) -> Any: 

418 return self[item] if item in self else default 

419 

420 def __getitem__(self, item: str) -> Any: 

421 if item in self.__class__.model_fields: 

422 return getattr(self, item) 

423 else: 

424 raise KeyError(item) 

425 

426 def __contains__(self, item: str) -> bool: 

427 return item in self.__class__.model_fields