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