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

1import os

2import subprocess

3from dataclasses import dataclass

4from datetime import datetime, timezone

5from io import StringIO

6from itertools import chain

7from pathlib import Path

8from tempfile import TemporaryDirectory

9from textwrap import TextWrapper

10from types import MappingProxyType

11from typing import (

12 Any,

13 Dict,

14 List,

15 Literal,

16 Mapping,

17 NamedTuple,

18 Optional,

19 Sequence,

20 Set,

21 Tuple,

22 Union,

23 no_type_check,

24)

26import markdown

27import rich.console

28import rich.markdown

29import rich.traceback

30from loguru import logger

31from pydantic import (

32 BaseModel,

33 Field,

34 field_serializer,

35 field_validator,

36 model_validator,

37)

38from pydantic_core.core_schema import ErrorType

39from typing_extensions import Self, assert_never

41from bioimageio.spec._internal.type_guards import is_dict

43from ._internal.constants import VERSION

44from ._internal.io import is_yaml_value

45from ._internal.io_utils import write_yaml

46from ._internal.validation_context import ValidationContextSummary

47from ._internal.warning_levels import (

48 ALERT,

49 ALERT_NAME,

50 ERROR,

51 ERROR_NAME,

52 INFO,

53 INFO_NAME,

54 WARNING,

55 WARNING_NAME,

56 WarningLevel,

57 WarningSeverity,

58)

59from .conda_env import CondaEnv

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

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

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

65WarningLevelName = Literal[WarningSeverityName, "error"]

67WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (

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

69)

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

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

72)

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

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

75)

78class ValidationEntry(BaseModel):

79 """Base of `ErrorEntry` and `WarningEntry`"""

81 loc: Loc

82 msg: str

83 type: Union[ErrorType, str]

86class ErrorEntry(ValidationEntry):

87 """An error in a `ValidationDetail`"""

89 with_traceback: bool = False

90 traceback_md: str = ""

91 traceback_html: str = ""

92 # private rich traceback that is not serialized

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

95 @property

96 def traceback_rich(self):

97 return self._traceback_rich

99 def model_post_init(self, __context: Any):

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

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

102 console = rich.console.Console(

103 record=True,

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

105 color_system="truecolor",

106 width=120,

107 tab_size=4,

108 soft_wrap=True,

109 )

110 console.print(self._traceback_rich)

111 if not self.traceback_md:

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

113

114 if not self.traceback_html:

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

116

117

118class WarningEntry(ValidationEntry):

119 """A warning in a `ValidationDetail`"""

120

121 severity: WarningSeverity = WARNING

122

123 @property

124 def severity_name(self) -> WarningSeverityName:

125 return WARNING_SEVERITY_TO_NAME[self.severity]

126

127

128def format_loc(

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

130) -> str:

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

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

133

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

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

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

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

138 if loc_str:

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

140 start = "`"

141 end = "`"

142 elif target == "html":

143 start = "<code>"

144 end = "</code>"

145 elif target == "plain":

146 start = ""

147 end = ""

148 else:

149 assert_never(target)

150

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

152 else:

153 return ""

154

155

156class InstalledPackage(NamedTuple):

157 name: str

158 version: str

159 build: str = ""

160 channel: str = ""

161

162

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

164 """a detail in a validation summary"""

165

166 name: str

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

168 loc: Loc = ()

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

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

171 default_factory=list

172 )

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

174 default_factory=list

175 )

176 context: Optional[ValidationContextSummary] = None

177

178 recommended_env: Optional[CondaEnv] = None

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

180

181 saved_conda_compare: Optional[str] = None

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

183

184 @field_serializer("saved_conda_compare")

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

186 return self.conda_compare

187

188 @model_validator(mode="before")

189 def _load_legacy(cls, data: Any):

190 if is_dict(data):

191 field_name = "conda_compare"

192 if (

193 field_name in data

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

195 ):

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

197

198 return data

199

200 @property

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

202 if self.recommended_env is None:

203 return None

204

205 if self.saved_conda_compare is None:

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

207 if is_yaml_value(dumped_env):

208 with TemporaryDirectory() as d:

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

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

211 write_yaml(dumped_env, f)

212

213 compare_proc = subprocess.run(

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

215 stdout=subprocess.PIPE,

216 stderr=subprocess.STDOUT,

217 shell=True,

218 text=True,

219 )

220 self.saved_conda_compare = (

221 compare_proc.stdout

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

223 )

224 else:

225 self.saved_conda_compare = (

226 "Failed to dump recommended env to valid yaml"

227 )

228

229 return self.saved_conda_compare

230

231 @property

232 def status_icon(self):

233 if self.status == "passed":

234 return "✔️"

235 else:

236 return "❌"

237

238

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

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

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

242

243 name: str

244 """name of the validation"""

245 source_name: str

246 """source of the validated bioimageio description"""

247 id: Optional[str] = None

248 """ID of the resource being validated"""

249 type: str

250 """type of the resource being validated"""

251 format_version: str

252 """format version of the resource being validated"""

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

254 """overall status of the bioimageio validation"""

255 details: List[ValidationDetail]

256 """list of validation details"""

257 env: Set[InstalledPackage] = Field(

258 default_factory=lambda: {

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

260 }

261 )

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

263

264 saved_conda_list: Optional[str] = None

265

266 @field_serializer("saved_conda_list")

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

268 return self.conda_list

269

270 @property

271 def conda_list(self):

272 if self.saved_conda_list is None:

273 p = subprocess.run(

274 ["conda", "list"],

275 stdout=subprocess.PIPE,

276 stderr=subprocess.STDOUT,

277 shell=True,

278 text=True,

279 )

280 self.saved_conda_list = (

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

282 )

283

284 return self.saved_conda_list

285

286 @property

287 def status_icon(self):

288 if self.status == "passed":

289 return "✔️"

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

291 return "🟡"

292 else:

293 return "❌"

294

295 @property

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

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

298

299 @property

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

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

302

303 def format(

304 self,

305 *,

306 width: Optional[int] = None,

307 include_conda_list: bool = False,

308 ):

309 """Format summary as Markdown string"""

310 return self._format(

311 width=width, target="md", include_conda_list=include_conda_list

312 )

313

314 format_md = format

315

316 def format_html(

317 self,

318 *,

319 width: Optional[int] = None,

320 include_conda_list: bool = False,

321 ):

322 md_with_html = self._format(

323 target="html", width=width, include_conda_list=include_conda_list

324 )

325 return markdown.markdown(

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

327 )

328

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

330 # (the generated markdown seems fine)

331 @no_type_check

332 def display(

333 self,

334 *,

335 width: Optional[int] = None,

336 include_conda_list: bool = False,

337 tab_size: int = 4,

338 soft_wrap: bool = True,

339 ) -> None:

340 try: # render as HTML in Jupyter notebook

341 from IPython.core.getipython import get_ipython

342 from IPython.display import display_html

343 except ImportError:

344 pass

345 else:

346 if get_ipython() is not None:

347 _ = display_html(

348 self.format_html(

349 width=width, include_conda_list=include_conda_list

350 ),

351 raw=True,

352 )

353 return

354

355 # render with rich

356 self._format(

357 target=rich.console.Console(

358 width=width,

359 tab_size=tab_size,

360 soft_wrap=soft_wrap,

361 ),

362 width=width,

363 include_conda_list=include_conda_list,

364 )

365

366 def add_detail(self, detail: ValidationDetail):

367 if detail.status == "failed":

368 self.status = "failed"

369 elif detail.status != "passed":

370 assert_never(detail.status)

371

372 self.details.append(detail)

373

374 def log(

375 self,

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

377 ) -> List[Path]:

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

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

380 if to == "display":

381 display = True

382 save_to = []

383 elif isinstance(to, Path):

384 display = False

385 save_to = [to]

386 else:

387 display = "display" in to

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

389

390 if display:

391 self.display()

392

393 return self.save(save_to)

394

395 def save(

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

397 ) -> List[Path]:

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

399

400 Returns:

401 List of file paths the summary was saved to.

402

403 Notes:

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

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

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

407 """

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

409 path = [Path(path)]

410

411 # folder to file paths

412 file_paths: List[Path] = []

413 for p in path:

414 if p.suffix:

415 file_paths.append(p)

416 else:

417 file_paths.extend(

418 [

419 p / "summary.json",

420 p / "summary.md",

421 p / "summary.html",

422 ]

423 )

424

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

426 for p in file_paths:

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

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

429 self.save_json(p)

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

431 self.save_markdown(p)

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

433 self.save_html(p)

434 else:

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

436

437 return file_paths

438

439 def save_json(

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

441 ):

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

443 json_str = self.model_dump_json(indent=indent)

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

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

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

447

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

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

450 formatted = self.format_md()

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

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

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

454

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

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

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

458

459 html = self.format_html()

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

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

462

463 @classmethod

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

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

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

467 return cls.model_validate_json(json_str)

468

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

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

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

472 if isinstance(value, list):

473 return [

474 (

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

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

477 else v

478 )

479 for v in value

480 ]

481 else:

482 return value

483

484 def _format(

485 self,

486 *,

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

488 width: Optional[int],

489 include_conda_list: bool,

490 ):

491 return _format_summary(

492 self,

493 target=target,

494 width=width or 100,

495 include_conda_list=include_conda_list,

496 )

497

498

499def _format_summary(

500 summary: ValidationSummary,

501 *,

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

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

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

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

506 include_conda_list: bool,

507 width: int,

508) -> str:

509 parts: List[str] = []

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

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

512 left_out_details: int = 0

513 left_out_details_header = "Left out details"

514

515 def add_part(part: str):

516 parts.append(part)

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

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

519

520 def add_section(header: str):

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

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

523 elif target == "html":

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

525 else:

526 assert_never(target)

527

528 def header_to_tag(header: str):

529 return (

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

531 .replace("(", "")

532 .replace(")", "")

533 .replace(" ", "-")

534 .lower()

535 )

536

537 def add_as_details_below(

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

539 ):

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

541

542 def make_link(header: str):

543 tag = header_to_tag(header)

544 if target == "md":

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

546 elif target == "html":

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

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

549 return f"{header} below"

550 else:

551 assert_never(target)

552

553 for n in range(1, 4):

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

555 if header in details_below:

556 if details_below[header] == text:

557 return make_link(header)

558 else:

559 details_below[header] = text

560 return make_link(header)

561

562 nonlocal left_out_details

563 left_out_details += 1

564 return make_link(left_out_details_header)

565

566 @dataclass

567 class CodeCell:

568 text: str

569

570 @dataclass

571 class CodeRef:

572 text: str

573

574 def format_code(

575 code: str,

576 lang: str = "",

577 title: str = "Details",

578 cell_line_limit: int = 15,

579 cell_width_limit: int = 120,

580 ) -> Union[CodeRef, CodeCell]:

581

582 if not code.strip():

583 return CodeCell("")

584

585 if target == "html":

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

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

588 put_below = (

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

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

591 )

592 else:

593 put_below = True

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

595

596 if put_below:

597 link = add_as_details_below(title, code)

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

599 else:

600 return CodeCell(code)

601

602 def format_traceback(entry: ErrorEntry):

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

604 if entry.traceback_rich is None:

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

606 else:

607 link = add_as_details_below(

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

609 )

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

611

612 if target == "md":

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

614 elif target == "html":

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

616 else:

617 assert_never(target)

618

619 def format_text(text: str):

620 if target == "html":

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

622 else:

623 return text.split("\n")

624

625 def get_info_table():

626 info_rows = [

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

628 ["status", summary.status],

629 ]

630 if not hide_source:

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

632

633 if summary.id is not None:

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

635

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

637 if not hide_env:

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

639

640 if include_conda_list:

641 info_rows.append(

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

643 )

644 return format_table(info_rows)

645

646 def get_details_table():

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

648

649 def append_detail(

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

651 ):

652

653 text_lines = format_text(text)

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

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

656 status_lines[0] = status

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

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

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

660

661 if code is not None:

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

663

664 for d in summary.details:

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

666

667 for entry in d.errors:

668 append_detail(

669 "❌",

670 entry.loc,

671 entry.msg,

672 None if hide_tracebacks else format_traceback(entry),

673 )

674

675 for entry in d.warnings:

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

677

678 if d.recommended_env is not None:

679 rec_env = StringIO()

680 json_env = d.recommended_env.model_dump(

681 mode="json", exclude_defaults=True

682 )

683 assert is_yaml_value(json_env)

684 write_yaml(json_env, rec_env)

685 append_detail(

686 "",

687 d.loc,

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

689 format_code(

690 rec_env.getvalue(),

691 lang="yaml",

692 title="Recommended Conda Environment",

693 ),

694 )

695

696 if d.conda_compare:

697 wrapped_conda_compare = "\n".join(

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

699 )

700 append_detail(

701 "",

702 d.loc,

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

704 format_code(

705 wrapped_conda_compare,

706 title="Conda Environment Comparison",

707 ),

708 )

709

710 return format_table(details)

711

712 add_part(get_info_table())

713 add_part(get_details_table())

714

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

716 add_section(header)

717 if isinstance(text, tuple):

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

719 text, rich_obj = text

720 target.print(rich_obj)

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

722 else:

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

724

725 if left_out_details:

726 parts.append(

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

728 )

729

730 return "".join(parts)

731

732

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

734 """format `rows` as markdown table"""

735 n_cols = len(rows[0])

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

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

738

739 # fix new lines in table cell

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

741

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

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

744 lines.extend(

745 [

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

747 for row in rows[1:]

748 ]

749 )

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

751

752

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

754 """format `rows` as HTML table"""

755

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

757 return (

758 [" <tr>"]

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

760 + [" </tr>"]

761 )

762

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

764 for r in rows[1:]:

765 table.extend(get_line(r))

766

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

768

769 return "\n".join(table)