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 

25from ._internal.validation_context import get_validation_context 

26from .common import PermissiveFileSource 

27from .dataset import AnyDatasetDescr, DatasetDescr 

28from .model import AnyModelDescr, ModelDescr 

29from .summary import ValidationSummary 

30 

31 

32@overload 

33def load_description( 

34 source: Union[PermissiveFileSource, ZipFile], 

35 /, 

36 *, 

37 format_version: Literal["latest"], 

38 perform_io_checks: Optional[bool] = None, 

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

40 sha256: Optional[Sha256] = None, 

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

42 

43 

44@overload 

45def load_description( 

46 source: Union[PermissiveFileSource, ZipFile], 

47 /, 

48 *, 

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

50 perform_io_checks: Optional[bool] = None, 

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

52 sha256: Optional[Sha256] = None, 

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

54 

55 

56def load_description( 

57 source: Union[PermissiveFileSource, ZipFile], 

58 /, 

59 *, 

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

61 perform_io_checks: Optional[bool] = None, 

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

63 sha256: Optional[Sha256] = None, 

64) -> Union[ResourceDescr, InvalidDescr]: 

65 """load a bioimage.io resource description 

66 

67 Args: 

68 source: 

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

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

71 format_version: 

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

73 convert its metadata to a higher format_version. 

74 Note: 

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

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

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

78 - Conversion to lower format versions is not supported. 

79 perform_io_checks: 

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

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

82 absolute file paths is still being checked. 

83 known_files: 

84 Allows to bypass download and hashing of referenced files 

85 (even if perform_io_checks is True). 

86 Checked files will be added to this dictionary 

87 with their SHA-256 value. 

88 sha256: 

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

90 

91 Returns: 

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

93 

94 """ 

95 if isinstance(source, ResourceDescrBase): 

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

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

98 return source # pyright: ignore[reportReturnType] 

99 

100 opened = open_bioimageio_yaml(source, sha256=sha256) 

101 

102 context = get_validation_context().replace( 

103 root=opened.original_root, 

104 file_name=opened.original_file_name, 

105 original_source_name=opened.original_source_name, 

106 perform_io_checks=perform_io_checks, 

107 known_files=known_files, 

108 ) 

109 

110 return build_description( 

111 opened.content, 

112 context=context, 

113 format_version=format_version, 

114 ) 

115 

116 

117@overload 

118def load_model_description( 

119 source: Union[PermissiveFileSource, ZipFile], 

120 /, 

121 *, 

122 format_version: Literal["latest"], 

123 perform_io_checks: Optional[bool] = None, 

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

125 sha256: Optional[Sha256] = None, 

126) -> ModelDescr: ... 

127 

128 

129@overload 

130def load_model_description( 

131 source: Union[PermissiveFileSource, ZipFile], 

132 /, 

133 *, 

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

135 perform_io_checks: Optional[bool] = None, 

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

137 sha256: Optional[Sha256] = None, 

138) -> AnyModelDescr: ... 

139 

140 

141def load_model_description( 

142 source: Union[PermissiveFileSource, ZipFile], 

143 /, 

144 *, 

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

146 perform_io_checks: Optional[bool] = None, 

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

148 sha256: Optional[Sha256] = None, 

149) -> AnyModelDescr: 

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

151 description is valid and of type 'model'. 

152 

153 Raises: 

154 ValueError: for invalid or non-model resources 

155 """ 

156 rd = load_description( 

157 source, 

158 format_version=format_version, 

159 perform_io_checks=perform_io_checks, 

160 known_files=known_files, 

161 sha256=sha256, 

162 ) 

163 return ensure_description_is_model(rd) 

164 

165 

166@overload 

167def load_dataset_description( 

168 source: Union[PermissiveFileSource, ZipFile], 

169 /, 

170 *, 

171 format_version: Literal["latest"], 

172 perform_io_checks: Optional[bool] = None, 

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

174 sha256: Optional[Sha256] = None, 

175) -> DatasetDescr: ... 

176 

177 

178@overload 

179def load_dataset_description( 

180 source: Union[PermissiveFileSource, ZipFile], 

181 /, 

182 *, 

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

184 perform_io_checks: Optional[bool] = None, 

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

186 sha256: Optional[Sha256] = None, 

187) -> AnyDatasetDescr: ... 

188 

189 

190def load_dataset_description( 

191 source: Union[PermissiveFileSource, ZipFile], 

192 /, 

193 *, 

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

195 perform_io_checks: Optional[bool] = None, 

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

197 sha256: Optional[Sha256] = None, 

198) -> AnyDatasetDescr: 

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

200 description is valid and of type 'dataset'. 

201 """ 

202 rd = load_description( 

203 source, 

204 format_version=format_version, 

205 perform_io_checks=perform_io_checks, 

206 known_files=known_files, 

207 sha256=sha256, 

208 ) 

209 return ensure_description_is_dataset(rd) 

210 

211 

212def save_bioimageio_yaml_only( 

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

214 /, 

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

216 *, 

217 exclude_unset: bool = True, 

218 exclude_defaults: bool = False, 

219): 

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

221 without writing any of the referenced files in it. 

222 

223 Args: 

224 rd: bioimageio resource description 

225 file: file or stream to save to 

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

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

228 

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

230 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

231 """ 

232 if isinstance(rd, ResourceDescrBase): 

233 content = dump_description( 

234 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

235 ) 

236 else: 

237 content = rd 

238 

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

240 

241 

242def load_description_and_validate_format_only( 

243 source: Union[PermissiveFileSource, ZipFile], 

244 /, 

245 *, 

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

247 perform_io_checks: Optional[bool] = None, 

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

249 sha256: Optional[Sha256] = None, 

250) -> ValidationSummary: 

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

252 

253 Returns: 

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

255 

256 """ 

257 rd = load_description( 

258 source, 

259 format_version=format_version, 

260 perform_io_checks=perform_io_checks, 

261 known_files=known_files, 

262 sha256=sha256, 

263 ) 

264 assert rd.validation_summary is not None 

265 return rd.validation_summary 

266 

267 

268def update_format( 

269 source: Union[ 

270 ResourceDescr, 

271 PermissiveFileSource, 

272 ZipFile, 

273 BioimageioYamlContent, 

274 InvalidDescr, 

275 ], 

276 /, 

277 *, 

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

279 exclude_defaults: bool = True, 

280 perform_io_checks: Optional[bool] = None, 

281) -> Union[LatestResourceDescr, InvalidDescr]: 

282 """Update a resource description. 

283 

284 Notes: 

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

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

287 """ 

288 

289 if isinstance(source, ResourceDescrBase): 

290 root = source.root 

291 source = dump_description(source) 

292 else: 

293 root = None 

294 

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

296 descr = build_description( 

297 source, 

298 context=get_validation_context().replace( 

299 root=root, perform_io_checks=perform_io_checks 

300 ), 

301 format_version=LATEST, 

302 ) 

303 

304 else: 

305 descr = load_description( 

306 source, 

307 perform_io_checks=perform_io_checks, 

308 format_version=LATEST, 

309 ) 

310 

311 if output is not None: 

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

313 

314 return descr 

315 

316 

317def update_hashes( 

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

319 /, 

320) -> Union[ResourceDescr, InvalidDescr]: 

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

322 if isinstance(source, ResourceDescrBase): 

323 root = source.root 

324 source = dump_description(source) 

325 else: 

326 root = None 

327 

328 context = get_validation_context().replace( 

329 update_hashes=True, root=root, perform_io_checks=True 

330 ) 

331 with context: 

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

333 return build_description(source) 

334 else: 

335 return load_description(source, perform_io_checks=True)