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 Checked files will be added to this dictionary 

86 with their SHA-256 value. 

87 sha256: 

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

89 

90 Returns: 

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

92 

93 """ 

94 if isinstance(source, ResourceDescrBase): 

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

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

97 return source # pyright: ignore[reportReturnType] 

98 

99 opened = open_bioimageio_yaml(source, sha256=sha256) 

100 

101 context = get_validation_context().replace( 

102 root=opened.original_root, 

103 file_name=opened.original_file_name, 

104 original_source_name=opened.original_source_name, 

105 perform_io_checks=perform_io_checks, 

106 known_files=known_files, 

107 ) 

108 

109 return build_description( 

110 opened.content, 

111 context=context, 

112 format_version=format_version, 

113 ) 

114 

115 

116@overload 

117def load_model_description( 

118 source: Union[PermissiveFileSource, ZipFile], 

119 /, 

120 *, 

121 format_version: Literal["latest"], 

122 perform_io_checks: Optional[bool] = None, 

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

124 sha256: Optional[Sha256] = None, 

125) -> ModelDescr: ... 

126 

127 

128@overload 

129def load_model_description( 

130 source: Union[PermissiveFileSource, ZipFile], 

131 /, 

132 *, 

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

134 perform_io_checks: Optional[bool] = None, 

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

136 sha256: Optional[Sha256] = None, 

137) -> AnyModelDescr: ... 

138 

139 

140def load_model_description( 

141 source: Union[PermissiveFileSource, ZipFile], 

142 /, 

143 *, 

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

145 perform_io_checks: Optional[bool] = None, 

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

147 sha256: Optional[Sha256] = None, 

148) -> AnyModelDescr: 

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

150 description is valid and of type 'model'. 

151 

152 Raises: 

153 ValueError: for invalid or non-model resources 

154 """ 

155 rd = load_description( 

156 source, 

157 format_version=format_version, 

158 perform_io_checks=perform_io_checks, 

159 known_files=known_files, 

160 sha256=sha256, 

161 ) 

162 return ensure_description_is_model(rd) 

163 

164 

165@overload 

166def load_dataset_description( 

167 source: Union[PermissiveFileSource, ZipFile], 

168 /, 

169 *, 

170 format_version: Literal["latest"], 

171 perform_io_checks: Optional[bool] = None, 

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

173 sha256: Optional[Sha256] = None, 

174) -> DatasetDescr: ... 

175 

176 

177@overload 

178def load_dataset_description( 

179 source: Union[PermissiveFileSource, ZipFile], 

180 /, 

181 *, 

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

183 perform_io_checks: Optional[bool] = None, 

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

185 sha256: Optional[Sha256] = None, 

186) -> AnyDatasetDescr: ... 

187 

188 

189def load_dataset_description( 

190 source: Union[PermissiveFileSource, ZipFile], 

191 /, 

192 *, 

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

194 perform_io_checks: Optional[bool] = None, 

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

196 sha256: Optional[Sha256] = None, 

197) -> AnyDatasetDescr: 

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

199 description is valid and of type 'dataset'. 

200 """ 

201 rd = load_description( 

202 source, 

203 format_version=format_version, 

204 perform_io_checks=perform_io_checks, 

205 known_files=known_files, 

206 sha256=sha256, 

207 ) 

208 return ensure_description_is_dataset(rd) 

209 

210 

211def save_bioimageio_yaml_only( 

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

213 /, 

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

215 *, 

216 exclude_unset: bool = True, 

217 exclude_defaults: bool = False, 

218): 

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

220 without writing any of the referenced files in it. 

221 

222 Args: 

223 rd: bioimageio resource description 

224 file: file or stream to save to 

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

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

227 

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

229 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

230 """ 

231 if isinstance(rd, ResourceDescrBase): 

232 content = dump_description( 

233 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

234 ) 

235 else: 

236 content = rd 

237 

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

239 

240 

241def load_description_and_validate_format_only( 

242 source: Union[PermissiveFileSource, ZipFile], 

243 /, 

244 *, 

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

246 perform_io_checks: Optional[bool] = None, 

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

248 sha256: Optional[Sha256] = None, 

249) -> ValidationSummary: 

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

251 

252 Returns: 

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

254 

255 """ 

256 rd = load_description( 

257 source, 

258 format_version=format_version, 

259 perform_io_checks=perform_io_checks, 

260 known_files=known_files, 

261 sha256=sha256, 

262 ) 

263 assert rd.validation_summary is not None 

264 return rd.validation_summary 

265 

266 

267def update_format( 

268 source: Union[ 

269 ResourceDescr, 

270 PermissiveFileSource, 

271 ZipFile, 

272 BioimageioYamlContent, 

273 InvalidDescr, 

274 ], 

275 /, 

276 *, 

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

278 exclude_defaults: bool = True, 

279 perform_io_checks: Optional[bool] = None, 

280) -> Union[LatestResourceDescr, InvalidDescr]: 

281 """Update a resource description. 

282 

283 Notes: 

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

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

286 """ 

287 

288 if isinstance(source, ResourceDescrBase): 

289 root = source.root 

290 source = dump_description(source) 

291 else: 

292 root = None 

293 

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

295 descr = build_description( 

296 source, 

297 context=get_validation_context().replace( 

298 root=root, perform_io_checks=perform_io_checks 

299 ), 

300 format_version=LATEST, 

301 ) 

302 

303 else: 

304 descr = load_description( 

305 source, 

306 perform_io_checks=perform_io_checks, 

307 format_version=LATEST, 

308 ) 

309 

310 if output is not None: 

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

312 

313 return descr 

314 

315 

316def update_hashes( 

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

318 /, 

319) -> Union[ResourceDescr, InvalidDescr]: 

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

321 if isinstance(source, ResourceDescrBase): 

322 root = source.root 

323 source = dump_description(source) 

324 else: 

325 root = None 

326 

327 context = get_validation_context().replace( 

328 update_hashes=True, root=root, perform_io_checks=True 

329 ) 

330 with context: 

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

332 return build_description(source) 

333 else: 

334 return load_description(source, perform_io_checks=True)