bioimageio.spec.summary

  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)
 26
 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
 42
 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
 61
 62CONDA_CMD = "conda.bat" if platform.system() == "Windows" else "conda"
 63
 64Loc = Tuple[Union[int, str], ...]
 65"""location of error/warning in a nested data structure"""
 66
 67WarningSeverityName = Literal["info", "warning", "alert"]
 68WarningLevelName = Literal[WarningSeverityName, "error"]
 69
 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)
 79
 80
 81class ValidationEntry(BaseModel):
 82    """Base of `ErrorEntry` and `WarningEntry`"""
 83
 84    loc: Loc
 85    msg: str
 86    type: Union[ErrorType, str]
 87
 88
 89class ErrorEntry(ValidationEntry):
 90    """An error in a `ValidationDetail`"""
 91
 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
 97
 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
249    source_name: str
250    """Source of the validated bioimageio description"""
251
252    id: Optional[str] = None
253    """ID of the resource being validated"""
254
255    type: str
256    """Type of the resource being validated"""
257
258    format_version: str
259    """Format version of the resource being validated"""
260
261    status: Literal["passed", "valid-format", "failed"]
262    """Overall status of the bioimageio validation"""
263
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.
266
267    Note: This completeness estimate may change with subsequent releases
268        and should be considered bioimageio.spec version specific.
269    """
270
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"""
282
283    saved_conda_list: Optional[str] = None
284
285    @field_serializer("saved_conda_list")
286    def _save_conda_list(self, value: Optional[str]):
287        return self.conda_list
288
289    @property
290    def conda_list(self):
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            )
302
303        return self.saved_conda_list
304
305    @property
306    def status_icon(self):
307        if self.status == "passed":
308            return "✔️"
309        elif self.status == "valid-format":
310            return "🟡"
311        else:
312            return "❌"
313
314    @property
315    def errors(self) -> List[ErrorEntry]:
316        return list(chain.from_iterable(d.errors for d in self.details))
317
318    @property
319    def warnings(self) -> List[WarningEntry]:
320        return list(chain.from_iterable(d.warnings for d in self.details))
321
322    def format(
323        self,
324        *,
325        width: Optional[int] = None,
326        include_conda_list: bool = False,
327    ):
328        """Format summary as Markdown string"""
329        return self._format(
330            width=width, target="md", include_conda_list=include_conda_list
331        )
332
333    format_md = format
334
335    def format_html(
336        self,
337        *,
338        width: Optional[int] = None,
339        include_conda_list: bool = False,
340    ):
341        md_with_html = self._format(
342            target="html", width=width, include_conda_list=include_conda_list
343        )
344        return markdown.markdown(
345            md_with_html, extensions=["tables", "fenced_code", "nl2br"]
346        )
347
348    def display(
349        self,
350        *,
351        width: Optional[int] = None,
352        include_conda_list: bool = False,
353        tab_size: int = 4,
354        soft_wrap: bool = True,
355    ) -> None:
356        try:  # render as HTML in Jupyter notebook
357            from IPython.core.getipython import get_ipython
358            from IPython.display import (
359                display_html,  # pyright: ignore[reportUnknownVariableType]
360            )
361        except ImportError:
362            pass
363        else:
364            if get_ipython() is not None:
365                _ = display_html(
366                    self.format_html(
367                        width=width, include_conda_list=include_conda_list
368                    ),
369                    raw=True,
370                )
371                return
372
373        # render with rich
374        _ = self._format(
375            target=rich.console.Console(
376                width=width,
377                tab_size=tab_size,
378                soft_wrap=soft_wrap,
379            ),
380            width=width,
381            include_conda_list=include_conda_list,
382        )
383
384    def add_detail(self, detail: ValidationDetail, update_status: bool = True):
385        if update_status:
386            if self.status == "valid-format" and detail.status == "passed":
387                # once status is 'valid-format' we can only improve to 'passed'
388                self.status = "passed"
389            elif self.status == "passed" and detail.status == "failed":
390                # once status is 'passed' it can only degrade to 'valid-format'
391                self.status = "valid-format"
392            # once format is 'failed' it cannot improve
393
394        self.details.append(detail)
395
396    def log(
397        self,
398        to: Union[Literal["display"], Path, Sequence[Union[Literal["display"], Path]]],
399    ) -> List[Path]:
400        """Convenience method to display the validation summary in the terminal and/or
401        save it to disk. See `save` for details."""
402        if to == "display":
403            display = True
404            save_to = []
405        elif isinstance(to, Path):
406            display = False
407            save_to = [to]
408        else:
409            display = "display" in to
410            save_to = [p for p in to if p != "display"]
411
412        if display:
413            self.display()
414
415        return self.save(save_to)
416
417    def save(
418        self, path: Union[Path, Sequence[Path]] = Path("{id}_summary_{now}")
419    ) -> List[Path]:
420        """Save the validation/test summary in JSON, Markdown or HTML format.
421
422        Returns:
423            List of file paths the summary was saved to.
424
425        Notes:
426        - Format is chosen based on the suffix: `.json`, `.md`, `.html`.
427        - If **path** has no suffix it is assumed to be a direcotry to which a
428          `summary.json`, `summary.md` and `summary.html` are saved to.
429        """
430        if isinstance(path, (str, Path)):
431            path = [Path(path)]
432
433        # folder to file paths
434        file_paths: List[Path] = []
435        for p in path:
436            if p.suffix:
437                file_paths.append(p)
438            else:
439                file_paths.extend(
440                    [
441                        p / "summary.json",
442                        p / "summary.md",
443                        p / "summary.html",
444                    ]
445                )
446
447        now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
448        for p in file_paths:
449            p = Path(str(p).format(id=self.id or "bioimageio", now=now))
450            if p.suffix == ".json":
451                self.save_json(p)
452            elif p.suffix == ".md":
453                self.save_markdown(p)
454            elif p.suffix == ".html":
455                self.save_html(p)
456            else:
457                raise ValueError(f"Unknown summary path suffix '{p.suffix}'")
458
459        return file_paths
460
461    def save_json(
462        self, path: Path = Path("summary.json"), *, indent: Optional[int] = 2
463    ):
464        """Save validation/test summary as JSON file."""
465        json_str = self.model_dump_json(indent=indent)
466        path.parent.mkdir(exist_ok=True, parents=True)
467        _ = path.write_text(json_str, encoding="utf-8")
468        logger.info("Saved summary to {}", path.absolute())
469
470    def save_markdown(self, path: Path = Path("summary.md")):
471        """Save rendered validation/test summary as Markdown file."""
472        formatted = self.format_md()
473        path.parent.mkdir(exist_ok=True, parents=True)
474        _ = path.write_text(formatted, encoding="utf-8")
475        logger.info("Saved Markdown formatted summary to {}", path.absolute())
476
477    def save_html(self, path: Path = Path("summary.html")) -> None:
478        """Save rendered validation/test summary as HTML file."""
479        path.parent.mkdir(exist_ok=True, parents=True)
480
481        html = self.format_html()
482        _ = path.write_text(html, encoding="utf-8")
483        logger.info("Saved HTML formatted summary to {}", path.absolute())
484
485    @classmethod
486    def load_json(cls, path: Path) -> Self:
487        """Load validation/test summary from a suitable JSON file"""
488        json_str = Path(path).read_text(encoding="utf-8")
489        return cls.model_validate_json(json_str)
490
491    @field_validator("env", mode="before")
492    def _convert_dict(cls, value: List[Union[List[str], Dict[str, str]]]):
493        """convert old env value for backwards compatibility"""
494        if isinstance(value, list):
495            return [
496                (
497                    (v["name"], v["version"], v.get("build", ""), v.get("channel", ""))
498                    if isinstance(v, dict) and "name" in v and "version" in v
499                    else v
500                )
501                for v in value
502            ]
503        else:
504            return value
505
506    def _format(
507        self,
508        *,
509        target: Union[rich.console.Console, Literal["html", "md"]],
510        width: Optional[int],
511        include_conda_list: bool,
512    ):
513        return _format_summary(
514            self,
515            target=target,
516            width=width or 100,
517            include_conda_list=include_conda_list,
518        )
519
520
521def _format_summary(
522    summary: ValidationSummary,
523    *,
524    hide_tracebacks: bool = False,  # TODO: remove?
525    hide_source: bool = False,  # TODO: remove?
526    hide_env: bool = False,  # TODO: remove?
527    target: Union[rich.console.Console, Literal["html", "md"]] = "md",
528    include_conda_list: bool,
529    width: int,
530) -> str:
531    parts: List[str] = []
532    format_table = _format_html_table if target == "html" else _format_md_table
533    details_below: Dict[str, Union[str, Tuple[str, rich.traceback.Traceback]]] = {}
534    left_out_details: int = 0
535    left_out_details_header = "Left out details"
536
537    def add_part(part: str):
538        parts.append(part)
539        if isinstance(target, rich.console.Console):
540            target.print(rich.markdown.Markdown(part))
541
542    def add_section(header: str):
543        if target == "md" or isinstance(target, rich.console.Console):
544            add_part(f"\n### {header}\n")
545        elif target == "html":
546            parts.append(f'<h3 id="{header_to_tag(header)}">{header}</h3>')
547        else:
548            assert_never(target)
549
550    def header_to_tag(header: str):
551        return (
552            header.replace("`", "")
553            .replace("(", "")
554            .replace(")", "")
555            .replace(" ", "-")
556            .lower()
557        )
558
559    def add_as_details_below(
560        title: str, text: Union[str, Tuple[str, rich.traceback.Traceback]]
561    ):
562        """returns a header and its tag to link to details below"""
563
564        def make_link(header: str):
565            tag = header_to_tag(header)
566            if target == "md":
567                return f"[{header}](#{tag})"
568            elif target == "html":
569                return f'<a href="#{tag}">{header}</a>'
570            elif isinstance(target, rich.console.Console):
571                return f"{header} below"
572            else:
573                assert_never(target)
574
575        for n in range(1, 4):
576            header = f"{title} {n}"
577            if header in details_below:
578                if details_below[header] == text:
579                    return make_link(header)
580            else:
581                details_below[header] = text
582                return make_link(header)
583
584        nonlocal left_out_details
585        left_out_details += 1
586        return make_link(left_out_details_header)
587
588    @dataclass
589    class CodeCell:
590        text: str
591
592    @dataclass
593    class CodeRef:
594        text: str
595
596    def format_code(
597        code: str,
598        lang: str = "",
599        title: str = "Details",
600        cell_line_limit: int = 15,
601        cell_width_limit: int = 120,
602    ) -> Union[CodeRef, CodeCell]:
603        if not code.strip():
604            return CodeCell("")
605
606        if target == "html":
607            html_lang = f' lang="{lang}"' if lang else ""
608            code = f"<pre{html_lang}>{code}</pre>"
609            put_below = (
610                code.count("\n") > cell_line_limit
611                or max(map(len, code.split("\n"))) > cell_width_limit
612            )
613        else:
614            put_below = True
615            code = f"\n```{lang}\n{code}\n```\n"
616
617        if put_below:
618            link = add_as_details_below(title, code)
619            return CodeRef(f"See {link}.")
620        else:
621            return CodeCell(code)
622
623    def format_traceback(entry: ErrorEntry):
624        if isinstance(target, rich.console.Console):
625            if entry.traceback_rich is None:
626                return format_code(entry.traceback_md, title="Traceback")
627            else:
628                link = add_as_details_below(
629                    "Traceback", (entry.traceback_md, entry.traceback_rich)
630                )
631                return CodeRef(f"See {link}.")
632
633        if target == "md":
634            return format_code(entry.traceback_md, title="Traceback")
635        elif target == "html":
636            return format_code(entry.traceback_html, title="Traceback")
637        else:
638            assert_never(target)
639
640    def format_text(text: str):
641        if target == "html":
642            return [f"<pre>{text}</pre>"]
643        else:
644            return text.split("\n")
645
646    def get_info_table():
647        info_rows = [
648            [summary.status_icon, summary.name.strip(".").strip()],
649            ["status", summary.status],
650        ]
651        if not hide_source:
652            info_rows.append(["source", html.escape(summary.source_name)])
653
654        if summary.id is not None:
655            info_rows.append(["id", summary.id])
656
657        info_rows.append(["format version", f"{summary.type} {summary.format_version}"])
658        if not hide_env:
659            info_rows.extend([[e.name, e.version] for e in sorted(summary.env)])
660
661        if include_conda_list:
662            info_rows.append(
663                ["conda list", format_code(summary.conda_list, title="Conda List").text]
664            )
665        return format_table(info_rows)
666
667    def get_details_table():
668        details = [["", "Location", "Details"]]
669
670        def append_detail(
671            status: str, loc: Loc, text: str, code: Union[CodeRef, CodeCell, None]
672        ):
673            text_lines = format_text(text)
674            status_lines = [""] * len(text_lines)
675            loc_lines = [""] * len(text_lines)
676            status_lines[0] = status
677            loc_lines[0] = format_loc(loc, target)
678            for s_line, loc_line, text_line in zip(status_lines, loc_lines, text_lines):
679                details.append([s_line, loc_line, text_line])
680
681            if code is not None:
682                details.append(["", "", code.text])
683
684        for d in summary.details:
685            details.append([d.status_icon, format_loc(d.loc, target), d.name])
686
687            for entry in d.errors:
688                append_detail(
689                    "❌",
690                    entry.loc,
691                    entry.msg,
692                    None if hide_tracebacks else format_traceback(entry),
693                )
694
695            for entry in d.warnings:
696                append_detail("⚠", entry.loc, entry.msg, None)
697
698            if d.recommended_env is not None:
699                rec_env = StringIO()
700                json_env = d.recommended_env.model_dump(
701                    mode="json", exclude_defaults=True
702                )
703                assert is_yaml_value(json_env)
704                write_yaml(json_env, rec_env)
705                append_detail(
706                    "",
707                    d.loc,
708                    f"recommended conda environment ({d.name})",
709                    format_code(
710                        rec_env.getvalue(),
711                        lang="yaml",
712                        title="Recommended Conda Environment",
713                    ),
714                )
715
716            if d.conda_compare:
717                wrapped_conda_compare = "\n".join(
718                    TextWrapper(width=width - 4).wrap(d.conda_compare)
719                )
720                append_detail(
721                    "",
722                    d.loc,
723                    f"conda compare ({d.name})",
724                    format_code(
725                        wrapped_conda_compare,
726                        title="Conda Environment Comparison",
727                    ),
728                )
729
730        return format_table(details)
731
732    add_part(get_info_table())
733    add_part(get_details_table())
734
735    for header, text in details_below.items():
736        add_section(header)
737        if isinstance(text, tuple):
738            assert isinstance(target, rich.console.Console)
739            text, rich_obj = text
740            target.print(rich_obj)
741            parts.append(f"{text}\n")
742        else:
743            add_part(f"{text}\n")
744
745    if left_out_details:
746        parts.append(
747            f"\n{left_out_details_header}\nLeft out {left_out_details} more details for brevity.\n"
748        )
749
750    return "".join(parts)
751
752
753def _format_md_table(rows: List[List[str]]) -> str:
754    """format `rows` as markdown table"""
755    n_cols = len(rows[0])
756    assert all(len(row) == n_cols for row in rows)
757    col_widths = [max(max(len(row[i]) for row in rows), 3) for i in range(n_cols)]
758
759    # fix new lines in table cell
760    rows = [[line.replace("\n", "<br>") for line in r] for r in rows]
761
762    lines = [" | ".join(rows[0][i].center(col_widths[i]) for i in range(n_cols))]
763    lines.append(" | ".join("---".center(col_widths[i]) for i in range(n_cols)))
764    lines.extend(
765        [
766            " | ".join(row[i].ljust(col_widths[i]) for i in range(n_cols))
767            for row in rows[1:]
768        ]
769    )
770    return "\n| " + " |\n| ".join(lines) + " |\n"
771
772
773def _format_html_table(rows: List[List[str]]) -> str:
774    """format `rows` as HTML table"""
775
776    def get_line(cells: List[str], cell_tag: Literal["th", "td"] = "td"):
777        return (
778            ["  <tr>"]
779            + [f"    <{cell_tag}>{c}</{cell_tag}>" for c in cells]
780            + ["  </tr>"]
781        )
782
783    table = ["<table>"] + get_line(rows[0], cell_tag="th")
784    for r in rows[1:]:
785        table.extend(get_line(r))
786
787    table.append("</table>")
788
789    return "\n".join(table)