Coverage for src / bioimageio / spec / _io.py: 72%

1import collections.abc 

2from pathlib import Path 

3from typing import Dict, Literal, Optional, TextIO, Union, cast, overload 

4from zipfile import ZipFile 

5 

6from loguru import logger 

7from pydantic import FilePath, NewPath 

8 

9from ._description import ( 

10 DISCOVER, 

11 LATEST, 

12 InvalidDescr, 

13 LatestResourceDescr, 

14 ResourceDescr, 

15 build_description, 

16 dump_description, 

17 ensure_description_is_dataset, 

18 ensure_description_is_model, 

19) 

20from ._internal.common_nodes import ResourceDescrBase 

21from ._internal.io import BioimageioYamlContent, YamlValue 

22from ._internal.io_basics import Sha256 

23from ._internal.io_utils import open_bioimageio_yaml, write_yaml 

24from ._internal.types import FormatVersionPlaceholder, PermissiveFileSource 

25from ._internal.validation_context import get_validation_context 

26from .dataset import AnyDatasetDescr, DatasetDescr 

27from .model import AnyModelDescr, ModelDescr 

28from .summary import ValidationSummary 

29 

30 

31@overload 

32def load_description( 

33 source: Union[PermissiveFileSource, ZipFile], 

34 /, 

35 *, 

36 format_version: Literal["latest"], 

37 perform_io_checks: Optional[bool] = None, 

38 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

39 sha256: Optional[Sha256] = None, 

40) -> Union[LatestResourceDescr, InvalidDescr]: ... 

41 

42 

43@overload 

44def load_description( 

45 source: Union[PermissiveFileSource, ZipFile], 

46 /, 

47 *, 

48 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

49 perform_io_checks: Optional[bool] = None, 

50 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

51 sha256: Optional[Sha256] = None, 

52) -> Union[ResourceDescr, InvalidDescr]: ... 

53 

54 

55def load_description( 

56 source: Union[PermissiveFileSource, ZipFile], 

57 /, 

58 *, 

59 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

60 perform_io_checks: Optional[bool] = None, 

61 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

62 sha256: Optional[Sha256] = None, 

63) -> Union[ResourceDescr, InvalidDescr]: 

64 """load a bioimage.io resource description 

65 

66 Args: 

67 source: 

68 Path or URL to an rdf.yaml or a bioimage.io package 

69 (zip-file with rdf.yaml in it). 

70 format_version: 

71 (optional) Use this argument to load the resource and 

72 convert its metadata to a higher format_version. 

73 Note: 

74 - Use "latest" to convert to the latest available format version. 

75 - Use "discover" to use the format version specified in the RDF. 

76 - Only considers major.minor format version, ignores patch version. 

77 - Conversion to lower format versions is not supported. 

78 perform_io_checks: 

79 Wether or not to perform validation that requires file io, 

80 e.g. downloading a remote files. The existence of local 

81 absolute file paths is still being checked. 

82 known_files: 

83 Allows to bypass download and hashing of referenced files 

84 (even if perform_io_checks is True). 

85 

86 Keys should be file paths or URL strings as they appear in the 

87 bioimageio.yaml file. 

88 

89 Values are Sha256 values compared to hash values in the description. 

90 For `None` values no hash value comparison is performed. 

91 

92 If `perfrom_io_checks` is True, checked files will be added to 

93 this dictionary with their SHA-256 value. 

94 

95 If `perform_io_checks` is False and `known_files` is not empty, 

96 missing, 'unknown' file references are considered invalid. 

97 sha256: 

98 Optional SHA-256 value of **source** 

99 

100 Returns: 

101 An object holding all metadata of the bioimage.io resource 

102 

103 """ 

104 if isinstance(source, ResourceDescrBase): 

105 name = getattr(source, "name", f"{str(source)[:10]}...") 

106 logger.warning("returning already loaded description '{}' as is", name) 

107 return source # pyright: ignore[reportReturnType] 

108 

109 opened = open_bioimageio_yaml(source, sha256=sha256) 

110 

111 context = get_validation_context().replace( 

112 root=opened.original_root, 

113 file_name=opened.original_file_name, 

114 original_source_name=opened.original_source_name, 

115 perform_io_checks=perform_io_checks, 

116 known_files=known_files, 

117 ) 

118 

119 return build_description( 

120 opened.content, 

121 context=context, 

122 format_version=format_version, 

123 ) 

124 

125 

126@overload 

127def load_model_description( 

128 source: Union[PermissiveFileSource, ZipFile], 

129 /, 

130 *, 

131 format_version: Literal["latest"], 

132 perform_io_checks: Optional[bool] = None, 

133 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

134 sha256: Optional[Sha256] = None, 

135) -> ModelDescr: ... 

136 

137 

138@overload 

139def load_model_description( 

140 source: Union[PermissiveFileSource, ZipFile], 

141 /, 

142 *, 

143 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

144 perform_io_checks: Optional[bool] = None, 

145 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

146 sha256: Optional[Sha256] = None, 

147) -> AnyModelDescr: ... 

148 

149 

150def load_model_description( 

151 source: Union[PermissiveFileSource, ZipFile], 

152 /, 

153 *, 

154 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

155 perform_io_checks: Optional[bool] = None, 

156 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

157 sha256: Optional[Sha256] = None, 

158) -> AnyModelDescr: 

159 """same as `load_description`, but addtionally ensures that the loaded 

160 description is valid and of type 'model'. 

161 

162 Raises: 

163 ValueError: for invalid or non-model resources 

164 """ 

165 rd = load_description( 

166 source, 

167 format_version=format_version, 

168 perform_io_checks=perform_io_checks, 

169 known_files=known_files, 

170 sha256=sha256, 

171 ) 

172 return ensure_description_is_model(rd) 

173 

174 

175@overload 

176def load_dataset_description( 

177 source: Union[PermissiveFileSource, ZipFile], 

178 /, 

179 *, 

180 format_version: Literal["latest"], 

181 perform_io_checks: Optional[bool] = None, 

182 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

183 sha256: Optional[Sha256] = None, 

184) -> DatasetDescr: ... 

185 

186 

187@overload 

188def load_dataset_description( 

189 source: Union[PermissiveFileSource, ZipFile], 

190 /, 

191 *, 

192 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

193 perform_io_checks: Optional[bool] = None, 

194 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

195 sha256: Optional[Sha256] = None, 

196) -> AnyDatasetDescr: ... 

197 

198 

199def load_dataset_description( 

200 source: Union[PermissiveFileSource, ZipFile], 

201 /, 

202 *, 

203 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

204 perform_io_checks: Optional[bool] = None, 

205 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

206 sha256: Optional[Sha256] = None, 

207) -> AnyDatasetDescr: 

208 """same as `load_description`, but addtionally ensures that the loaded 

209 description is valid and of type 'dataset'. 

210 """ 

211 rd = load_description( 

212 source, 

213 format_version=format_version, 

214 perform_io_checks=perform_io_checks, 

215 known_files=known_files, 

216 sha256=sha256, 

217 ) 

218 return ensure_description_is_dataset(rd) 

219 

220 

221def save_bioimageio_yaml_only( 

222 rd: Union[ResourceDescr, BioimageioYamlContent, InvalidDescr], 

223 /, 

224 file: Union[NewPath, FilePath, TextIO], 

225 *, 

226 exclude_unset: bool = True, 

227 exclude_defaults: bool = False, 

228): 

229 """write the metadata of a resource description (`rd`) to `file` 

230 without writing any of the referenced files in it. 

231 

232 Args: 

233 rd: bioimageio resource description 

234 file: file or stream to save to 

235 exclude_unset: Exclude fields that have not explicitly be set. 

236 exclude_defaults: Exclude fields that have the default value (even if set explicitly). 

237 

238 Note: To save a resource description with its associated files as a package, 

239 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

240 """ 

241 if isinstance(rd, ResourceDescrBase): 

242 content = dump_description( 

243 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

244 ) 

245 else: 

246 content = rd 

247 

248 write_yaml(cast(YamlValue, content), file) 

249 

250 

251def load_description_and_validate_format_only( 

252 source: Union[PermissiveFileSource, ZipFile], 

253 /, 

254 *, 

255 format_version: Union[FormatVersionPlaceholder, str] = DISCOVER, 

256 perform_io_checks: Optional[bool] = None, 

257 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

258 sha256: Optional[Sha256] = None, 

259) -> ValidationSummary: 

260 """same as `load_description`, but only return the validation summary. 

261 

262 Returns: 

263 Validation summary of the bioimage.io resource found at `source`. 

264 

265 """ 

266 rd = load_description( 

267 source, 

268 format_version=format_version, 

269 perform_io_checks=perform_io_checks, 

270 known_files=known_files, 

271 sha256=sha256, 

272 ) 

273 assert rd.validation_summary is not None 

274 return rd.validation_summary 

275 

276 

277def update_format( 

278 source: Union[ 

279 ResourceDescr, 

280 PermissiveFileSource, 

281 ZipFile, 

282 BioimageioYamlContent, 

283 InvalidDescr, 

284 ], 

285 /, 

286 *, 

287 output: Union[Path, TextIO, None] = None, 

288 exclude_defaults: bool = True, 

289 perform_io_checks: Optional[bool] = None, 

290) -> Union[LatestResourceDescr, InvalidDescr]: 

291 """Update a resource description. 

292 

293 Notes: 

294 - Invalid **source** descriptions may fail to update. 

295 - The updated description might be invalid (even if the **source** was valid). 

296 """ 

297 

298 if isinstance(source, ResourceDescrBase): 

299 root = source.root 

300 source = dump_description(source) 

301 else: 

302 root = None 

303 

304 if isinstance(source, collections.abc.Mapping): 

305 descr = build_description( 

306 source, 

307 context=get_validation_context().replace( 

308 root=root, perform_io_checks=perform_io_checks 

309 ), 

310 format_version=LATEST, 

311 ) 

312 

313 else: 

314 descr = load_description( 

315 source, 

316 perform_io_checks=perform_io_checks, 

317 format_version=LATEST, 

318 ) 

319 

320 if output is not None: 

321 save_bioimageio_yaml_only(descr, file=output, exclude_defaults=exclude_defaults) 

322 

323 return descr 

324 

325 

326def update_hashes( 

327 source: Union[PermissiveFileSource, ZipFile, ResourceDescr, BioimageioYamlContent], 

328 /, 

329) -> Union[ResourceDescr, InvalidDescr]: 

330 """Update hash values of the files referenced in **source**.""" 

331 if isinstance(source, ResourceDescrBase): 

332 root = source.root 

333 source = dump_description(source) 

334 else: 

335 root = None 

336 

337 context = get_validation_context().replace( 

338 update_hashes=True, root=root, perform_io_checks=True 

339 ) 

340 with context: 

341 if isinstance(source, collections.abc.Mapping): 

342 return build_description(source) 

343 else: 

344 return load_description(source, perform_io_checks=True)