Coverage for src / bioimageio / spec / summary.py: 66%
386 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-08 13:04 +0000
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-08 13:04 +0000
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) -> Optional[rich.traceback.Traceback]:
100 return self._traceback_rich
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)
117 if not self.traceback_html:
118 self.traceback_html = console.export_html(clear=False)
121class WarningEntry(ValidationEntry):
122 """A warning in a `ValidationDetail`"""
124 severity: WarningSeverity = WARNING
126 @property
127 def severity_name(self) -> WarningSeverityName:
128 return WARNING_SEVERITY_TO_NAME[self.severity]
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))
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)
154 return f"{start}{loc_str}{end}"
155 else:
156 return ""
159class InstalledPackage(NamedTuple):
160 name: str
161 version: str
162 build: str = ""
163 channel: str = ""
166class ValidationDetail(BaseModel, extra="allow"):
167 """a detail in a validation summary"""
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
181 recommended_env: Optional[CondaEnv] = None
182 """recommended conda environemnt for this validation detail"""
184 saved_conda_compare: Optional[str] = None
185 """output of `conda compare <recommended env>`"""
187 @field_serializer("saved_conda_compare")
188 def _save_conda_compare(self, value: Optional[str]):
189 return self.conda_compare
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)
201 return data
203 @property
204 def conda_compare(self) -> Optional[str]:
205 if self.recommended_env is None:
206 return None
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)
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 )
232 return self.saved_conda_compare
234 @property
235 def status_icon(self) -> str:
236 if self.status == "passed":
237 return "✔️"
238 else:
239 return "❌"
242class ValidationSummary(BaseModel, extra="allow"):
243 """Summarizes output of all bioimageio validations and tests
244 for one specific `ResourceDescr` instance."""
246 name: str
247 """Name of the validation"""
249 source_name: str
250 """Source of the validated bioimageio description"""
252 id: Optional[str] = None
253 """ID of the resource being validated"""
255 type: str
256 """Type of the resource being validated"""
258 format_version: str
259 """Format version of the resource being validated"""
261 status: Literal["passed", "valid-format", "failed"]
262 """Overall status of the bioimageio validation"""
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.
267 Note: This completeness estimate may change with subsequent releases
268 and should be considered bioimageio.spec version specific.
269 """
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"""
283 saved_conda_list: Optional[str] = None
285 @field_serializer("saved_conda_list")
286 def _save_conda_list(self, value: Optional[str]):
287 return self.conda_list
289 @property
290 def conda_list(self) -> str:
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 )
303 return self.saved_conda_list
305 @property
306 def status_icon(self) -> str:
307 if self.status == "passed":
308 return "✔️"
309 elif self.status == "valid-format":
310 return "🟡"
311 else:
312 return "❌"
314 @property
315 def errors(self) -> List[ErrorEntry]:
316 return list(chain.from_iterable(d.errors for d in self.details))
318 @property
319 def warnings(self) -> List[WarningEntry]:
320 return list(chain.from_iterable(d.warnings for d in self.details))
322 def format(
323 self,
324 *,
325 width: Optional[int] = None,
326 include_conda_list: bool = False,
327 ) -> str:
328 """Format summary as Markdown string (alias to `format_md`)"""
329 return self.format_md(width=width, include_conda_list=include_conda_list)
331 def format_md(
332 self,
333 *,
334 width: Optional[int] = None,
335 include_conda_list: bool = False,
336 ) -> str:
337 """Format summary as Markdown string"""
338 return self._format(
339 width=width, target="md", include_conda_list=include_conda_list
340 )
342 def format_html(
343 self,
344 *,
345 width: Optional[int] = None,
346 include_conda_list: bool = False,
347 ) -> str:
348 md_with_html = self._format(
349 target="html", width=width, include_conda_list=include_conda_list
350 )
351 return markdown.markdown(
352 md_with_html, extensions=["tables", "fenced_code", "nl2br"]
353 )
355 def display(
356 self,
357 *,
358 width: Optional[int] = None,
359 include_conda_list: bool = False,
360 tab_size: int = 4,
361 soft_wrap: bool = True,
362 ) -> None:
363 try: # render as HTML in Jupyter notebook
364 from IPython.core.getipython import get_ipython
365 from IPython.display import (
366 display_html, # pyright: ignore[reportUnknownVariableType]
367 )
368 except ImportError:
369 pass
370 else:
371 if get_ipython() is not None:
372 _ = display_html(
373 self.format_html(
374 width=width, include_conda_list=include_conda_list
375 ),
376 raw=True,
377 )
378 return
380 # render with rich
381 _ = self._format(
382 target=rich.console.Console(
383 width=width,
384 tab_size=tab_size,
385 soft_wrap=soft_wrap,
386 ),
387 width=width,
388 include_conda_list=include_conda_list,
389 )
391 def add_detail(self, detail: ValidationDetail, update_status: bool = True) -> None:
392 if update_status:
393 if self.status == "valid-format" and detail.status == "passed":
394 # once status is 'valid-format' we can only improve to 'passed'
395 self.status = "passed"
396 elif self.status == "passed" and detail.status == "failed":
397 # once status is 'passed' it can only degrade to 'valid-format'
398 self.status = "valid-format"
399 # once format is 'failed' it cannot improve
401 self.details.append(detail)
403 def log(
404 self,
405 to: Union[Literal["display"], Path, Sequence[Union[Literal["display"], Path]]],
406 ) -> List[Path]:
407 """Convenience method to display the validation summary in the terminal and/or
408 save it to disk. See `save` for details."""
409 if to == "display":
410 display = True
411 save_to = []
412 elif isinstance(to, Path):
413 display = False
414 save_to = [to]
415 else:
416 display = "display" in to
417 save_to = [p for p in to if p != "display"]
419 if display:
420 self.display()
422 return self.save(save_to)
424 def save(
425 self, path: Union[Path, Sequence[Path]] = Path("{id}_summary_{now}")
426 ) -> List[Path]:
427 """Save the validation/test summary in JSON, Markdown or HTML format.
429 Returns:
430 List of file paths the summary was saved to.
432 Notes:
433 - Format is chosen based on the suffix: `.json`, `.md`, `.html`.
434 - If **path** has no suffix it is assumed to be a direcotry to which a
435 `summary.json`, `summary.md` and `summary.html` are saved to.
436 """
437 if isinstance(path, (str, Path)):
438 path = [Path(path)]
440 # folder to file paths
441 file_paths: List[Path] = []
442 for p in path:
443 if p.suffix:
444 file_paths.append(p)
445 else:
446 file_paths.extend(
447 [
448 p / "summary.json",
449 p / "summary.md",
450 p / "summary.html",
451 ]
452 )
454 now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
455 for p in file_paths:
456 p = Path(str(p).format(id=self.id or "bioimageio", now=now))
457 if p.suffix == ".json":
458 self.save_json(p)
459 elif p.suffix == ".md":
460 self.save_markdown(p)
461 elif p.suffix == ".html":
462 self.save_html(p)
463 else:
464 raise ValueError(f"Unknown summary path suffix '{p.suffix}'")
466 return file_paths
468 def save_json(
469 self, path: Path = Path("summary.json"), *, indent: Optional[int] = 2
470 ) -> None:
471 """Save validation/test summary as JSON file."""
472 json_str = self.model_dump_json(indent=indent)
473 path.parent.mkdir(exist_ok=True, parents=True)
474 _ = path.write_text(json_str, encoding="utf-8")
475 logger.info("Saved summary to {}", path.absolute())
477 def save_markdown(self, path: Path = Path("summary.md")) -> None:
478 """Save rendered validation/test summary as Markdown file."""
479 formatted = self.format_md()
480 path.parent.mkdir(exist_ok=True, parents=True)
481 _ = path.write_text(formatted, encoding="utf-8")
482 logger.info("Saved Markdown formatted summary to {}", path.absolute())
484 def save_html(self, path: Path = Path("summary.html")) -> None:
485 """Save rendered validation/test summary as HTML file."""
486 path.parent.mkdir(exist_ok=True, parents=True)
488 html = self.format_html()
489 _ = path.write_text(html, encoding="utf-8")
490 logger.info("Saved HTML formatted summary to {}", path.absolute())
492 @classmethod
493 def load_json(cls, path: Path) -> Self:
494 """Load validation/test summary from a suitable JSON file"""
495 json_str = Path(path).read_text(encoding="utf-8")
496 return cls.model_validate_json(json_str)
498 @field_validator("env", mode="before")
499 def _convert_old_env(cls, value: List[Union[List[str], Dict[str, str]]]):
500 """convert old style dict values of `env` for backwards compatibility"""
501 if isinstance(value, list):
502 return [
503 (
504 (v["name"], v["version"], v.get("build", ""), v.get("channel", ""))
505 if isinstance(v, dict) and "name" in v and "version" in v
506 else v
507 )
508 for v in value
509 ]
510 else:
511 return value
513 def _format(
514 self,
515 *,
516 target: Union[rich.console.Console, Literal["html", "md"]],
517 width: Optional[int],
518 include_conda_list: bool,
519 ) -> str:
520 return _format_summary(
521 self,
522 target=target,
523 width=width or 100,
524 include_conda_list=include_conda_list,
525 )
528def _format_summary(
529 summary: ValidationSummary,
530 *,
531 hide_tracebacks: bool = False, # TODO: remove?
532 hide_source: bool = False, # TODO: remove?
533 hide_env: bool = False, # TODO: remove?
534 target: Union[rich.console.Console, Literal["html", "md"]] = "md",
535 include_conda_list: bool,
536 width: int,
537) -> str:
538 parts: List[str] = []
539 format_table = _format_html_table if target == "html" else _format_md_table
540 details_below: Dict[str, Union[str, Tuple[str, rich.traceback.Traceback]]] = {}
541 left_out_details: int = 0
542 left_out_details_header = "Left out details"
544 def add_part(part: str):
545 parts.append(part)
546 if isinstance(target, rich.console.Console):
547 target.print(rich.markdown.Markdown(part))
549 def add_section(header: str):
550 if target == "md" or isinstance(target, rich.console.Console):
551 add_part(f"\n### {header}\n")
552 elif target == "html":
553 parts.append(f'<h3 id="{header_to_tag(header)}">{header}</h3>')
554 else:
555 assert_never(target)
557 def header_to_tag(header: str):
558 return (
559 header.replace("`", "")
560 .replace("(", "")
561 .replace(")", "")
562 .replace(" ", "-")
563 .lower()
564 )
566 def add_as_details_below(
567 title: str, text: Union[str, Tuple[str, rich.traceback.Traceback]]
568 ):
569 """returns a header and its tag to link to details below"""
571 def make_link(header: str):
572 tag = header_to_tag(header)
573 if target == "md":
574 return f"[{header}](#{tag})"
575 elif target == "html":
576 return f'<a href="#{tag}">{header}</a>'
577 elif isinstance(target, rich.console.Console):
578 return f"{header} below"
579 else:
580 assert_never(target)
582 for n in range(1, 4):
583 header = f"{title} {n}"
584 if header in details_below:
585 if details_below[header] == text:
586 return make_link(header)
587 else:
588 details_below[header] = text
589 return make_link(header)
591 nonlocal left_out_details
592 left_out_details += 1
593 return make_link(left_out_details_header)
595 @dataclass
596 class CodeCell:
597 text: str
599 @dataclass
600 class CodeRef:
601 text: str
603 def format_code(
604 code: str,
605 lang: str = "",
606 title: str = "Details",
607 cell_line_limit: int = 15,
608 cell_width_limit: int = 120,
609 ) -> Union[CodeRef, CodeCell]:
610 if not code.strip():
611 return CodeCell("")
613 if target == "html":
614 html_lang = f' lang="{lang}"' if lang else ""
615 code = f"<pre{html_lang}>{code}</pre>"
616 put_below = (
617 code.count("\n") > cell_line_limit
618 or max(map(len, code.split("\n"))) > cell_width_limit
619 )
620 else:
621 put_below = True
622 code = f"\n```{lang}\n{code}\n```\n"
624 if put_below:
625 link = add_as_details_below(title, code)
626 return CodeRef(f"See {link}.")
627 else:
628 return CodeCell(code)
630 def format_traceback(entry: ErrorEntry):
631 if isinstance(target, rich.console.Console):
632 if entry.traceback_rich is None:
633 return format_code(entry.traceback_md, title="Traceback")
634 else:
635 link = add_as_details_below(
636 "Traceback", (entry.traceback_md, entry.traceback_rich)
637 )
638 return CodeRef(f"See {link}.")
640 if target == "md":
641 return format_code(entry.traceback_md, title="Traceback")
642 elif target == "html":
643 return format_code(entry.traceback_html, title="Traceback")
644 else:
645 assert_never(target)
647 def format_text(text: str):
648 if target == "html":
649 return [f"<pre>{text}</pre>"]
650 else:
651 return text.split("\n")
653 def get_info_table():
654 info_rows = [
655 [summary.status_icon, summary.name.strip(".").strip()],
656 ["status", summary.status],
657 ]
658 if not hide_source:
659 info_rows.append(["source", html.escape(summary.source_name)])
661 if summary.id is not None:
662 info_rows.append(["id", summary.id])
664 info_rows.append(["format version", f"{summary.type} {summary.format_version}"])
665 if not hide_env:
666 info_rows.extend([[e.name, e.version] for e in sorted(summary.env)])
668 if include_conda_list:
669 info_rows.append(
670 ["conda list", format_code(summary.conda_list, title="Conda List").text]
671 )
672 return format_table(info_rows)
674 def get_details_table():
675 details = [["", "Location", "Details"]]
677 def append_detail(
678 status: str, loc: Loc, text: str, code: Union[CodeRef, CodeCell, None]
679 ):
680 text_lines = format_text(text)
681 status_lines = [""] * len(text_lines)
682 loc_lines = [""] * len(text_lines)
683 status_lines[0] = status
684 loc_lines[0] = format_loc(loc, target)
685 for s_line, loc_line, text_line in zip(status_lines, loc_lines, text_lines):
686 details.append([s_line, loc_line, text_line])
688 if code is not None:
689 details.append(["", "", code.text])
691 for d in summary.details:
692 details.append([d.status_icon, format_loc(d.loc, target), d.name])
694 for entry in d.errors:
695 append_detail(
696 "❌",
697 entry.loc,
698 entry.msg,
699 None if hide_tracebacks else format_traceback(entry),
700 )
702 for entry in d.warnings:
703 append_detail("⚠", entry.loc, entry.msg, None)
705 if d.recommended_env is not None:
706 rec_env = StringIO()
707 json_env = d.recommended_env.model_dump(
708 mode="json", exclude_defaults=True
709 )
710 assert is_yaml_value(json_env)
711 write_yaml(json_env, rec_env)
712 append_detail(
713 "",
714 d.loc,
715 f"recommended conda environment ({d.name})",
716 format_code(
717 rec_env.getvalue(),
718 lang="yaml",
719 title="Recommended Conda Environment",
720 ),
721 )
723 if d.conda_compare:
724 wrapped_conda_compare = "\n".join(
725 TextWrapper(width=width - 4).wrap(d.conda_compare)
726 )
727 append_detail(
728 "",
729 d.loc,
730 f"conda compare ({d.name})",
731 format_code(
732 wrapped_conda_compare,
733 title="Conda Environment Comparison",
734 ),
735 )
737 return format_table(details)
739 add_part(get_info_table())
740 add_part(get_details_table())
742 for header, text in details_below.items():
743 add_section(header)
744 if isinstance(text, tuple):
745 assert isinstance(target, rich.console.Console)
746 text, rich_obj = text
747 target.print(rich_obj)
748 parts.append(f"{text}\n")
749 else:
750 add_part(f"{text}\n")
752 if left_out_details:
753 parts.append(
754 f"\n{left_out_details_header}\nLeft out {left_out_details} more details for brevity.\n"
755 )
757 return "".join(parts)
760def _format_md_table(rows: List[List[str]]) -> str:
761 """format `rows` as markdown table"""
762 n_cols = len(rows[0])
763 assert all(len(row) == n_cols for row in rows)
764 col_widths = [max(max(len(row[i]) for row in rows), 3) for i in range(n_cols)]
766 # fix new lines in table cell
767 rows = [[line.replace("\n", "<br>") for line in r] for r in rows]
769 lines = [" | ".join(rows[0][i].center(col_widths[i]) for i in range(n_cols))]
770 lines.append(" | ".join("---".center(col_widths[i]) for i in range(n_cols)))
771 lines.extend(
772 [
773 " | ".join(row[i].ljust(col_widths[i]) for i in range(n_cols))
774 for row in rows[1:]
775 ]
776 )
777 return "\n| " + " |\n| ".join(lines) + " |\n"
780def _format_html_table(rows: List[List[str]]) -> str:
781 """format `rows` as HTML table"""
783 def get_line(cells: List[str], cell_tag: Literal["th", "td"] = "td"):
784 return (
785 [" <tr>"]
786 + [f" <{cell_tag}>{c}</{cell_tag}>" for c in cells]
787 + [" </tr>"]
788 )
790 table = ["<table>"] + get_line(rows[0], cell_tag="th")
791 for r in rows[1:]:
792 table.extend(get_line(r))
794 table.append("</table>")
796 return "\n".join(table)