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 Keys should be file paths or URL strings as they appear in the 

86 bioimageio.yaml file. 

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

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

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

90 this dictionary with their SHA-256 value. 

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

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

93 sha256: 

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

95 

96 Returns: 

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

98 

99 """ 

100 if isinstance(source, ResourceDescrBase): 

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

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

103 return source # pyright: ignore[reportReturnType] 

104 

105 opened = open_bioimageio_yaml(source, sha256=sha256) 

106 

107 context = get_validation_context().replace( 

108 root=opened.original_root, 

109 file_name=opened.original_file_name, 

110 original_source_name=opened.original_source_name, 

111 perform_io_checks=perform_io_checks, 

112 known_files=known_files, 

113 ) 

114 

115 return build_description( 

116 opened.content, 

117 context=context, 

118 format_version=format_version, 

119 ) 

120 

121 

122@overload 

123def load_model_description( 

124 source: Union[PermissiveFileSource, ZipFile], 

125 /, 

126 *, 

127 format_version: Literal["latest"], 

128 perform_io_checks: Optional[bool] = None, 

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

130 sha256: Optional[Sha256] = None, 

131) -> ModelDescr: ... 

132 

133 

134@overload 

135def load_model_description( 

136 source: Union[PermissiveFileSource, ZipFile], 

137 /, 

138 *, 

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

140 perform_io_checks: Optional[bool] = None, 

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

142 sha256: Optional[Sha256] = None, 

143) -> AnyModelDescr: ... 

144 

145 

146def load_model_description( 

147 source: Union[PermissiveFileSource, ZipFile], 

148 /, 

149 *, 

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

151 perform_io_checks: Optional[bool] = None, 

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

153 sha256: Optional[Sha256] = None, 

154) -> AnyModelDescr: 

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

156 description is valid and of type 'model'. 

157 

158 Raises: 

159 ValueError: for invalid or non-model resources 

160 """ 

161 rd = load_description( 

162 source, 

163 format_version=format_version, 

164 perform_io_checks=perform_io_checks, 

165 known_files=known_files, 

166 sha256=sha256, 

167 ) 

168 return ensure_description_is_model(rd) 

169 

170 

171@overload 

172def load_dataset_description( 

173 source: Union[PermissiveFileSource, ZipFile], 

174 /, 

175 *, 

176 format_version: Literal["latest"], 

177 perform_io_checks: Optional[bool] = None, 

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

179 sha256: Optional[Sha256] = None, 

180) -> DatasetDescr: ... 

181 

182 

183@overload 

184def load_dataset_description( 

185 source: Union[PermissiveFileSource, ZipFile], 

186 /, 

187 *, 

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

189 perform_io_checks: Optional[bool] = None, 

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

191 sha256: Optional[Sha256] = None, 

192) -> AnyDatasetDescr: ... 

193 

194 

195def load_dataset_description( 

196 source: Union[PermissiveFileSource, ZipFile], 

197 /, 

198 *, 

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

200 perform_io_checks: Optional[bool] = None, 

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

202 sha256: Optional[Sha256] = None, 

203) -> AnyDatasetDescr: 

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

205 description is valid and of type 'dataset'. 

206 """ 

207 rd = load_description( 

208 source, 

209 format_version=format_version, 

210 perform_io_checks=perform_io_checks, 

211 known_files=known_files, 

212 sha256=sha256, 

213 ) 

214 return ensure_description_is_dataset(rd) 

215 

216 

217def save_bioimageio_yaml_only( 

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

219 /, 

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

221 *, 

222 exclude_unset: bool = True, 

223 exclude_defaults: bool = False, 

224): 

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

226 without writing any of the referenced files in it. 

227 

228 Args: 

229 rd: bioimageio resource description 

230 file: file or stream to save to 

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

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

233 

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

235 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

236 """ 

237 if isinstance(rd, ResourceDescrBase): 

238 content = dump_description( 

239 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

240 ) 

241 else: 

242 content = rd 

243 

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

245 

246 

247def load_description_and_validate_format_only( 

248 source: Union[PermissiveFileSource, ZipFile], 

249 /, 

250 *, 

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

252 perform_io_checks: Optional[bool] = None, 

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

254 sha256: Optional[Sha256] = None, 

255) -> ValidationSummary: 

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

257 

258 Returns: 

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

260 

261 """ 

262 rd = load_description( 

263 source, 

264 format_version=format_version, 

265 perform_io_checks=perform_io_checks, 

266 known_files=known_files, 

267 sha256=sha256, 

268 ) 

269 assert rd.validation_summary is not None 

270 return rd.validation_summary 

271 

272 

273def update_format( 

274 source: Union[ 

275 ResourceDescr, 

276 PermissiveFileSource, 

277 ZipFile, 

278 BioimageioYamlContent, 

279 InvalidDescr, 

280 ], 

281 /, 

282 *, 

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

284 exclude_defaults: bool = True, 

285 perform_io_checks: Optional[bool] = None, 

286) -> Union[LatestResourceDescr, InvalidDescr]: 

287 """Update a resource description. 

288 

289 Notes: 

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

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

292 """ 

293 

294 if isinstance(source, ResourceDescrBase): 

295 root = source.root 

296 source = dump_description(source) 

297 else: 

298 root = None 

299 

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

301 descr = build_description( 

302 source, 

303 context=get_validation_context().replace( 

304 root=root, perform_io_checks=perform_io_checks 

305 ), 

306 format_version=LATEST, 

307 ) 

308 

309 else: 

310 descr = load_description( 

311 source, 

312 perform_io_checks=perform_io_checks, 

313 format_version=LATEST, 

314 ) 

315 

316 if output is not None: 

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

318 

319 return descr 

320 

321 

322def update_hashes( 

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

324 /, 

325) -> Union[ResourceDescr, InvalidDescr]: 

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

327 if isinstance(source, ResourceDescrBase): 

328 root = source.root 

329 source = dump_description(source) 

330 else: 

331 root = None 

332 

333 context = get_validation_context().replace( 

334 update_hashes=True, root=root, perform_io_checks=True 

335 ) 

336 with context: 

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

338 return build_description(source) 

339 else: 

340 return load_description(source, perform_io_checks=True)