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

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 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 bioimageio.spec._internal.type_guards import is_dict

44from ._internal.constants import VERSION

45from ._internal.io import is_yaml_value

46from ._internal.io_utils import write_yaml

47from ._internal.validation_context import ValidationContextSummary

48from ._internal.warning_levels import (

49 ALERT,

50 ALERT_NAME,

51 ERROR,

52 ERROR_NAME,

53 INFO,

54 INFO_NAME,

55 WARNING,

56 WARNING_NAME,

57 WarningLevel,

58 WarningSeverity,

59)

60from .conda_env import CondaEnv

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

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

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

66WarningLevelName = Literal[WarningSeverityName, "error"]

68WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (

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

70)

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

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

73)

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

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

76)

79class ValidationEntry(BaseModel):

80 """Base of `ErrorEntry` and `WarningEntry`"""

82 loc: Loc

83 msg: str

84 type: Union[ErrorType, str]

87class ErrorEntry(ValidationEntry):

88 """An error in a `ValidationDetail`"""

90 with_traceback: bool = False

91 traceback_md: str = ""

92 traceback_html: str = ""

93 # private rich traceback that is not serialized

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

96 @property

97 def traceback_rich(self):

98 return self._traceback_rich

100 def model_post_init(self, __context: Any):

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

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

103 console = rich.console.Console(

104 record=True,

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

106 color_system="truecolor",

107 width=120,

108 tab_size=4,

109 soft_wrap=True,

110 )

111 console.print(self._traceback_rich)

112 if not self.traceback_md:

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

114

115 if not self.traceback_html:

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

117

118

119class WarningEntry(ValidationEntry):

120 """A warning in a `ValidationDetail`"""

121

122 severity: WarningSeverity = WARNING

123

124 @property

125 def severity_name(self) -> WarningSeverityName:

126 return WARNING_SEVERITY_TO_NAME[self.severity]

127

128

129def format_loc(

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

131) -> str:

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

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

134

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

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

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

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

139 if loc_str:

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

141 start = "`"

142 end = "`"

143 elif target == "html":

144 start = "<code>"

145 end = "</code>"

146 elif target == "plain":

147 start = ""

148 end = ""

149 else:

150 assert_never(target)

151

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

153 else:

154 return ""

155

156

157class InstalledPackage(NamedTuple):

158 name: str

159 version: str

160 build: str = ""

161 channel: str = ""

162

163

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

165 """a detail in a validation summary"""

166

167 name: str

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

169 loc: Loc = ()

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

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

172 default_factory=list

173 )

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

175 default_factory=list

176 )

177 context: Optional[ValidationContextSummary] = None

178

179 recommended_env: Optional[CondaEnv] = None

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

181

182 saved_conda_compare: Optional[str] = None

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

184

185 @field_serializer("saved_conda_compare")

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

187 return self.conda_compare

188

189 @model_validator(mode="before")

190 def _load_legacy(cls, data: Any):

191 if is_dict(data):

192 field_name = "conda_compare"

193 if (

194 field_name in data

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

196 ):

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

198

199 return data

200

201 @property

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

203 if self.recommended_env is None:

204 return None

205

206 if self.saved_conda_compare is None:

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

208 if is_yaml_value(dumped_env):

209 with TemporaryDirectory() as d:

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

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

212 write_yaml(dumped_env, f)

213

214 compare_proc = subprocess.run(

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

216 stdout=subprocess.PIPE,

217 stderr=subprocess.STDOUT,

218 shell=True,

219 text=True,

220 )

221 self.saved_conda_compare = (

222 compare_proc.stdout

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

224 )

225 else:

226 self.saved_conda_compare = (

227 "Failed to dump recommended env to valid yaml"

228 )

229

230 return self.saved_conda_compare

231

232 @property

233 def status_icon(self):

234 if self.status == "passed":

235 return "✔️"

236 else:

237 return "❌"

238

239

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

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

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

243

244 name: str

245 """Name of the validation"""

246 source_name: str

247 """Source of the validated bioimageio description"""

248 id: Optional[str] = None

249 """ID of the resource being validated"""

250 type: str

251 """Type of the resource being validated"""

252 format_version: str

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

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

255 """overall status of the bioimageio validation"""

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

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

258

259 Note: This completeness estimate may change with subsequent releases

260 and should be considered bioimageio.spec version specific.

261 """

262

263 details: List[ValidationDetail]

264 """List of validation details"""

265 env: Set[InstalledPackage] = Field(

266 default_factory=lambda: {

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

268 }

269 )

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

271

272 saved_conda_list: Optional[str] = None

273

274 @field_serializer("saved_conda_list")

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

276 return self.conda_list

277

278 @property

279 def conda_list(self):

280 if self.saved_conda_list is None:

281 p = subprocess.run(

282 ["conda", "list"],

283 stdout=subprocess.PIPE,

284 stderr=subprocess.STDOUT,

285 shell=True,

286 text=True,

287 )

288 self.saved_conda_list = (

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

290 )

291

292 return self.saved_conda_list

293

294 @property

295 def status_icon(self):

296 if self.status == "passed":

297 return "✔️"

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

299 return "🟡"

300 else:

301 return "❌"

302

303 @property

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

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

306

307 @property

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

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

310

311 def format(

312 self,

313 *,

314 width: Optional[int] = None,

315 include_conda_list: bool = False,

316 ):

317 """Format summary as Markdown string"""

318 return self._format(

319 width=width, target="md", include_conda_list=include_conda_list

320 )

321

322 format_md = format

323

324 def format_html(

325 self,

326 *,

327 width: Optional[int] = None,

328 include_conda_list: bool = False,

329 ):

330 md_with_html = self._format(

331 target="html", width=width, include_conda_list=include_conda_list

332 )

333 return markdown.markdown(

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

335 )

336

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

338 # (the generated markdown seems fine)

339 @no_type_check

340 def display(

341 self,

342 *,

343 width: Optional[int] = None,

344 include_conda_list: bool = False,

345 tab_size: int = 4,

346 soft_wrap: bool = True,

347 ) -> None:

348 try: # render as HTML in Jupyter notebook

349 from IPython.core.getipython import get_ipython

350 from IPython.display import display_html

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

590 if not code.strip():

591 return CodeCell("")

592

593 if target == "html":

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

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

596 put_below = (

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

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

599 )

600 else:

601 put_below = True

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

603

604 if put_below:

605 link = add_as_details_below(title, code)

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

607 else:

608 return CodeCell(code)

609

610 def format_traceback(entry: ErrorEntry):

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

612 if entry.traceback_rich is None:

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

614 else:

615 link = add_as_details_below(

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

617 )

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

619

620 if target == "md":

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

622 elif target == "html":

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

624 else:

625 assert_never(target)

626

627 def format_text(text: str):

628 if target == "html":

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

630 else:

631 return text.split("\n")

632

633 def get_info_table():

634 info_rows = [

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

636 ["status", summary.status],

637 ]

638 if not hide_source:

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

640

641 if summary.id is not None:

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

643

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

645 if not hide_env:

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

647

648 if include_conda_list:

649 info_rows.append(

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

651 )

652 return format_table(info_rows)

653

654 def get_details_table():

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

656

657 def append_detail(

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

659 ):

660

661 text_lines = format_text(text)

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

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

664 status_lines[0] = status

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

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

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

668

669 if code is not None:

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

671

672 for d in summary.details:

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

674

675 for entry in d.errors:

676 append_detail(

677 "❌",

678 entry.loc,

679 entry.msg,

680 None if hide_tracebacks else format_traceback(entry),

681 )

682

683 for entry in d.warnings:

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

685

686 if d.recommended_env is not None:

687 rec_env = StringIO()

688 json_env = d.recommended_env.model_dump(

689 mode="json", exclude_defaults=True

690 )

691 assert is_yaml_value(json_env)

692 write_yaml(json_env, rec_env)

693 append_detail(

694 "",

695 d.loc,

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

697 format_code(

698 rec_env.getvalue(),

699 lang="yaml",

700 title="Recommended Conda Environment",

701 ),

702 )

703

704 if d.conda_compare:

705 wrapped_conda_compare = "\n".join(

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

707 )

708 append_detail(

709 "",

710 d.loc,

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

712 format_code(

713 wrapped_conda_compare,

714 title="Conda Environment Comparison",

715 ),

716 )

717

718 return format_table(details)

719

720 add_part(get_info_table())

721 add_part(get_details_table())

722

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

724 add_section(header)

725 if isinstance(text, tuple):

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

727 text, rich_obj = text

728 target.print(rich_obj)

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

730 else:

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

732

733 if left_out_details:

734 parts.append(

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

736 )

737

738 return "".join(parts)

739

740

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

742 """format `rows` as markdown table"""

743 n_cols = len(rows[0])

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

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

746

747 # fix new lines in table cell

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

749

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

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

752 lines.extend(

753 [

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

755 for row in rows[1:]

756 ]

757 )

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

759

760

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

762 """format `rows` as HTML table"""

763

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

765 return (

766 [" <tr>"]

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

768 + [" </tr>"]

769 )

770

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

772 for r in rows[1:]:

773 table.extend(get_line(r))

774

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

776

777 return "\n".join(table)