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

1import html

2import os

3import platform

4import subprocess

5from dataclasses import dataclass

6from datetime import datetime, timezone

7from io import StringIO

8from itertools import chain

9from pathlib import Path

10from tempfile import TemporaryDirectory

11from textwrap import TextWrapper

12from types import MappingProxyType

13from typing import (

14 Any,

15 Dict,

16 List,

17 Literal,

18 Mapping,

19 NamedTuple,

20 Optional,

21 Sequence,

22 Set,

23 Tuple,

24 Union,

25)

27import annotated_types

28import markdown

29import rich.console

30import rich.markdown

31import rich.traceback

32from loguru import logger

33from pydantic import (

34 BaseModel,

35 Field,

36 field_serializer,

37 field_validator,

38 model_validator,

39)

40from pydantic_core.core_schema import ErrorType

41from typing_extensions import Annotated, Self, assert_never

43from ._internal.io import is_yaml_value

44from ._internal.io_utils import write_yaml

45from ._internal.type_guards import is_dict

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 ._version import VERSION

60from .conda_env import CondaEnv

62CONDA_CMD = "conda.bat" if platform.system() == "Windows" else "conda"

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

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

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

68WarningLevelName = Literal[WarningSeverityName, "error"]

70WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (

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

72)

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

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

75)

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

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

78)

81class ValidationEntry(BaseModel):

82 """Base of `ErrorEntry` and `WarningEntry`"""

84 loc: Loc

85 msg: str

86 type: Union[ErrorType, str]

89class ErrorEntry(ValidationEntry):

90 """An error in a `ValidationDetail`"""

92 with_traceback: bool = False

93 traceback_md: str = ""

94 traceback_html: str = ""

95 # private rich traceback that is not serialized

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

98 @property

99 def traceback_rich(self):

100 return self._traceback_rich

101

102 def model_post_init(self, __context: Any):

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

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

105 console = rich.console.Console(

106 record=True,

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

108 color_system="truecolor",

109 width=120,

110 tab_size=4,

111 soft_wrap=True,

112 )

113 console.print(self._traceback_rich)

114 if not self.traceback_md:

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

116

117 if not self.traceback_html:

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

119

120

121class WarningEntry(ValidationEntry):

122 """A warning in a `ValidationDetail`"""

123

124 severity: WarningSeverity = WARNING

125

126 @property

127 def severity_name(self) -> WarningSeverityName:

128 return WARNING_SEVERITY_TO_NAME[self.severity]

129

130

131def format_loc(

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

133) -> str:

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

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

136

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

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

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

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

141 if loc_str:

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

143 start = "`"

144 end = "`"

145 elif target == "html":

146 start = "<code>"

147 end = "</code>"

148 elif target == "plain":

149 start = ""

150 end = ""

151 else:

152 assert_never(target)

153

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

155 else:

156 return ""

157

158

159class InstalledPackage(NamedTuple):

160 name: str

161 version: str

162 build: str = ""

163 channel: str = ""

164

165

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

167 """a detail in a validation summary"""

168

169 name: str

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

171 loc: Loc = ()

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

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

174 default_factory=list

175 )

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

177 default_factory=list

178 )

179 context: Optional[ValidationContextSummary] = None

180

181 recommended_env: Optional[CondaEnv] = None

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

183

184 saved_conda_compare: Optional[str] = None

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

186

187 @field_serializer("saved_conda_compare")

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

189 return self.conda_compare

190

191 @model_validator(mode="before")

192 def _load_legacy(cls, data: Any):

193 if is_dict(data):

194 field_name = "conda_compare"

195 if (

196 field_name in data

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

198 ):

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

200

201 return data

202

203 @property

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

205 if self.recommended_env is None:

206 return None

207

208 if self.saved_conda_compare is None:

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

210 if is_yaml_value(dumped_env):

211 with TemporaryDirectory() as d:

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

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

214 write_yaml(dumped_env, f)

215

216 compare_proc = subprocess.run(

217 [CONDA_CMD, "compare", str(path)],

218 stdout=subprocess.PIPE,

219 stderr=subprocess.STDOUT,

220 shell=False,

221 text=True,

222 )

223 self.saved_conda_compare = (

224 compare_proc.stdout

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

226 )

227 else:

228 self.saved_conda_compare = (

229 "Failed to dump recommended env to valid yaml"

230 )

231

232 return self.saved_conda_compare

233

234 @property

235 def status_icon(self):

236 if self.status == "passed":

237 return "✔️"

238 else:

239 return "❌"

240

241

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

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

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

245

246 name: str

247 """Name of the validation"""

248

249 source_name: str

250 """Source of the validated bioimageio description"""

251

252 id: Optional[str] = None

253 """ID of the resource being validated"""

254

255 type: str

256 """Type of the resource being validated"""

257

258 format_version: str

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

260

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

262 """Overall status of the bioimageio validation"""

263

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

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

266

267 Note: This completeness estimate may change with subsequent releases

268 and should be considered bioimageio.spec version specific.

269 """

270

271 details: List[ValidationDetail]

272 """List of validation details"""

273 env: Set[InstalledPackage] = Field(

274 default_factory=lambda: {

275 InstalledPackage(

276 name="bioimageio.spec",

277 version=VERSION,

278 )

279 }

280 )

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

282

283 saved_conda_list: Optional[str] = None

284

285 @field_serializer("saved_conda_list")

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

287 return self.conda_list

288

289 @property

290 def conda_list(self):

291 if self.saved_conda_list is None:

292 p = subprocess.run(

293 [CONDA_CMD, "list"],

294 stdout=subprocess.PIPE,

295 stderr=subprocess.STDOUT,

296 shell=False,

297 text=True,

298 )

299 self.saved_conda_list = (

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

301 )

302

303 return self.saved_conda_list

304

305 @property

306 def status_icon(self):

307 if self.status == "passed":

308 return "✔️"

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

310 return "🟡"

311 else:

312 return "❌"

313

314 @property

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

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

317

318 @property

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

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

321

322 def format(

323 self,

324 *,

325 width: Optional[int] = None,

326 include_conda_list: bool = False,

327 ):

328 """Format summary as Markdown string"""

329 return self._format(

330 width=width, target="md", include_conda_list=include_conda_list

331 )

332

333 format_md = format

334

335 def format_html(

336 self,

337 *,

338 width: Optional[int] = None,

339 include_conda_list: bool = False,

340 ):

341 md_with_html = self._format(

342 target="html", width=width, include_conda_list=include_conda_list

343 )

344 return markdown.markdown(

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

346 )

347

348 def display(

349 self,

350 *,

351 width: Optional[int] = None,

352 include_conda_list: bool = False,

353 tab_size: int = 4,

354 soft_wrap: bool = True,

355 ) -> None:

356 try: # render as HTML in Jupyter notebook

357 from IPython.core.getipython import get_ipython

358 from IPython.display import (

359 display_html, # pyright: ignore[reportUnknownVariableType]

360 )

361 except ImportError:

362 pass

363 else:

364 if get_ipython() is not None:

365 _ = display_html(

366 self.format_html(

367 width=width, include_conda_list=include_conda_list

368 ),

369 raw=True,

370 )

371 return

372

373 # render with rich

374 _ = self._format(

375 target=rich.console.Console(

376 width=width,

377 tab_size=tab_size,

378 soft_wrap=soft_wrap,

379 ),

380 width=width,

381 include_conda_list=include_conda_list,

382 )

383

384 def add_detail(self, detail: ValidationDetail, update_status: bool = True):

385 if update_status:

386 if self.status == "valid-format" and detail.status == "passed":

387 # once status is 'valid-format' we can only improve to 'passed'

388 self.status = "passed"

389 elif self.status == "passed" and detail.status == "failed":

390 # once status is 'passed' it can only degrade to 'valid-format'

391 self.status = "valid-format"

392 # once format is 'failed' it cannot improve

393

394 self.details.append(detail)

395

396 def log(

397 self,

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

399 ) -> List[Path]:

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

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

402 if to == "display":

403 display = True

404 save_to = []

405 elif isinstance(to, Path):

406 display = False

407 save_to = [to]

408 else:

409 display = "display" in to

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

411

412 if display:

413 self.display()

414

415 return self.save(save_to)

416

417 def save(

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

419 ) -> List[Path]:

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

421

422 Returns:

423 List of file paths the summary was saved to.

424

425 Notes:

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

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

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

429 """

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

431 path = [Path(path)]

432

433 # folder to file paths

434 file_paths: List[Path] = []

435 for p in path:

436 if p.suffix:

437 file_paths.append(p)

438 else:

439 file_paths.extend(

440 [

441 p / "summary.json",

442 p / "summary.md",

443 p / "summary.html",

444 ]

445 )

446

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

448 for p in file_paths:

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

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

451 self.save_json(p)

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

453 self.save_markdown(p)

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

455 self.save_html(p)

456 else:

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

458

459 return file_paths

460

461 def save_json(

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

463 ):

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

465 json_str = self.model_dump_json(indent=indent)

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

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

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

469

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

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

472 formatted = self.format_md()

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

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

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

476

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

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

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

480

481 html = self.format_html()

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

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

484

485 @classmethod

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

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

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

489 return cls.model_validate_json(json_str)

490

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

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

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

494 if isinstance(value, list):

495 return [

496 (

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

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

499 else v

500 )

501 for v in value

502 ]

503 else:

504 return value

505

506 def _format(

507 self,

508 *,

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

510 width: Optional[int],

511 include_conda_list: bool,

512 ):

513 return _format_summary(

514 self,

515 target=target,

516 width=width or 100,

517 include_conda_list=include_conda_list,

518 )

519

520

521def _format_summary(

522 summary: ValidationSummary,

523 *,

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

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

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

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

528 include_conda_list: bool,

529 width: int,

530) -> str:

531 parts: List[str] = []

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

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

534 left_out_details: int = 0

535 left_out_details_header = "Left out details"

536

537 def add_part(part: str):

538 parts.append(part)

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

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

541

542 def add_section(header: str):

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

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

545 elif target == "html":

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

547 else:

548 assert_never(target)

549

550 def header_to_tag(header: str):

551 return (

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

553 .replace("(", "")

554 .replace(")", "")

555 .replace(" ", "-")

556 .lower()

557 )

558

559 def add_as_details_below(

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

561 ):

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

563

564 def make_link(header: str):

565 tag = header_to_tag(header)

566 if target == "md":

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

568 elif target == "html":

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

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

571 return f"{header} below"

572 else:

573 assert_never(target)

574

575 for n in range(1, 4):

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

577 if header in details_below:

578 if details_below[header] == text:

579 return make_link(header)

580 else:

581 details_below[header] = text

582 return make_link(header)

583

584 nonlocal left_out_details

585 left_out_details += 1

586 return make_link(left_out_details_header)

587

588 @dataclass

589 class CodeCell:

590 text: str

591

592 @dataclass

593 class CodeRef:

594 text: str

595

596 def format_code(

597 code: str,

598 lang: str = "",

599 title: str = "Details",

600 cell_line_limit: int = 15,

601 cell_width_limit: int = 120,

602 ) -> Union[CodeRef, CodeCell]:

603 if not code.strip():

604 return CodeCell("")

605

606 if target == "html":

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

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

609 put_below = (

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

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

612 )

613 else:

614 put_below = True

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

616

617 if put_below:

618 link = add_as_details_below(title, code)

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

620 else:

621 return CodeCell(code)

622

623 def format_traceback(entry: ErrorEntry):

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

625 if entry.traceback_rich is None:

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

627 else:

628 link = add_as_details_below(

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

630 )

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

632

633 if target == "md":

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

635 elif target == "html":

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

637 else:

638 assert_never(target)

639

640 def format_text(text: str):

641 if target == "html":

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

643 else:

644 return text.split("\n")

645

646 def get_info_table():

647 info_rows = [

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

649 ["status", summary.status],

650 ]

651 if not hide_source:

652 info_rows.append(["source", html.escape(summary.source_name)])

653

654 if summary.id is not None:

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

656

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

658 if not hide_env:

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

660

661 if include_conda_list:

662 info_rows.append(

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

664 )

665 return format_table(info_rows)

666

667 def get_details_table():

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

669

670 def append_detail(

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

672 ):

673 text_lines = format_text(text)

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

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

676 status_lines[0] = status

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

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

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

680

681 if code is not None:

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

683

684 for d in summary.details:

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

686

687 for entry in d.errors:

688 append_detail(

689 "❌",

690 entry.loc,

691 entry.msg,

692 None if hide_tracebacks else format_traceback(entry),

693 )

694

695 for entry in d.warnings:

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

697

698 if d.recommended_env is not None:

699 rec_env = StringIO()

700 json_env = d.recommended_env.model_dump(

701 mode="json", exclude_defaults=True

702 )

703 assert is_yaml_value(json_env)

704 write_yaml(json_env, rec_env)

705 append_detail(

706 "",

707 d.loc,

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

709 format_code(

710 rec_env.getvalue(),

711 lang="yaml",

712 title="Recommended Conda Environment",

713 ),

714 )

715

716 if d.conda_compare:

717 wrapped_conda_compare = "\n".join(

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

719 )

720 append_detail(

721 "",

722 d.loc,

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

724 format_code(

725 wrapped_conda_compare,

726 title="Conda Environment Comparison",

727 ),

728 )

729

730 return format_table(details)

731

732 add_part(get_info_table())

733 add_part(get_details_table())

734

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

736 add_section(header)

737 if isinstance(text, tuple):

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

739 text, rich_obj = text

740 target.print(rich_obj)

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

742 else:

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

744

745 if left_out_details:

746 parts.append(

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

748 )

749

750 return "".join(parts)

751

752

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

754 """format `rows` as markdown table"""

755 n_cols = len(rows[0])

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

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

758

759 # fix new lines in table cell

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

761

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

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

764 lines.extend(

765 [

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

767 for row in rows[1:]

768 ]

769 )

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

771

772

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

774 """format `rows` as HTML table"""

775

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

777 return (

778 [" <tr>"]

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

780 + [" </tr>"]

781 )

782

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

784 for r in rows[1:]:

785 table.extend(get_line(r))

786

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

788

789 return "\n".join(table)