Coverage for bioimageio/spec/_io.py: 67%

1from pathlib import Path 

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

3from zipfile import ZipFile 

4 

5from loguru import logger 

6from pydantic import FilePath, NewPath 

7 

8from ._description import ( 

9 DISCOVER, 

10 LATEST, 

11 InvalidDescr, 

12 LatestResourceDescr, 

13 ResourceDescr, 

14 build_description, 

15 dump_description, 

16 ensure_description_is_dataset, 

17 ensure_description_is_model, 

18) 

19from ._internal.common_nodes import ResourceDescrBase 

20from ._internal.io import BioimageioYamlContent, YamlValue 

21from ._internal.io_basics import Sha256 

22from ._internal.io_utils import open_bioimageio_yaml, write_yaml 

23from ._internal.types import FormatVersionPlaceholder 

24from ._internal.validation_context import get_validation_context 

25from .common import PermissiveFileSource 

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: Path or URL to an rdf.yaml or a bioimage.io package 

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

69 format_version: (optional) Use this argument to load the resource and 

70 convert its metadata to a higher format_version. 

71 perform_io_checks: Wether or not to perform validation that requires file io, 

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

73 absolute file paths is still being checked. 

74 known_files: Allows to bypass download and hashing of referenced files 

75 (even if perform_io_checks is True). 

76 Checked files will be added to this dictionary 

77 with their SHA-256 value. 

78 sha256: Optional SHA-256 value of **source** 

79 

80 Returns: 

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

82 

83 """ 

84 if isinstance(source, ResourceDescrBase): 

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

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

87 return source # pyright: ignore[reportReturnType] 

88 

89 opened = open_bioimageio_yaml(source, sha256=sha256) 

90 

91 context = get_validation_context().replace( 

92 root=opened.original_root, 

93 file_name=opened.original_file_name, 

94 perform_io_checks=perform_io_checks, 

95 known_files=known_files, 

96 ) 

97 

98 return build_description( 

99 opened.content, 

100 context=context, 

101 format_version=format_version, 

102 ) 

103 

104 

105@overload 

106def load_model_description( 

107 source: Union[PermissiveFileSource, ZipFile], 

108 /, 

109 *, 

110 format_version: Literal["latest"], 

111 perform_io_checks: Optional[bool] = None, 

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

113 sha256: Optional[Sha256] = None, 

114) -> ModelDescr: ... 

115 

116 

117@overload 

118def load_model_description( 

119 source: Union[PermissiveFileSource, ZipFile], 

120 /, 

121 *, 

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

123 perform_io_checks: Optional[bool] = None, 

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

125 sha256: Optional[Sha256] = None, 

126) -> AnyModelDescr: ... 

127 

128 

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 """same as `load_description`, but addtionally ensures that the loaded 

139 description is valid and of type 'model'. 

140 

141 Raises: 

142 ValueError: for invalid or non-model resources 

143 """ 

144 rd = load_description( 

145 source, 

146 format_version=format_version, 

147 perform_io_checks=perform_io_checks, 

148 known_files=known_files, 

149 sha256=sha256, 

150 ) 

151 return ensure_description_is_model(rd) 

152 

153 

154@overload 

155def load_dataset_description( 

156 source: Union[PermissiveFileSource, ZipFile], 

157 /, 

158 *, 

159 format_version: Literal["latest"], 

160 perform_io_checks: Optional[bool] = None, 

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

162 sha256: Optional[Sha256] = None, 

163) -> DatasetDescr: ... 

164 

165 

166@overload 

167def load_dataset_description( 

168 source: Union[PermissiveFileSource, ZipFile], 

169 /, 

170 *, 

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

172 perform_io_checks: Optional[bool] = None, 

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

174 sha256: Optional[Sha256] = None, 

175) -> AnyDatasetDescr: ... 

176 

177 

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 """same as `load_description`, but addtionally ensures that the loaded 

188 description is valid and of type 'dataset'. 

189 """ 

190 rd = load_description( 

191 source, 

192 format_version=format_version, 

193 perform_io_checks=perform_io_checks, 

194 known_files=known_files, 

195 sha256=sha256, 

196 ) 

197 return ensure_description_is_dataset(rd) 

198 

199 

200def save_bioimageio_yaml_only( 

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

202 /, 

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

204 *, 

205 exclude_unset: bool = True, 

206 exclude_defaults: bool = False, 

207): 

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

209 without writing any of the referenced files in it. 

210 

211 Args: 

212 rd: bioimageio resource description 

213 file: file or stream to save to 

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

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

216 

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

218 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

219 """ 

220 if isinstance(rd, ResourceDescrBase): 

221 content = dump_description( 

222 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

223 ) 

224 else: 

225 content = rd 

226 

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

228 

229 

230def load_description_and_validate_format_only( 

231 source: Union[PermissiveFileSource, ZipFile], 

232 /, 

233 *, 

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

235 perform_io_checks: Optional[bool] = None, 

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

237 sha256: Optional[Sha256] = None, 

238) -> ValidationSummary: 

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

240 

241 Returns: 

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

243 

244 """ 

245 rd = load_description( 

246 source, 

247 format_version=format_version, 

248 perform_io_checks=perform_io_checks, 

249 known_files=known_files, 

250 sha256=sha256, 

251 ) 

252 assert rd.validation_summary is not None 

253 return rd.validation_summary 

254 

255 

256def update_format( 

257 source: Union[ 

258 ResourceDescr, 

259 PermissiveFileSource, 

260 ZipFile, 

261 BioimageioYamlContent, 

262 InvalidDescr, 

263 ], 

264 /, 

265 *, 

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

267 exclude_defaults: bool = True, 

268 perform_io_checks: Optional[bool] = None, 

269) -> Union[LatestResourceDescr, InvalidDescr]: 

270 """Update a resource description. 

271 

272 Notes: 

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

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

275 """ 

276 

277 if isinstance(source, ResourceDescrBase): 

278 root = source.root 

279 source = dump_description(source) 

280 else: 

281 root = None 

282 

283 if isinstance(source, dict): 

284 descr = build_description( 

285 source, 

286 context=get_validation_context().replace( 

287 root=root, perform_io_checks=perform_io_checks 

288 ), 

289 format_version=LATEST, 

290 ) 

291 

292 else: 

293 descr = load_description( 

294 source, 

295 perform_io_checks=perform_io_checks, 

296 format_version=LATEST, 

297 ) 

298 

299 if output is not None: 

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

301 

302 return descr 

303 

304 

305def update_hashes( 

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

307 /, 

308) -> Union[ResourceDescr, InvalidDescr]: 

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

310 if isinstance(source, ResourceDescrBase): 

311 root = source.root 

312 source = dump_description(source) 

313 else: 

314 root = None 

315 

316 context = get_validation_context().replace( 

317 update_hashes=True, root=root, perform_io_checks=True 

318 ) 

319 with context: 

320 if isinstance(source, dict): 

321 return build_description(source) 

322 else: 

323 return load_description(source, perform_io_checks=True)