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.types import NotEmpty

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(default_factory=list)

172 warnings: List[WarningEntry] = Field(default_factory=list)

173 context: Optional[ValidationContextSummary] = None

174

175 recommended_env: Optional[CondaEnv] = None

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

177

178 saved_conda_compare: Optional[str] = None

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

180

181 @field_serializer("saved_conda_compare")

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

183 return self.conda_compare

184

185 @model_validator(mode="before")

186 def _load_legacy(cls, data: Any):

187 if is_dict(data):

188 field_name = "conda_compare"

189 if (

190 field_name in data

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

192 ):

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

194

195 return data

196

197 @property

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

199 if self.recommended_env is None:

200 return None

201

202 if self.saved_conda_compare is None:

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

204 if is_yaml_value(dumped_env):

205 with TemporaryDirectory() as d:

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

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

208 write_yaml(dumped_env, f)

209

210 compare_proc = subprocess.run(

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

212 stdout=subprocess.PIPE,

213 stderr=subprocess.STDOUT,

214 shell=True,

215 text=True,

216 )

217 self.saved_conda_compare = (

218 compare_proc.stdout

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

220 )

221 else:

222 self.saved_conda_compare = (

223 "Failed to dump recommended env to valid yaml"

224 )

225

226 return self.saved_conda_compare

227

228 @property

229 def status_icon(self):

230 if self.status == "passed":

231 return "✔️"

232 else:

233 return "❌"

234

235

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

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

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

239

240 name: str

241 """name of the validation"""

242 source_name: str

243 """source of the validated bioimageio description"""

244 id: Optional[str] = None

245 """ID of the resource being validated"""

246 type: str

247 """type of the resource being validated"""

248 format_version: str

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

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

251 """overall status of the bioimageio validation"""

252 details: NotEmpty[List[ValidationDetail]]

253 """list of validation details"""

254 env: Set[InstalledPackage] = Field(

255 default_factory=lambda: {

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

257 }

258 )

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

260

261 saved_conda_list: Optional[str] = None

262

263 @field_serializer("saved_conda_list")

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

265 return self.conda_list

266

267 @property

268 def conda_list(self):

269 if self.saved_conda_list is None:

270 p = subprocess.run(

271 ["conda", "list"],

272 stdout=subprocess.PIPE,

273 stderr=subprocess.STDOUT,

274 shell=True,

275 text=True,

276 )

277 self.saved_conda_list = (

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

279 )

280

281 return self.saved_conda_list

282

283 @property

284 def status_icon(self):

285 if self.status == "passed":

286 return "✔️"

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

288 return "🟡"

289 else:

290 return "❌"

291

292 @property

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

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

295

296 @property

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

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

299

300 def format(

301 self,

302 *,

303 width: Optional[int] = None,

304 include_conda_list: bool = False,

305 ):

306 """Format summary as Markdown string"""

307 return self._format(

308 width=width, target="md", include_conda_list=include_conda_list

309 )

310

311 format_md = format

312

313 def format_html(

314 self,

315 *,

316 width: Optional[int] = None,

317 include_conda_list: bool = False,

318 ):

319 md_with_html = self._format(

320 target="html", width=width, include_conda_list=include_conda_list

321 )

322 return markdown.markdown(

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

324 )

325

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

327 # (the generated markdown seems fine)

328 @no_type_check

329 def display(

330 self,

331 *,

332 width: Optional[int] = None,

333 include_conda_list: bool = False,

334 tab_size: int = 4,

335 soft_wrap: bool = True,

336 ) -> None:

337 try: # render as HTML in Jupyter notebook

338 from IPython.core.getipython import get_ipython

339 from IPython.display import display_html

340 except ImportError:

341 pass

342 else:

343 if get_ipython() is not None:

344 _ = display_html(

345 self.format_html(

346 width=width, include_conda_list=include_conda_list

347 ),

348 raw=True,

349 )

350 return

351

352 # render with rich

353 self._format(

354 target=rich.console.Console(

355 width=width,

356 tab_size=tab_size,

357 soft_wrap=soft_wrap,

358 ),

359 width=width,

360 include_conda_list=include_conda_list,

361 )

362

363 def add_detail(self, detail: ValidationDetail):

364 if detail.status == "failed":

365 self.status = "failed"

366 elif detail.status != "passed":

367 assert_never(detail.status)

368

369 self.details.append(detail)

370

371 def log(

372 self,

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

374 ) -> List[Path]:

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

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

377 if to == "display":

378 display = True

379 save_to = []

380 elif isinstance(to, Path):

381 display = False

382 save_to = [to]

383 else:

384 display = "display" in to

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

386

387 if display:

388 self.display()

389

390 return self.save(save_to)

391

392 def save(

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

394 ) -> List[Path]:

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

396

397 Returns:

398 List of file paths the summary was saved to.

399

400 Notes:

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

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

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

404 """

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

406 path = [Path(path)]

407

408 # folder to file paths

409 file_paths: List[Path] = []

410 for p in path:

411 if p.suffix:

412 file_paths.append(p)

413 else:

414 file_paths.extend(

415 [

416 p / "summary.json",

417 p / "summary.md",

418 p / "summary.html",

419 ]

420 )

421

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

423 for p in file_paths:

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

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

426 self.save_json(p)

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

428 self.save_markdown(p)

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

430 self.save_html(p)

431 else:

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

433

434 return file_paths

435

436 def save_json(

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

438 ):

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

440 json_str = self.model_dump_json(indent=indent)

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

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

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

444

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

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

447 formatted = self.format_md()

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

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

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

451

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

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

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

455

456 html = self.format_html()

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

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

459

460 @classmethod

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

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

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

464 return cls.model_validate_json(json_str)

465

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

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

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

469 if isinstance(value, list):

470 return [

471 (

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

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

474 else v

475 )

476 for v in value

477 ]

478 else:

479 return value

480

481 def _format(

482 self,

483 *,

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

485 width: Optional[int],

486 include_conda_list: bool,

487 ):

488 return _format_summary(

489 self,

490 target=target,

491 width=width or 100,

492 include_conda_list=include_conda_list,

493 )

494

495

496def _format_summary(

497 summary: ValidationSummary,

498 *,

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

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

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

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

503 include_conda_list: bool,

504 width: int,

505) -> str:

506 parts: List[str] = []

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

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

509 left_out_details: int = 0

510 left_out_details_header = "Left out details"

511

512 def add_part(part: str):

513 parts.append(part)

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

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

516

517 def add_section(header: str):

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

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

520 elif target == "html":

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

522 else:

523 assert_never(target)

524

525 def header_to_tag(header: str):

526 return (

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

528 .replace("(", "")

529 .replace(")", "")

530 .replace(" ", "-")

531 .lower()

532 )

533

534 def add_as_details_below(

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

536 ):

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

538

539 def make_link(header: str):

540 tag = header_to_tag(header)

541 if target == "md":

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

543 elif target == "html":

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

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

546 return f"{header} below"

547 else:

548 assert_never(target)

549

550 for n in range(1, 4):

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

552 if header in details_below:

553 if details_below[header] == text:

554 return make_link(header)

555 else:

556 details_below[header] = text

557 return make_link(header)

558

559 nonlocal left_out_details

560 left_out_details += 1

561 return make_link(left_out_details_header)

562

563 @dataclass

564 class CodeCell:

565 text: str

566

567 @dataclass

568 class CodeRef:

569 text: str

570

571 def format_code(

572 code: str,

573 lang: str = "",

574 title: str = "Details",

575 cell_line_limit: int = 15,

576 cell_width_limit: int = 120,

577 ) -> Union[CodeRef, CodeCell]:

578

579 if not code.strip():

580 return CodeCell("")

581

582 if target == "html":

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

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

585 put_below = (

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

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

588 )

589 else:

590 put_below = True

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

592

593 if put_below:

594 link = add_as_details_below(title, code)

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

596 else:

597 return CodeCell(code)

598

599 def format_traceback(entry: ErrorEntry):

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

601 if entry.traceback_rich is None:

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

603 else:

604 link = add_as_details_below(

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

606 )

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

608

609 if target == "md":

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

611 elif target == "html":

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

613 else:

614 assert_never(target)

615

616 def format_text(text: str):

617 if target == "html":

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

619 else:

620 return text.split("\n")

621

622 def get_info_table():

623 info_rows = [

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

625 ["status", summary.status],

626 ]

627 if not hide_source:

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

629

630 if summary.id is not None:

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

632

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

634 if not hide_env:

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

636

637 if include_conda_list:

638 info_rows.append(

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

640 )

641 return format_table(info_rows)

642

643 def get_details_table():

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

645

646 def append_detail(

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

648 ):

649

650 text_lines = format_text(text)

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

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

653 status_lines[0] = status

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

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

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

657

658 if code is not None:

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

660

661 for d in summary.details:

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

663

664 for entry in d.errors:

665 append_detail(

666 "❌",

667 entry.loc,

668 entry.msg,

669 None if hide_tracebacks else format_traceback(entry),

670 )

671

672 for entry in d.warnings:

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

674

675 if d.recommended_env is not None:

676 rec_env = StringIO()

677 json_env = d.recommended_env.model_dump(

678 mode="json", exclude_defaults=True

679 )

680 assert is_yaml_value(json_env)

681 write_yaml(json_env, rec_env)

682 append_detail(

683 "",

684 d.loc,

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

686 format_code(

687 rec_env.getvalue(),

688 lang="yaml",

689 title="Recommended Conda Environment",

690 ),

691 )

692

693 if d.conda_compare:

694 wrapped_conda_compare = "\n".join(

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

696 )

697 append_detail(

698 "",

699 d.loc,

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

701 format_code(

702 wrapped_conda_compare,

703 title="Conda Environment Comparison",

704 ),

705 )

706

707 return format_table(details)

708

709 add_part(get_info_table())

710 add_part(get_details_table())

711

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

713 add_section(header)

714 if isinstance(text, tuple):

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

716 text, rich_obj = text

717 target.print(rich_obj)

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

719 else:

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

721

722 if left_out_details:

723 parts.append(

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

725 )

726

727 return "".join(parts)

728

729

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

731 """format `rows` as markdown table"""

732 n_cols = len(rows[0])

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

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

735

736 # fix new lines in table cell

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

738

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

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

741 lines.extend(

742 [

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

744 for row in rows[1:]

745 ]

746 )

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

748

749

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

751 """format `rows` as HTML table"""

752

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

754 return (

755 [" <tr>"]

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

757 + [" </tr>"]

758 )

759

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

761 for r in rows[1:]:

762 table.extend(get_line(r))

763

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

765

766 return "\n".join(table)