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

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 source_name: str

249 """Source of the validated bioimageio description"""

250 id: Optional[str] = None

251 """ID of the resource being validated"""

252 type: str

253 """Type of the resource being validated"""

254 format_version: str

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

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

257 """overall status of the bioimageio validation"""

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

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

260

261 Note: This completeness estimate may change with subsequent releases

262 and should be considered bioimageio.spec version specific.

263 """

264

265 details: List[ValidationDetail]

266 """List of validation details"""

267 env: Set[InstalledPackage] = Field(

268 default_factory=lambda: {

269 InstalledPackage(

270 name="bioimageio.spec",

271 version=VERSION,

272 )

273 }

274 )

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

276

277 saved_conda_list: Optional[str] = None

278

279 @field_serializer("saved_conda_list")

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

281 return self.conda_list

282

283 @property

284 def conda_list(self):

285 if self.saved_conda_list is None:

286 p = subprocess.run(

287 [CONDA_CMD, "list"],

288 stdout=subprocess.PIPE,

289 stderr=subprocess.STDOUT,

290 shell=False,

291 text=True,

292 )

293 self.saved_conda_list = (

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

295 )

296

297 return self.saved_conda_list

298

299 @property

300 def status_icon(self):

301 if self.status == "passed":

302 return "✔️"

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

304 return "🟡"

305 else:

306 return "❌"

307

308 @property

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

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

311

312 @property

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

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

315

316 def format(

317 self,

318 *,

319 width: Optional[int] = None,

320 include_conda_list: bool = False,

321 ):

322 """Format summary as Markdown string"""

323 return self._format(

324 width=width, target="md", include_conda_list=include_conda_list

325 )

326

327 format_md = format

328

329 def format_html(

330 self,

331 *,

332 width: Optional[int] = None,

333 include_conda_list: bool = False,

334 ):

335 md_with_html = self._format(

336 target="html", width=width, include_conda_list=include_conda_list

337 )

338 return markdown.markdown(

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

340 )

341

342 def display(

343 self,

344 *,

345 width: Optional[int] = None,

346 include_conda_list: bool = False,

347 tab_size: int = 4,

348 soft_wrap: bool = True,

349 ) -> None:

350 try: # render as HTML in Jupyter notebook

351 from IPython.core.getipython import get_ipython

352 from IPython.display import (

353 display_html, # pyright: ignore[reportUnknownVariableType]

354 )

355 except ImportError:

356 pass

357 else:

358 if get_ipython() is not None:

359 _ = display_html(

360 self.format_html(

361 width=width, include_conda_list=include_conda_list

362 ),

363 raw=True,

364 )

365 return

366

367 # render with rich

368 _ = self._format(

369 target=rich.console.Console(

370 width=width,

371 tab_size=tab_size,

372 soft_wrap=soft_wrap,

373 ),

374 width=width,

375 include_conda_list=include_conda_list,

376 )

377

378 def add_detail(self, detail: ValidationDetail):

379 if detail.status == "failed":

380 self.status = "failed"

381 elif detail.status != "passed":

382 assert_never(detail.status)

383

384 self.details.append(detail)

385

386 def log(

387 self,

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

389 ) -> List[Path]:

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

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

392 if to == "display":

393 display = True

394 save_to = []

395 elif isinstance(to, Path):

396 display = False

397 save_to = [to]

398 else:

399 display = "display" in to

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

401

402 if display:

403 self.display()

404

405 return self.save(save_to)

406

407 def save(

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

409 ) -> List[Path]:

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

411

412 Returns:

413 List of file paths the summary was saved to.

414

415 Notes:

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

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

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

419 """

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

421 path = [Path(path)]

422

423 # folder to file paths

424 file_paths: List[Path] = []

425 for p in path:

426 if p.suffix:

427 file_paths.append(p)

428 else:

429 file_paths.extend(

430 [

431 p / "summary.json",

432 p / "summary.md",

433 p / "summary.html",

434 ]

435 )

436

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

438 for p in file_paths:

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

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

441 self.save_json(p)

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

443 self.save_markdown(p)

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

445 self.save_html(p)

446 else:

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

448

449 return file_paths

450

451 def save_json(

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

453 ):

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

455 json_str = self.model_dump_json(indent=indent)

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

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

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

459

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

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

462 formatted = self.format_md()

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

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

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

466

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

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

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

470

471 html = self.format_html()

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

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

474

475 @classmethod

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

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

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

479 return cls.model_validate_json(json_str)

480

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

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

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

484 if isinstance(value, list):

485 return [

486 (

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

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

489 else v

490 )

491 for v in value

492 ]

493 else:

494 return value

495

496 def _format(

497 self,

498 *,

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

500 width: Optional[int],

501 include_conda_list: bool,

502 ):

503 return _format_summary(

504 self,

505 target=target,

506 width=width or 100,

507 include_conda_list=include_conda_list,

508 )

509

510

511def _format_summary(

512 summary: ValidationSummary,

513 *,

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

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

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

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

518 include_conda_list: bool,

519 width: int,

520) -> str:

521 parts: List[str] = []

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

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

524 left_out_details: int = 0

525 left_out_details_header = "Left out details"

526

527 def add_part(part: str):

528 parts.append(part)

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

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

531

532 def add_section(header: str):

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

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

535 elif target == "html":

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

537 else:

538 assert_never(target)

539

540 def header_to_tag(header: str):

541 return (

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

543 .replace("(", "")

544 .replace(")", "")

545 .replace(" ", "-")

546 .lower()

547 )

548

549 def add_as_details_below(

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

551 ):

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

553

554 def make_link(header: str):

555 tag = header_to_tag(header)

556 if target == "md":

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

558 elif target == "html":

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

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

561 return f"{header} below"

562 else:

563 assert_never(target)

564

565 for n in range(1, 4):

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

567 if header in details_below:

568 if details_below[header] == text:

569 return make_link(header)

570 else:

571 details_below[header] = text

572 return make_link(header)

573

574 nonlocal left_out_details

575 left_out_details += 1

576 return make_link(left_out_details_header)

577

578 @dataclass

579 class CodeCell:

580 text: str

581

582 @dataclass

583 class CodeRef:

584 text: str

585

586 def format_code(

587 code: str,

588 lang: str = "",

589 title: str = "Details",

590 cell_line_limit: int = 15,

591 cell_width_limit: int = 120,

592 ) -> Union[CodeRef, CodeCell]:

593 if not code.strip():

594 return CodeCell("")

595

596 if target == "html":

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

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

599 put_below = (

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

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

602 )

603 else:

604 put_below = True

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

606

607 if put_below:

608 link = add_as_details_below(title, code)

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

610 else:

611 return CodeCell(code)

612

613 def format_traceback(entry: ErrorEntry):

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

615 if entry.traceback_rich is None:

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

617 else:

618 link = add_as_details_below(

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

620 )

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

622

623 if target == "md":

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

625 elif target == "html":

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

627 else:

628 assert_never(target)

629

630 def format_text(text: str):

631 if target == "html":

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

633 else:

634 return text.split("\n")

635

636 def get_info_table():

637 info_rows = [

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

639 ["status", summary.status],

640 ]

641 if not hide_source:

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

643

644 if summary.id is not None:

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

646

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

648 if not hide_env:

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

650

651 if include_conda_list:

652 info_rows.append(

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

654 )

655 return format_table(info_rows)

656

657 def get_details_table():

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

659

660 def append_detail(

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

662 ):

663 text_lines = format_text(text)

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

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

666 status_lines[0] = status

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

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

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

670

671 if code is not None:

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

673

674 for d in summary.details:

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

676

677 for entry in d.errors:

678 append_detail(

679 "❌",

680 entry.loc,

681 entry.msg,

682 None if hide_tracebacks else format_traceback(entry),

683 )

684

685 for entry in d.warnings:

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

687

688 if d.recommended_env is not None:

689 rec_env = StringIO()

690 json_env = d.recommended_env.model_dump(

691 mode="json", exclude_defaults=True

692 )

693 assert is_yaml_value(json_env)

694 write_yaml(json_env, rec_env)

695 append_detail(

696 "",

697 d.loc,

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

699 format_code(

700 rec_env.getvalue(),

701 lang="yaml",

702 title="Recommended Conda Environment",

703 ),

704 )

705

706 if d.conda_compare:

707 wrapped_conda_compare = "\n".join(

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

709 )

710 append_detail(

711 "",

712 d.loc,

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

714 format_code(

715 wrapped_conda_compare,

716 title="Conda Environment Comparison",

717 ),

718 )

719

720 return format_table(details)

721

722 add_part(get_info_table())

723 add_part(get_details_table())

724

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

726 add_section(header)

727 if isinstance(text, tuple):

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

729 text, rich_obj = text

730 target.print(rich_obj)

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

732 else:

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

734

735 if left_out_details:

736 parts.append(

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

738 )

739

740 return "".join(parts)

741

742

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

744 """format `rows` as markdown table"""

745 n_cols = len(rows[0])

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

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

748

749 # fix new lines in table cell

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

751

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

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

754 lines.extend(

755 [

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

757 for row in rows[1:]

758 ]

759 )

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

761

762

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

764 """format `rows` as HTML table"""

765

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

767 return (

768 [" <tr>"]

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

770 + [" </tr>"]

771 )

772

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

774 for r in rows[1:]:

775 table.extend(get_line(r))

776

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

778

779 return "\n".join(table)