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)

25import annotated_types

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 Annotated, 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 metadata_completeness: Annotated[float, annotated_types.Interval(ge=0, le=1)] = 0.0

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

257

258 Note: This completeness estimate may change with subsequent releases

259 and should be considered bioimageio.spec version specific.

260 """

261

262 details: List[ValidationDetail]

263 """List of validation details"""

264 env: Set[InstalledPackage] = Field(

265 default_factory=lambda: {

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

267 }

268 )

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

270

271 saved_conda_list: Optional[str] = None

272

273 @field_serializer("saved_conda_list")

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

275 return self.conda_list

276

277 @property

278 def conda_list(self):

279 if self.saved_conda_list is None:

280 p = subprocess.run(

281 ["conda", "list"],

282 stdout=subprocess.PIPE,

283 stderr=subprocess.STDOUT,

284 shell=True,

285 text=True,

286 )

287 self.saved_conda_list = (

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

289 )

290

291 return self.saved_conda_list

292

293 @property

294 def status_icon(self):

295 if self.status == "passed":

296 return "✔️"

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

298 return "🟡"

299 else:

300 return "❌"

301

302 @property

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

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

305

306 @property

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

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

309

310 def format(

311 self,

312 *,

313 width: Optional[int] = None,

314 include_conda_list: bool = False,

315 ):

316 """Format summary as Markdown string"""

317 return self._format(

318 width=width, target="md", include_conda_list=include_conda_list

319 )

320

321 format_md = format

322

323 def format_html(

324 self,

325 *,

326 width: Optional[int] = None,

327 include_conda_list: bool = False,

328 ):

329 md_with_html = self._format(

330 target="html", width=width, include_conda_list=include_conda_list

331 )

332 return markdown.markdown(

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

334 )

335

336 def display(

337 self,

338 *,

339 width: Optional[int] = None,

340 include_conda_list: bool = False,

341 tab_size: int = 4,

342 soft_wrap: bool = True,

343 ) -> None:

344 try: # render as HTML in Jupyter notebook

345 from IPython.core.getipython import get_ipython

346 from IPython.display import (

347 display_html, # pyright: ignore[reportUnknownVariableType]

348 )

349 except ImportError:

350 pass

351 else:

352 if get_ipython() is not None:

353 _ = display_html(

354 self.format_html(

355 width=width, include_conda_list=include_conda_list

356 ),

357 raw=True,

358 )

359 return

360

361 # render with rich

362 _ = self._format(

363 target=rich.console.Console(

364 width=width,

365 tab_size=tab_size,

366 soft_wrap=soft_wrap,

367 ),

368 width=width,

369 include_conda_list=include_conda_list,

370 )

371

372 def add_detail(self, detail: ValidationDetail):

373 if detail.status == "failed":

374 self.status = "failed"

375 elif detail.status != "passed":

376 assert_never(detail.status)

377

378 self.details.append(detail)

379

380 def log(

381 self,

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

383 ) -> List[Path]:

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

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

386 if to == "display":

387 display = True

388 save_to = []

389 elif isinstance(to, Path):

390 display = False

391 save_to = [to]

392 else:

393 display = "display" in to

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

395

396 if display:

397 self.display()

398

399 return self.save(save_to)

400

401 def save(

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

403 ) -> List[Path]:

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

405

406 Returns:

407 List of file paths the summary was saved to.

408

409 Notes:

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

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

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

413 """

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

415 path = [Path(path)]

416

417 # folder to file paths

418 file_paths: List[Path] = []

419 for p in path:

420 if p.suffix:

421 file_paths.append(p)

422 else:

423 file_paths.extend(

424 [

425 p / "summary.json",

426 p / "summary.md",

427 p / "summary.html",

428 ]

429 )

430

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

432 for p in file_paths:

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

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

435 self.save_json(p)

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

437 self.save_markdown(p)

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

439 self.save_html(p)

440 else:

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

442

443 return file_paths

444

445 def save_json(

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

447 ):

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

449 json_str = self.model_dump_json(indent=indent)

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

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

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

453

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

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

456 formatted = self.format_md()

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

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

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

460

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

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

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

464

465 html = self.format_html()

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

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

468

469 @classmethod

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

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

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

473 return cls.model_validate_json(json_str)

474

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

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

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

478 if isinstance(value, list):

479 return [

480 (

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

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

483 else v

484 )

485 for v in value

486 ]

487 else:

488 return value

489

490 def _format(

491 self,

492 *,

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

494 width: Optional[int],

495 include_conda_list: bool,

496 ):

497 return _format_summary(

498 self,

499 target=target,

500 width=width or 100,

501 include_conda_list=include_conda_list,

502 )

503

504

505def _format_summary(

506 summary: ValidationSummary,

507 *,

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

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

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

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

512 include_conda_list: bool,

513 width: int,

514) -> str:

515 parts: List[str] = []

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

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

518 left_out_details: int = 0

519 left_out_details_header = "Left out details"

520

521 def add_part(part: str):

522 parts.append(part)

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

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

525

526 def add_section(header: str):

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

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

529 elif target == "html":

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

531 else:

532 assert_never(target)

533

534 def header_to_tag(header: str):

535 return (

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

537 .replace("(", "")

538 .replace(")", "")

539 .replace(" ", "-")

540 .lower()

541 )

542

543 def add_as_details_below(

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

545 ):

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

547

548 def make_link(header: str):

549 tag = header_to_tag(header)

550 if target == "md":

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

552 elif target == "html":

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

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

555 return f"{header} below"

556 else:

557 assert_never(target)

558

559 for n in range(1, 4):

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

561 if header in details_below:

562 if details_below[header] == text:

563 return make_link(header)

564 else:

565 details_below[header] = text

566 return make_link(header)

567

568 nonlocal left_out_details

569 left_out_details += 1

570 return make_link(left_out_details_header)

571

572 @dataclass

573 class CodeCell:

574 text: str

575

576 @dataclass

577 class CodeRef:

578 text: str

579

580 def format_code(

581 code: str,

582 lang: str = "",

583 title: str = "Details",

584 cell_line_limit: int = 15,

585 cell_width_limit: int = 120,

586 ) -> Union[CodeRef, CodeCell]:

587

588 if not code.strip():

589 return CodeCell("")

590

591 if target == "html":

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

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

594 put_below = (

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

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

597 )

598 else:

599 put_below = True

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

601

602 if put_below:

603 link = add_as_details_below(title, code)

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

605 else:

606 return CodeCell(code)

607

608 def format_traceback(entry: ErrorEntry):

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

610 if entry.traceback_rich is None:

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

612 else:

613 link = add_as_details_below(

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

615 )

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

617

618 if target == "md":

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

620 elif target == "html":

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

622 else:

623 assert_never(target)

624

625 def format_text(text: str):

626 if target == "html":

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

628 else:

629 return text.split("\n")

630

631 def get_info_table():

632 info_rows = [

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

634 ["status", summary.status],

635 ]

636 if not hide_source:

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

638

639 if summary.id is not None:

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

641

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

643 if not hide_env:

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

645

646 if include_conda_list:

647 info_rows.append(

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

649 )

650 return format_table(info_rows)

651

652 def get_details_table():

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

654

655 def append_detail(

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

657 ):

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)