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

1import importlib.metadata

2import os

3import subprocess

4from dataclasses import dataclass

5from datetime import datetime, timezone

6from io import StringIO

7from itertools import chain

8from pathlib import Path

9from tempfile import TemporaryDirectory

10from textwrap import TextWrapper

11from types import MappingProxyType

12from typing import (

13 Any,

14 Dict,

15 List,

16 Literal,

17 Mapping,

18 NamedTuple,

19 Optional,

20 Sequence,

21 Set,

22 Tuple,

23 Union,

24)

26import annotated_types

27import markdown

28import rich.console

29import rich.markdown

30import rich.traceback

31from loguru import logger

32from pydantic import (

33 BaseModel,

34 Field,

35 field_serializer,

36 field_validator,

37 model_validator,

38)

39from pydantic_core.core_schema import ErrorType

40from typing_extensions import Annotated, Self, assert_never

42from ._internal.io import is_yaml_value

43from ._internal.io_utils import write_yaml

44from ._internal.type_guards import is_dict

45from ._internal.validation_context import ValidationContextSummary

46from ._internal.warning_levels import (

47 ALERT,

48 ALERT_NAME,

49 ERROR,

50 ERROR_NAME,

51 INFO,

52 INFO_NAME,

53 WARNING,

54 WARNING_NAME,

55 WarningLevel,

56 WarningSeverity,

57)

58from .conda_env import CondaEnv

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

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

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

64WarningLevelName = Literal[WarningSeverityName, "error"]

66WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (

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

68)

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

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

71)

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

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

74)

77class ValidationEntry(BaseModel):

78 """Base of `ErrorEntry` and `WarningEntry`"""

80 loc: Loc

81 msg: str

82 type: Union[ErrorType, str]

85class ErrorEntry(ValidationEntry):

86 """An error in a `ValidationDetail`"""

88 with_traceback: bool = False

89 traceback_md: str = ""

90 traceback_html: str = ""

91 # private rich traceback that is not serialized

92 _traceback_rich: Optional[rich.traceback.Traceback] = None

94 @property

95 def traceback_rich(self):

96 return self._traceback_rich

98 def model_post_init(self, __context: Any):

99 if self.with_traceback and not (self.traceback_md or self.traceback_html):

100 self._traceback_rich = rich.traceback.Traceback()

101 console = rich.console.Console(

102 record=True,

103 file=open(os.devnull, "wt", encoding="utf-8"),

104 color_system="truecolor",

105 width=120,

106 tab_size=4,

107 soft_wrap=True,

108 )

109 console.print(self._traceback_rich)

110 if not self.traceback_md:

111 self.traceback_md = console.export_text(clear=False)

112

113 if not self.traceback_html:

114 self.traceback_html = console.export_html(clear=False)

115

116

117class WarningEntry(ValidationEntry):

118 """A warning in a `ValidationDetail`"""

119

120 severity: WarningSeverity = WARNING

121

122 @property

123 def severity_name(self) -> WarningSeverityName:

124 return WARNING_SEVERITY_TO_NAME[self.severity]

125

126

127def format_loc(

128 loc: Loc, target: Union[Literal["md", "html", "plain"], rich.console.Console]

129) -> str:

130 """helper to format a location tuple **loc**"""

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

132

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

134 # `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

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

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

137 if loc_str:

138 if target == "md" or isinstance(target, rich.console.Console):

139 start = "`"

140 end = "`"

141 elif target == "html":

142 start = "<code>"

143 end = "</code>"

144 elif target == "plain":

145 start = ""

146 end = ""

147 else:

148 assert_never(target)

149

150 return f"{start}{loc_str}{end}"

151 else:

152 return ""

153

154

155class InstalledPackage(NamedTuple):

156 name: str

157 version: str

158 build: str = ""

159 channel: str = ""

160

161

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

163 """a detail in a validation summary"""

164

165 name: str

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

167 loc: Loc = ()

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

169 errors: List[ErrorEntry] = Field( # pyright: ignore[reportUnknownVariableType]

170 default_factory=list

171 )

172 warnings: List[WarningEntry] = Field( # pyright: ignore[reportUnknownVariableType]

173 default_factory=list

174 )

175 context: Optional[ValidationContextSummary] = None

176

177 recommended_env: Optional[CondaEnv] = None

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

179

180 saved_conda_compare: Optional[str] = None

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

182

183 @field_serializer("saved_conda_compare")

184 def _save_conda_compare(self, value: Optional[str]):

185 return self.conda_compare

186

187 @model_validator(mode="before")

188 def _load_legacy(cls, data: Any):

189 if is_dict(data):

190 field_name = "conda_compare"

191 if (

192 field_name in data

193 and (saved_field_name := f"saved_{field_name}") not in data

194 ):

195 data[saved_field_name] = data.pop(field_name)

196

197 return data

198

199 @property

200 def conda_compare(self) -> Optional[str]:

201 if self.recommended_env is None:

202 return None

203

204 if self.saved_conda_compare is None:

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

206 if is_yaml_value(dumped_env):

207 with TemporaryDirectory() as d:

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

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

210 write_yaml(dumped_env, f)

211

212 compare_proc = subprocess.run(

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

214 stdout=subprocess.PIPE,

215 stderr=subprocess.STDOUT,

216 shell=True,

217 text=True,

218 )

219 self.saved_conda_compare = (

220 compare_proc.stdout

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

222 )

223 else:

224 self.saved_conda_compare = (

225 "Failed to dump recommended env to valid yaml"

226 )

227

228 return self.saved_conda_compare

229

230 @property

231 def status_icon(self):

232 if self.status == "passed":

233 return "✔️"

234 else:

235 return "❌"

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 """Name of the validation"""

244 source_name: str

245 """Source of the validated bioimageio description"""

246 id: Optional[str] = None

247 """ID of the resource being validated"""

248 type: str

249 """Type of the resource being validated"""

250 format_version: str

251 """Format version of the resource being validated"""

252 status: Literal["passed", "valid-format", "failed"]

253 """overall status of the bioimageio validation"""

254 metadata_completeness: Annotated[float, annotated_types.Interval(ge=0, le=1)] = 0.0

255 """Estimate of completeness of the metadata in the resource description.

256

257 Note: This completeness estimate may change with subsequent releases

258 and should be considered bioimageio.spec version specific.

259 """

260

261 details: List[ValidationDetail]

262 """List of validation details"""

263 env: Set[InstalledPackage] = Field(

264 default_factory=lambda: {

265 InstalledPackage(

266 name="bioimageio.spec",

267 version=importlib.metadata.version("bioimageio.spec"),

268 )

269 }

270 )

271 """List of selected, relevant package versions"""

272

273 saved_conda_list: Optional[str] = None

274

275 @field_serializer("saved_conda_list")

276 def _save_conda_list(self, value: Optional[str]):

277 return self.conda_list

278

279 @property

280 def conda_list(self):

281 if self.saved_conda_list is None:

282 p = subprocess.run(

283 ["conda", "list"],

284 stdout=subprocess.PIPE,

285 stderr=subprocess.STDOUT,

286 shell=True,

287 text=True,

288 )

289 self.saved_conda_list = (

290 p.stdout or f"`conda list` exited with {p.returncode}"

291 )

292

293 return self.saved_conda_list

294

295 @property

296 def status_icon(self):

297 if self.status == "passed":

298 return "✔️"

299 elif self.status == "valid-format":

300 return "🟡"

301 else:

302 return "❌"

303

304 @property

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

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

307

308 @property

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

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

311

312 def format(

313 self,

314 *,

315 width: Optional[int] = None,

316 include_conda_list: bool = False,

317 ):

318 """Format summary as Markdown string"""

319 return self._format(

320 width=width, target="md", include_conda_list=include_conda_list

321 )

322

323 format_md = format

324

325 def format_html(

326 self,

327 *,

328 width: Optional[int] = None,

329 include_conda_list: bool = False,

330 ):

331 md_with_html = self._format(

332 target="html", width=width, include_conda_list=include_conda_list

333 )

334 return markdown.markdown(

335 md_with_html, extensions=["tables", "fenced_code", "nl2br"]

336 )

337

338 def display(

339 self,

340 *,

341 width: Optional[int] = None,

342 include_conda_list: bool = False,

343 tab_size: int = 4,

344 soft_wrap: bool = True,

345 ) -> None:

346 try: # render as HTML in Jupyter notebook

347 from IPython.core.getipython import get_ipython

348 from IPython.display import (

349 display_html, # pyright: ignore[reportUnknownVariableType]

350 )

351 except ImportError:

352 pass

353 else:

354 if get_ipython() is not None:

355 _ = display_html(

356 self.format_html(

357 width=width, include_conda_list=include_conda_list

358 ),

359 raw=True,

360 )

361 return

362

363 # render with rich

364 _ = self._format(

365 target=rich.console.Console(

366 width=width,

367 tab_size=tab_size,

368 soft_wrap=soft_wrap,

369 ),

370 width=width,

371 include_conda_list=include_conda_list,

372 )

373

374 def add_detail(self, detail: ValidationDetail):

375 if detail.status == "failed":

376 self.status = "failed"

377 elif detail.status != "passed":

378 assert_never(detail.status)

379

380 self.details.append(detail)

381

382 def log(

383 self,

384 to: Union[Literal["display"], Path, Sequence[Union[Literal["display"], Path]]],

385 ) -> List[Path]:

386 """Convenience method to display the validation summary in the terminal and/or

387 save it to disk. See `save` for details."""

388 if to == "display":

389 display = True

390 save_to = []

391 elif isinstance(to, Path):

392 display = False

393 save_to = [to]

394 else:

395 display = "display" in to

396 save_to = [p for p in to if p != "display"]

397

398 if display:

399 self.display()

400

401 return self.save(save_to)

402

403 def save(

404 self, path: Union[Path, Sequence[Path]] = Path("{id}_summary_{now}")

405 ) -> List[Path]:

406 """Save the validation/test summary in JSON, Markdown or HTML format.

407

408 Returns:

409 List of file paths the summary was saved to.

410

411 Notes:

412 - Format is chosen based on the suffix: `.json`, `.md`, `.html`.

413 - If **path** has no suffix it is assumed to be a direcotry to which a

414 `summary.json`, `summary.md` and `summary.html` are saved to.

415 """

416 if isinstance(path, (str, Path)):

417 path = [Path(path)]

418

419 # folder to file paths

420 file_paths: List[Path] = []

421 for p in path:

422 if p.suffix:

423 file_paths.append(p)

424 else:

425 file_paths.extend(

426 [

427 p / "summary.json",

428 p / "summary.md",

429 p / "summary.html",

430 ]

431 )

432

433 now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")

434 for p in file_paths:

435 p = Path(str(p).format(id=self.id or "bioimageio", now=now))

436 if p.suffix == ".json":

437 self.save_json(p)

438 elif p.suffix == ".md":

439 self.save_markdown(p)

440 elif p.suffix == ".html":

441 self.save_html(p)

442 else:

443 raise ValueError(f"Unknown summary path suffix '{p.suffix}'")

444

445 return file_paths

446

447 def save_json(

448 self, path: Path = Path("summary.json"), *, indent: Optional[int] = 2

449 ):

450 """Save validation/test summary as JSON file."""

451 json_str = self.model_dump_json(indent=indent)

452 path.parent.mkdir(exist_ok=True, parents=True)

453 _ = path.write_text(json_str, encoding="utf-8")

454 logger.info("Saved summary to {}", path.absolute())

455

456 def save_markdown(self, path: Path = Path("summary.md")):

457 """Save rendered validation/test summary as Markdown file."""

458 formatted = self.format_md()

459 path.parent.mkdir(exist_ok=True, parents=True)

460 _ = path.write_text(formatted, encoding="utf-8")

461 logger.info("Saved Markdown formatted summary to {}", path.absolute())

462

463 def save_html(self, path: Path = Path("summary.html")) -> None:

464 """Save rendered validation/test summary as HTML file."""

465 path.parent.mkdir(exist_ok=True, parents=True)

466

467 html = self.format_html()

468 _ = path.write_text(html, encoding="utf-8")

469 logger.info("Saved HTML formatted summary to {}", path.absolute())

470

471 @classmethod

472 def load_json(cls, path: Path) -> Self:

473 """Load validation/test summary from a suitable JSON file"""

474 json_str = Path(path).read_text(encoding="utf-8")

475 return cls.model_validate_json(json_str)

476

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

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

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

480 if isinstance(value, list):

481 return [

482 (

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

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

485 else v

486 )

487 for v in value

488 ]

489 else:

490 return value

491

492 def _format(

493 self,

494 *,

495 target: Union[rich.console.Console, Literal["html", "md"]],

496 width: Optional[int],

497 include_conda_list: bool,

498 ):

499 return _format_summary(

500 self,

501 target=target,

502 width=width or 100,

503 include_conda_list=include_conda_list,

504 )

505

506

507def _format_summary(

508 summary: ValidationSummary,

509 *,

510 hide_tracebacks: bool = False, # TODO: remove?

511 hide_source: bool = False, # TODO: remove?

512 hide_env: bool = False, # TODO: remove?

513 target: Union[rich.console.Console, Literal["html", "md"]] = "md",

514 include_conda_list: bool,

515 width: int,

516) -> str:

517 parts: List[str] = []

518 format_table = _format_html_table if target == "html" else _format_md_table

519 details_below: Dict[str, Union[str, Tuple[str, rich.traceback.Traceback]]] = {}

520 left_out_details: int = 0

521 left_out_details_header = "Left out details"

522

523 def add_part(part: str):

524 parts.append(part)

525 if isinstance(target, rich.console.Console):

526 target.print(rich.markdown.Markdown(part))

527

528 def add_section(header: str):

529 if target == "md" or isinstance(target, rich.console.Console):

530 add_part(f"\n### {header}\n")

531 elif target == "html":

532 parts.append(f'<h3 id="{header_to_tag(header)}">{header}</h3>')

533 else:

534 assert_never(target)

535

536 def header_to_tag(header: str):

537 return (

538 header.replace("`", "")

539 .replace("(", "")

540 .replace(")", "")

541 .replace(" ", "-")

542 .lower()

543 )

544

545 def add_as_details_below(

546 title: str, text: Union[str, Tuple[str, rich.traceback.Traceback]]

547 ):

548 """returns a header and its tag to link to details below"""

549

550 def make_link(header: str):

551 tag = header_to_tag(header)

552 if target == "md":

553 return f"[{header}](#{tag})"

554 elif target == "html":

555 return f'<a href="#{tag}">{header}</a>'

556 elif isinstance(target, rich.console.Console):

557 return f"{header} below"

558 else:

559 assert_never(target)

560

561 for n in range(1, 4):

562 header = f"{title} {n}"

563 if header in details_below:

564 if details_below[header] == text:

565 return make_link(header)

566 else:

567 details_below[header] = text

568 return make_link(header)

569

570 nonlocal left_out_details

571 left_out_details += 1

572 return make_link(left_out_details_header)

573

574 @dataclass

575 class CodeCell:

576 text: str

577

578 @dataclass

579 class CodeRef:

580 text: str

581

582 def format_code(

583 code: str,

584 lang: str = "",

585 title: str = "Details",

586 cell_line_limit: int = 15,

587 cell_width_limit: int = 120,

588 ) -> Union[CodeRef, CodeCell]:

589 if not code.strip():

590 return CodeCell("")

591

592 if target == "html":

593 html_lang = f' lang="{lang}"' if lang else ""

594 code = f"<pre{html_lang}>{code}</pre>"

595 put_below = (

596 code.count("\n") > cell_line_limit

597 or max(map(len, code.split("\n"))) > cell_width_limit

598 )

599 else:

600 put_below = True

601 code = f"\n```{lang}\n{code}\n```\n"

602

603 if put_below:

604 link = add_as_details_below(title, code)

605 return CodeRef(f"See {link}.")

606 else:

607 return CodeCell(code)

608

609 def format_traceback(entry: ErrorEntry):

610 if isinstance(target, rich.console.Console):

611 if entry.traceback_rich is None:

612 return format_code(entry.traceback_md, title="Traceback")

613 else:

614 link = add_as_details_below(

615 "Traceback", (entry.traceback_md, entry.traceback_rich)

616 )

617 return CodeRef(f"See {link}.")

618

619 if target == "md":

620 return format_code(entry.traceback_md, title="Traceback")

621 elif target == "html":

622 return format_code(entry.traceback_html, title="Traceback")

623 else:

624 assert_never(target)

625

626 def format_text(text: str):

627 if target == "html":

628 return [f"<pre>{text}</pre>"]

629 else:

630 return text.split("\n")

631

632 def get_info_table():

633 info_rows = [

634 [summary.status_icon, summary.name.strip(".").strip()],

635 ["status", summary.status],

636 ]

637 if not hide_source:

638 info_rows.append(["source", summary.source_name])

639

640 if summary.id is not None:

641 info_rows.append(["id", summary.id])

642

643 info_rows.append(["format version", f"{summary.type} {summary.format_version}"])

644 if not hide_env:

645 info_rows.extend([[e.name, e.version] for e in summary.env])

646

647 if include_conda_list:

648 info_rows.append(

649 ["conda list", format_code(summary.conda_list, title="Conda List").text]

650 )

651 return format_table(info_rows)

652

653 def get_details_table():

654 details = [["", "Location", "Details"]]

655

656 def append_detail(

657 status: str, loc: Loc, text: str, code: Union[CodeRef, CodeCell, None]

658 ):

659 text_lines = format_text(text)

660 status_lines = [""] * len(text_lines)

661 loc_lines = [""] * len(text_lines)

662 status_lines[0] = status

663 loc_lines[0] = format_loc(loc, target)

664 for s_line, loc_line, text_line in zip(status_lines, loc_lines, text_lines):

665 details.append([s_line, loc_line, text_line])

666

667 if code is not None:

668 details.append(["", "", code.text])

669

670 for d in summary.details:

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

672

673 for entry in d.errors:

674 append_detail(

675 "❌",

676 entry.loc,

677 entry.msg,

678 None if hide_tracebacks else format_traceback(entry),

679 )

680

681 for entry in d.warnings:

682 append_detail("⚠", entry.loc, entry.msg, None)

683

684 if d.recommended_env is not None:

685 rec_env = StringIO()

686 json_env = d.recommended_env.model_dump(

687 mode="json", exclude_defaults=True

688 )

689 assert is_yaml_value(json_env)

690 write_yaml(json_env, rec_env)

691 append_detail(

692 "",

693 d.loc,

694 f"recommended conda environment ({d.name})",

695 format_code(

696 rec_env.getvalue(),

697 lang="yaml",

698 title="Recommended Conda Environment",

699 ),

700 )

701

702 if d.conda_compare:

703 wrapped_conda_compare = "\n".join(

704 TextWrapper(width=width - 4).wrap(d.conda_compare)

705 )

706 append_detail(

707 "",

708 d.loc,

709 f"conda compare ({d.name})",

710 format_code(

711 wrapped_conda_compare,

712 title="Conda Environment Comparison",

713 ),

714 )

715

716 return format_table(details)

717

718 add_part(get_info_table())

719 add_part(get_details_table())

720

721 for header, text in details_below.items():

722 add_section(header)

723 if isinstance(text, tuple):

724 assert isinstance(target, rich.console.Console)

725 text, rich_obj = text

726 target.print(rich_obj)

727 parts.append(f"{text}\n")

728 else:

729 add_part(f"{text}\n")

730

731 if left_out_details:

732 parts.append(

733 f"\n{left_out_details_header}\nLeft out {left_out_details} more details for brevity.\n"

734 )

735

736 return "".join(parts)

737

738

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

740 """format `rows` as markdown table"""

741 n_cols = len(rows[0])

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

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

744

745 # fix new lines in table cell

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

747

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

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

750 lines.extend(

751 [

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

753 for row in rows[1:]

754 ]

755 )

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

757

758

759def _format_html_table(rows: List[List[str]]) -> str:

760 """format `rows` as HTML table"""

761

762 def get_line(cells: List[str], cell_tag: Literal["th", "td"] = "td"):

763 return (

764 [" <tr>"]

765 + [f" <{cell_tag}>{c}</{cell_tag}>" for c in cells]

766 + [" </tr>"]

767 )

768

769 table = ["<table>"] + get_line(rows[0], cell_tag="th")

770 for r in rows[1:]:

771 table.extend(get_line(r))

772

773 table.append("</table>")

774

775 return "\n".join(table)