bioimageio.spec.summary

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