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

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

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

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

71 convert its metadata to a higher format_version. 

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

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

74 absolute file paths is still being checked. 

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

76 (even if perform_io_checks is True). 

77 Checked files will be added to this dictionary 

78 with their SHA-256 value. 

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

80 

81 Returns: 

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

83 

84 """ 

85 if isinstance(source, ResourceDescrBase): 

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

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

88 return source # pyright: ignore[reportReturnType] 

89 

90 opened = open_bioimageio_yaml(source, sha256=sha256) 

91 

92 context = get_validation_context().replace( 

93 root=opened.original_root, 

94 file_name=opened.original_file_name, 

95 perform_io_checks=perform_io_checks, 

96 known_files=known_files, 

97 ) 

98 

99 return build_description( 

100 opened.content, 

101 context=context, 

102 format_version=format_version, 

103 ) 

104 

105 

106@overload 

107def load_model_description( 

108 source: Union[PermissiveFileSource, ZipFile], 

109 /, 

110 *, 

111 format_version: Literal["latest"], 

112 perform_io_checks: Optional[bool] = None, 

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

114 sha256: Optional[Sha256] = None, 

115) -> ModelDescr: ... 

116 

117 

118@overload 

119def load_model_description( 

120 source: Union[PermissiveFileSource, ZipFile], 

121 /, 

122 *, 

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

124 perform_io_checks: Optional[bool] = None, 

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

126 sha256: Optional[Sha256] = None, 

127) -> AnyModelDescr: ... 

128 

129 

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

140 description is valid and of type 'model'. 

141 

142 Raises: 

143 ValueError: for invalid or non-model resources 

144 """ 

145 rd = load_description( 

146 source, 

147 format_version=format_version, 

148 perform_io_checks=perform_io_checks, 

149 known_files=known_files, 

150 sha256=sha256, 

151 ) 

152 return ensure_description_is_model(rd) 

153 

154 

155@overload 

156def load_dataset_description( 

157 source: Union[PermissiveFileSource, ZipFile], 

158 /, 

159 *, 

160 format_version: Literal["latest"], 

161 perform_io_checks: Optional[bool] = None, 

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

163 sha256: Optional[Sha256] = None, 

164) -> DatasetDescr: ... 

165 

166 

167@overload 

168def load_dataset_description( 

169 source: Union[PermissiveFileSource, ZipFile], 

170 /, 

171 *, 

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

173 perform_io_checks: Optional[bool] = None, 

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

175 sha256: Optional[Sha256] = None, 

176) -> AnyDatasetDescr: ... 

177 

178 

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

189 description is valid and of type 'dataset'. 

190 """ 

191 rd = load_description( 

192 source, 

193 format_version=format_version, 

194 perform_io_checks=perform_io_checks, 

195 known_files=known_files, 

196 sha256=sha256, 

197 ) 

198 return ensure_description_is_dataset(rd) 

199 

200 

201def save_bioimageio_yaml_only( 

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

203 /, 

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

205 *, 

206 exclude_unset: bool = True, 

207 exclude_defaults: bool = False, 

208): 

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

210 without writing any of the referenced files in it. 

211 

212 Args: 

213 rd: bioimageio resource description 

214 file: file or stream to save to 

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

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

217 

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

219 use `save_bioimageio_package` or `save_bioimageio_package_as_folder`. 

220 """ 

221 if isinstance(rd, ResourceDescrBase): 

222 content = dump_description( 

223 rd, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults 

224 ) 

225 else: 

226 content = rd 

227 

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

229 

230 

231def load_description_and_validate_format_only( 

232 source: Union[PermissiveFileSource, ZipFile], 

233 /, 

234 *, 

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

236 perform_io_checks: Optional[bool] = None, 

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

238 sha256: Optional[Sha256] = None, 

239) -> ValidationSummary: 

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

241 

242 Returns: 

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

244 

245 """ 

246 rd = load_description( 

247 source, 

248 format_version=format_version, 

249 perform_io_checks=perform_io_checks, 

250 known_files=known_files, 

251 sha256=sha256, 

252 ) 

253 assert rd.validation_summary is not None 

254 return rd.validation_summary 

255 

256 

257def update_format( 

258 source: Union[ 

259 ResourceDescr, 

260 PermissiveFileSource, 

261 ZipFile, 

262 BioimageioYamlContent, 

263 InvalidDescr, 

264 ], 

265 /, 

266 *, 

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

268 exclude_defaults: bool = True, 

269 perform_io_checks: Optional[bool] = None, 

270) -> Union[LatestResourceDescr, InvalidDescr]: 

271 """Update a resource description. 

272 

273 Notes: 

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

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

276 """ 

277 

278 if isinstance(source, ResourceDescrBase): 

279 root = source.root 

280 source = dump_description(source) 

281 else: 

282 root = None 

283 

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

285 descr = build_description( 

286 source, 

287 context=get_validation_context().replace( 

288 root=root, perform_io_checks=perform_io_checks 

289 ), 

290 format_version=LATEST, 

291 ) 

292 

293 else: 

294 descr = load_description( 

295 source, 

296 perform_io_checks=perform_io_checks, 

297 format_version=LATEST, 

298 ) 

299 

300 if output is not None: 

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

302 

303 return descr 

304 

305 

306def update_hashes( 

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

308 /, 

309) -> Union[ResourceDescr, InvalidDescr]: 

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

311 if isinstance(source, ResourceDescrBase): 

312 root = source.root 

313 source = dump_description(source) 

314 else: 

315 root = None 

316 

317 context = get_validation_context().replace( 

318 update_hashes=True, root=root, perform_io_checks=True 

319 ) 

320 with context: 

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

322 return build_description(source) 

323 else: 

324 return load_description(source, perform_io_checks=True)