Coverage for bioimageio/spec/summary.py: 41%

1import subprocess 

2from io import StringIO 

3from itertools import chain 

4from pathlib import Path 

5from tempfile import TemporaryDirectory 

6from types import MappingProxyType 

7from typing import ( 

8 Any, 

9 Dict, 

10 Iterable, 

11 List, 

12 Literal, 

13 Mapping, 

14 NamedTuple, 

15 Optional, 

16 Sequence, 

17 Set, 

18 Tuple, 

19 Union, 

20 no_type_check, 

21) 

22 

23import rich.console 

24import rich.markdown 

25from pydantic import BaseModel, Field, field_validator, model_validator 

26from pydantic_core.core_schema import ErrorType 

27from typing_extensions import TypedDict, assert_never 

28 

29from ._internal.constants import VERSION 

30from ._internal.io import is_yaml_value 

31from ._internal.io_utils import write_yaml 

32from ._internal.type_guards import is_mapping 

33from ._internal.warning_levels import ( 

34 ALERT, 

35 ALERT_NAME, 

36 ERROR, 

37 ERROR_NAME, 

38 INFO, 

39 INFO_NAME, 

40 WARNING, 

41 WARNING_NAME, 

42 WarningLevel, 

43 WarningSeverity, 

44) 

45from .conda_env import CondaEnv 

46 

47Loc = Tuple[Union[int, str], ...] 

48"""location of error/warning in a nested data structure""" 

49 

50WarningSeverityName = Literal["info", "warning", "alert"] 

51WarningLevelName = Literal[WarningSeverityName, "error"] 

52 

53WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = ( 

54 MappingProxyType({INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME}) 

55) 

56WARNING_LEVEL_TO_NAME: Mapping[WarningLevel, WarningLevelName] = MappingProxyType( 

57 {INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME, ERROR: ERROR_NAME} 

58) 

59WARNING_NAME_TO_LEVEL: Mapping[WarningLevelName, WarningLevel] = MappingProxyType( 

60 {v: k for k, v in WARNING_LEVEL_TO_NAME.items()} 

61) 

62 

63 

64class ValidationEntry(BaseModel): 

65 """Base of `ErrorEntry` and `WarningEntry`""" 

66 

67 loc: Loc 

68 msg: str 

69 type: Union[ErrorType, str] 

70 

71 

72class ErrorEntry(ValidationEntry): 

73 """An error in a `ValidationDetail`""" 

74 

75 traceback: List[str] = Field(default_factory=list) 

76 

77 

78class WarningEntry(ValidationEntry): 

79 """A warning in a `ValidationDetail`""" 

80 

81 severity: WarningSeverity = WARNING 

82 severity_name: WarningSeverityName = WARNING_NAME 

83 

84 @model_validator(mode="before") 

85 @classmethod 

86 def sync_severity_with_severity_name( 

87 cls, data: Union[Mapping[Any, Any], Any] 

88 ) -> Any: 

89 if is_mapping(data): 

90 data = dict(data) 

91 if ( 

92 "severity" in data 

93 and "severity_name" not in data 

94 and data["severity"] in WARNING_SEVERITY_TO_NAME 

95 ): 

96 data["severity_name"] = WARNING_SEVERITY_TO_NAME[data["severity"]] 

97 

98 if ( 

99 "severity" in data 

100 and "severity_name" not in data 

101 and data["severity"] in WARNING_SEVERITY_TO_NAME 

102 ): 

103 data["severity"] = WARNING_NAME_TO_LEVEL[data["severity_name"]] 

104 

105 return data 

106 

107 

108def format_loc(loc: Loc, enclose_in: str = "`") -> str: 

109 """helper to format a location tuple `Loc` as Markdown string""" 

110 if not loc: 

111 loc = ("__root__",) 

112 

113 loc_str = ".".join(f"({x})" if x[0].isupper() else x for x in map(str, loc)) 

114 

115 # additional field validation can make the location information quite convoluted, e.g. 

116 # `weights.pytorch_state_dict.dependencies.source.function-after[validate_url_ok(), url['http','https']]` Input should be a valid URL, relative URL without a base 

117 # therefore we remove the `.function-after[validate_url_ok(), url['http','https']]` here 

118 brief_loc_str, *_ = loc_str.split(".function-after") 

119 return f"{enclose_in}{brief_loc_str}{enclose_in}" 

120 

121 

122class InstalledPackage(NamedTuple): 

123 name: str 

124 version: str 

125 build: str = "" 

126 channel: str = "" 

127 

128 

129class ValidationContextSummary(TypedDict): 

130 perform_io_checks: bool 

131 known_files: Mapping[str, str] 

132 root: str 

133 warning_level: str 

134 

135 

136class ValidationDetail(BaseModel, extra="allow"): 

137 """a detail in a validation summary""" 

138 

139 name: str 

140 status: Literal["passed", "failed"] 

141 loc: Loc = () 

142 """location in the RDF that this detail applies to""" 

143 errors: List[ErrorEntry] = Field(default_factory=list) 

144 warnings: List[WarningEntry] = Field(default_factory=list) 

145 context: Optional[ValidationContextSummary] = None 

146 

147 recommended_env: Optional[CondaEnv] = None 

148 """recommended conda environemnt for this validation detail""" 

149 conda_compare: Optional[str] = None 

150 """output of `conda compare <recommended env>`""" 

151 

152 def model_post_init(self, __context: Any): 

153 """create `conda_compare` default value if needed""" 

154 super().model_post_init(__context) 

155 if self.recommended_env is None or self.conda_compare is not None: 

156 return 

157 

158 dumped_env = self.recommended_env.model_dump(mode="json") 

159 if not is_yaml_value(dumped_env): 

160 self.conda_compare = "Failed to dump recommended env to valid yaml" 

161 return 

162 

163 with TemporaryDirectory() as d: 

164 path = Path(d) / "env.yaml" 

165 with path.open("w", encoding="utf-8") as f: 

166 write_yaml(dumped_env, f) 

167 

168 compare_proc = subprocess.run( 

169 ["conda", "compare", str(path)], 

170 stdout=subprocess.PIPE, 

171 stderr=subprocess.STDOUT, 

172 shell=True, 

173 text=True, 

174 ) 

175 self.conda_compare = ( 

176 compare_proc.stdout 

177 or f"conda compare exited with {compare_proc.returncode}" 

178 ) 

179 

180 def __str__(self): 

181 return f"{self.__class__.__name__}:\n" + self.format() 

182 

183 @property 

184 def status_icon(self): 

185 if self.status == "passed": 

186 return "✔️" 

187 else: 

188 return "❌" 

189 

190 def format(self, hide_tracebacks: bool = False, root_loc: Loc = ()) -> str: 

191 """format as Markdown string""" 

192 indent = " " if root_loc else "" 

193 errs_wrns = self._format_errors_and_warnings( 

194 hide_tracebacks=hide_tracebacks, root_loc=root_loc 

195 ) 

196 return f"{indent}{self.status_icon} {self.name.strip('.')}: {self.status}{errs_wrns}" 

197 

198 def _format_errors_and_warnings(self, hide_tracebacks: bool, root_loc: Loc): 

199 indent = " " if root_loc else "" 

200 if hide_tracebacks: 

201 tbs = [""] * len(self.errors) 

202 else: 

203 slim_tracebacks = [ 

204 [tt.replace("\n", "<br>") for t in e.traceback if (tt := t.strip())] 

205 for e in self.errors 

206 ] 

207 tbs = [ 

208 ("<br> Traceback:<br> " if st else "") + "<br> ".join(st) 

209 for st in slim_tracebacks 

210 ] 

211 

212 def join_parts(parts: Iterable[Tuple[str, str]]): 

213 last_loc = None 

214 lines: List[str] = [] 

215 for loc, msg in parts: 

216 if loc == last_loc: 

217 lines.append(f"<br> {loc} {msg}") 

218 else: 

219 lines.append(f"<br>- {loc} {msg}") 

220 

221 last_loc = loc 

222 

223 return "".join(lines) 

224 

225 es = join_parts( 

226 (format_loc(root_loc + e.loc), f"{e.msg}{tb}") 

227 for e, tb in zip(self.errors, tbs) 

228 ) 

229 ws = join_parts((format_loc(root_loc + w.loc), w.msg) for w in self.warnings) 

230 

231 return ( 

232 f"\n{indent}errors:\n{es}" 

233 if es 

234 else "" + f"\n{indent}warnings:\n{ws}" if ws else "" 

235 ) 

236 

237 

238class ValidationSummary(BaseModel, extra="allow"): 

239 """Summarizes output of all bioimageio validations and tests 

240 for one specific `ResourceDescr` instance.""" 

241 

242 name: str 

243 source_name: str 

244 type: str 

245 format_version: str 

246 status: Literal["passed", "failed"] 

247 details: List[ValidationDetail] 

248 env: Set[InstalledPackage] = Field( 

249 default_factory=lambda: { 

250 InstalledPackage(name="bioimageio.spec", version=VERSION) 

251 } 

252 ) 

253 """list of selected, relevant package versions""" 

254 

255 conda_list: Optional[Sequence[InstalledPackage]] = None 

256 """parsed output of conda list""" 

257 

258 @property 

259 def status_icon(self): 

260 if self.status == "passed": 

261 return "✔️" 

262 else: 

263 return "❌" 

264 

265 @property 

266 def errors(self) -> List[ErrorEntry]: 

267 return list(chain.from_iterable(d.errors for d in self.details)) 

268 

269 @property 

270 def warnings(self) -> List[WarningEntry]: 

271 return list(chain.from_iterable(d.warnings for d in self.details)) 

272 

273 def __str__(self): 

274 return f"{self.__class__.__name__}:\n" + self.format() 

275 

276 @staticmethod 

277 def _format_md_table(rows: List[List[str]]) -> str: 

278 """format `rows` as markdown table""" 

279 n_cols = len(rows[0]) 

280 assert all(len(row) == n_cols for row in rows) 

281 col_widths = [max(max(len(row[i]) for row in rows), 3) for i in range(n_cols)] 

282 

283 # fix new lines in table cell 

284 rows = [[line.replace("\n", "<br>") for line in r] for r in rows] 

285 

286 lines = [" | ".join(rows[0][i].center(col_widths[i]) for i in range(n_cols))] 

287 lines.append(" | ".join("---".center(col_widths[i]) for i in range(n_cols))) 

288 lines.extend( 

289 [ 

290 " | ".join(row[i].ljust(col_widths[i]) for i in range(n_cols)) 

291 for row in rows[1:] 

292 ] 

293 ) 

294 return "\n| " + " |\n| ".join(lines) + " |\n" 

295 

296 def format( 

297 self, 

298 hide_tracebacks: bool = False, 

299 hide_source: bool = False, 

300 hide_env: bool = False, 

301 root_loc: Loc = (), 

302 ) -> str: 

303 """Format summary as Markdown string 

304 

305 Suitable to embed in HTML using '<br>' instead of '\n'. 

306 """ 

307 info = self._format_md_table( 

308 [[self.status_icon, f"{self.name.strip('.').strip()} {self.status}"]] 

309 + ([] if hide_source else [["source", self.source_name]]) 

310 + [ 

311 ["format version", f"{self.type} {self.format_version}"], 

312 ] 

313 + ([] if hide_env else [[e.name, e.version] for e in self.env]) 

314 ) 

315 

316 def format_loc(loc: Loc): 

317 return "`" + (".".join(map(str, root_loc + loc)) or ".") + "`" 

318 

319 details = [["❓", "location", "detail"]] 

320 for d in self.details: 

321 details.append([d.status_icon, format_loc(d.loc), d.name]) 

322 if d.context is not None: 

323 details.append( 

324 [ 

325 "🔍", 

326 "context.perform_io_checks", 

327 str(d.context["perform_io_checks"]), 

328 ] 

329 ) 

330 if d.context["perform_io_checks"]: 

331 details.append(["🔍", "context.root", d.context["root"]]) 

332 for kfn, sha in d.context["known_files"].items(): 

333 details.append(["🔍", f"context.known_files.{kfn}", sha]) 

334 

335 details.append( 

336 ["🔍", "context.warning_level", d.context["warning_level"]] 

337 ) 

338 

339 if d.recommended_env is not None: 

340 rec_env = StringIO() 

341 json_env = d.recommended_env.model_dump( 

342 mode="json", exclude_defaults=True 

343 ) 

344 assert is_yaml_value(json_env) 

345 write_yaml(json_env, rec_env) 

346 rec_env_code = rec_env.getvalue().replace("\n", "</code><br><code>") 

347 details.append( 

348 [ 

349 "🐍", 

350 format_loc(d.loc), 

351 f"recommended conda env ({d.name})<br>" 

352 + f"<pre><code>{rec_env_code}</code></pre>", 

353 ] 

354 ) 

355 

356 if d.conda_compare: 

357 details.append( 

358 [ 

359 "🐍", 

360 format_loc(d.loc), 

361 f"conda compare ({d.name}):<br>" 

362 + d.conda_compare.replace("\n", "<br>"), 

363 ] 

364 ) 

365 

366 for entry in d.errors: 

367 details.append( 

368 [ 

369 "❌", 

370 format_loc(entry.loc), 

371 entry.msg.replace("\n\n", "<br>").replace("\n", "<br>"), 

372 ] 

373 ) 

374 if hide_tracebacks: 

375 continue 

376 

377 formatted_tb_lines: List[str] = [] 

378 for tb in entry.traceback: 

379 if not (tb_stripped := tb.strip()): 

380 continue 

381 

382 first_tb_line, *tb_lines = tb_stripped.split("\n") 

383 if ( 

384 first_tb_line.startswith('File "') 

385 and '", line' in first_tb_line 

386 ): 

387 path, where = first_tb_line[len('File "') :].split('", line') 

388 try: 

389 p = Path(path) 

390 except Exception: 

391 file_name = path 

392 else: 

393 path = p.as_posix() 

394 file_name = p.name 

395 

396 where = ", line" + where 

397 first_tb_line = f'[{file_name}]({file_name} "{path}"){where}' 

398 

399 if tb_lines: 

400 tb_rest = "<br>`" + "`<br>`".join(tb_lines) + "`" 

401 else: 

402 tb_rest = "" 

403 

404 formatted_tb_lines.append(first_tb_line + tb_rest) 

405 

406 details.append(["", "", "<br>".join(formatted_tb_lines)]) 

407 

408 for entry in d.warnings: 

409 details.append(["⚠", format_loc(entry.loc), entry.msg]) 

410 

411 return f"{info}{self._format_md_table(details)}" 

412 

413 # TODO: fix bug which casuses extensive white space between the info table and details table 

414 @no_type_check 

415 def display(self) -> None: 

416 formatted = self.format() 

417 try: 

418 from IPython.core.getipython import get_ipython 

419 from IPython.display import Markdown, display 

420 except ImportError: 

421 pass 

422 else: 

423 if get_ipython() is not None: 

424 _ = display(Markdown(formatted)) 

425 return 

426 

427 rich_markdown = rich.markdown.Markdown(formatted) 

428 console = rich.console.Console() 

429 console.print(rich_markdown) 

430 

431 def add_detail(self, detail: ValidationDetail): 

432 if detail.status == "failed": 

433 self.status = "failed" 

434 elif detail.status != "passed": 

435 assert_never(detail.status) 

436 

437 self.details.append(detail) 

438 

439 @field_validator("env", mode="before") 

440 def _convert_dict(cls, value: List[Union[List[str], Dict[str, str]]]): 

441 """convert old env value for backwards compatibility""" 

442 if isinstance(value, list): 

443 return [ 

444 ( 

445 (v["name"], v["version"], v.get("build", ""), v.get("channel", "")) 

446 if isinstance(v, dict) and "name" in v and "version" in v 

447 else v 

448 ) 

449 for v in value 

450 ] 

451 else: 

452 return value