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 annotated_types
 27import markdown
 28import rich.console
 29import rich.markdown
 30import rich.traceback
 31from loguru import logger
 32from pydantic import (
 33    BaseModel,
 34    Field,
 35    field_serializer,
 36    field_validator,
 37    model_validator,
 38)
 39from pydantic_core.core_schema import ErrorType
 40from typing_extensions import Annotated, Self, assert_never
 41
 42from bioimageio.spec._internal.type_guards import is_dict
 43
 44from ._internal.constants import VERSION
 45from ._internal.io import is_yaml_value
 46from ._internal.io_utils import write_yaml
 47from ._internal.validation_context import ValidationContextSummary
 48from ._internal.warning_levels import (
 49    ALERT,
 50    ALERT_NAME,
 51    ERROR,
 52    ERROR_NAME,
 53    INFO,
 54    INFO_NAME,
 55    WARNING,
 56    WARNING_NAME,
 57    WarningLevel,
 58    WarningSeverity,
 59)
 60from .conda_env import CondaEnv
 61
 62Loc = Tuple[Union[int, str], ...]
 63"""location of error/warning in a nested data structure"""
 64
 65WarningSeverityName = Literal["info", "warning", "alert"]
 66WarningLevelName = Literal[WarningSeverityName, "error"]
 67
 68WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (
 69    MappingProxyType({INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME})
 70)
 71WARNING_LEVEL_TO_NAME: Mapping[WarningLevel, WarningLevelName] = MappingProxyType(
 72    {INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME, ERROR: ERROR_NAME}
 73)
 74WARNING_NAME_TO_LEVEL: Mapping[WarningLevelName, WarningLevel] = MappingProxyType(
 75    {v: k for k, v in WARNING_LEVEL_TO_NAME.items()}
 76)
 77
 78
 79class ValidationEntry(BaseModel):
 80    """Base of `ErrorEntry` and `WarningEntry`"""
 81
 82    loc: Loc
 83    msg: str
 84    type: Union[ErrorType, str]
 85
 86
 87class ErrorEntry(ValidationEntry):
 88    """An error in a `ValidationDetail`"""
 89
 90    with_traceback: bool = False
 91    traceback_md: str = ""
 92    traceback_html: str = ""
 93    # private rich traceback that is not serialized
 94    _traceback_rich: Optional[rich.traceback.Traceback] = None
 95
 96    @property
 97    def traceback_rich(self):
 98        return self._traceback_rich
 99
100    def model_post_init(self, __context: Any):
101        if self.with_traceback and not (self.traceback_md or self.traceback_html):
102            self._traceback_rich = rich.traceback.Traceback()
103            console = rich.console.Console(
104                record=True,
105                file=open(os.devnull, "wt", encoding="utf-8"),
106                color_system="truecolor",
107                width=120,
108                tab_size=4,
109                soft_wrap=True,
110            )
111            console.print(self._traceback_rich)
112            if not self.traceback_md:
113                self.traceback_md = console.export_text(clear=False)
114
115            if not self.traceback_html:
116                self.traceback_html = console.export_html(clear=False)
117
118
119class WarningEntry(ValidationEntry):
120    """A warning in a `ValidationDetail`"""
121
122    severity: WarningSeverity = WARNING
123
124    @property
125    def severity_name(self) -> WarningSeverityName:
126        return WARNING_SEVERITY_TO_NAME[self.severity]
127
128
129def format_loc(
130    loc: Loc, target: Union[Literal["md", "html", "plain"], rich.console.Console]
131) -> str:
132    """helper to format a location tuple **loc**"""
133    loc_str = ".".join(f"({x})" if x[0].isupper() else x for x in map(str, loc))
134
135    # additional field validation can make the location information quite convoluted, e.g.
136    # `weights.pytorch_state_dict.dependencies.source.function-after[validate_url_ok(), url['http','https']]` Input should be a valid URL, relative URL without a base
137    # therefore we remove the `.function-after[validate_url_ok(), url['http','https']]` here
138    loc_str, *_ = loc_str.split(".function-after")
139    if loc_str:
140        if target == "md" or isinstance(target, rich.console.Console):
141            start = "`"
142            end = "`"
143        elif target == "html":
144            start = "<code>"
145            end = "</code>"
146        elif target == "plain":
147            start = ""
148            end = ""
149        else:
150            assert_never(target)
151
152        return f"{start}{loc_str}{end}"
153    else:
154        return ""
155
156
157class InstalledPackage(NamedTuple):
158    name: str
159    version: str
160    build: str = ""
161    channel: str = ""
162
163
164class ValidationDetail(BaseModel, extra="allow"):
165    """a detail in a validation summary"""
166
167    name: str
168    status: Literal["passed", "failed"]
169    loc: Loc = ()
170    """location in the RDF that this detail applies to"""
171    errors: List[ErrorEntry] = Field(  # pyright: ignore[reportUnknownVariableType]
172        default_factory=list
173    )
174    warnings: List[WarningEntry] = Field(  # pyright: ignore[reportUnknownVariableType]
175        default_factory=list
176    )
177    context: Optional[ValidationContextSummary] = None
178
179    recommended_env: Optional[CondaEnv] = None
180    """recommended conda environemnt for this validation detail"""
181
182    saved_conda_compare: Optional[str] = None
183    """output of `conda compare <recommended env>`"""
184
185    @field_serializer("saved_conda_compare")
186    def _save_conda_compare(self, value: Optional[str]):
187        return self.conda_compare
188
189    @model_validator(mode="before")
190    def _load_legacy(cls, data: Any):
191        if is_dict(data):
192            field_name = "conda_compare"
193            if (
194                field_name in data
195                and (saved_field_name := f"saved_{field_name}") not in data
196            ):
197                data[saved_field_name] = data.pop(field_name)
198
199        return data
200
201    @property
202    def conda_compare(self) -> Optional[str]:
203        if self.recommended_env is None:
204            return None
205
206        if self.saved_conda_compare is None:
207            dumped_env = self.recommended_env.model_dump(mode="json")
208            if is_yaml_value(dumped_env):
209                with TemporaryDirectory() as d:
210                    path = Path(d) / "env.yaml"
211                    with path.open("w", encoding="utf-8") as f:
212                        write_yaml(dumped_env, f)
213
214                    compare_proc = subprocess.run(
215                        ["conda", "compare", str(path)],
216                        stdout=subprocess.PIPE,
217                        stderr=subprocess.STDOUT,
218                        shell=True,
219                        text=True,
220                    )
221                    self.saved_conda_compare = (
222                        compare_proc.stdout
223                        or f"`conda compare` exited with {compare_proc.returncode}"
224                    )
225            else:
226                self.saved_conda_compare = (
227                    "Failed to dump recommended env to valid yaml"
228                )
229
230        return self.saved_conda_compare
231
232    @property
233    def status_icon(self):
234        if self.status == "passed":
235            return "✔️"
236        else:
237            return "❌"
238
239
240class ValidationSummary(BaseModel, extra="allow"):
241    """Summarizes output of all bioimageio validations and tests
242    for one specific `ResourceDescr` instance."""
243
244    name: str
245    """Name of the validation"""
246    source_name: str
247    """Source of the validated bioimageio description"""
248    id: Optional[str] = None
249    """ID of the resource being validated"""
250    type: str
251    """Type of the resource being validated"""
252    format_version: str
253    """Format version of the resource being validated"""
254    status: Literal["passed", "valid-format", "failed"]
255    """overall status of the bioimageio validation"""
256    metadata_completeness: Annotated[float, annotated_types.Interval(ge=0, le=1)] = 0.0
257    """Estimate of completeness of the metadata in the resource description.
258
259    Note: This completeness estimate may change with subsequent releases
260        and should be considered bioimageio.spec version specific.
261    """
262
263    details: List[ValidationDetail]
264    """List of validation details"""
265    env: Set[InstalledPackage] = Field(
266        default_factory=lambda: {
267            InstalledPackage(name="bioimageio.spec", version=VERSION)
268        }
269    )
270    """List of selected, relevant package versions"""
271
272    saved_conda_list: Optional[str] = None
273
274    @field_serializer("saved_conda_list")
275    def _save_conda_list(self, value: Optional[str]):
276        return self.conda_list
277
278    @property
279    def conda_list(self):
280        if self.saved_conda_list is None:
281            p = subprocess.run(
282                ["conda", "list"],
283                stdout=subprocess.PIPE,
284                stderr=subprocess.STDOUT,
285                shell=True,
286                text=True,
287            )
288            self.saved_conda_list = (
289                p.stdout or f"`conda list` exited with {p.returncode}"
290            )
291
292        return self.saved_conda_list
293
294    @property
295    def status_icon(self):
296        if self.status == "passed":
297            return "✔️"
298        elif self.status == "valid-format":
299            return "🟡"
300        else:
301            return "❌"
302
303    @property
304    def errors(self) -> List[ErrorEntry]:
305        return list(chain.from_iterable(d.errors for d in self.details))
306
307    @property
308    def warnings(self) -> List[WarningEntry]:
309        return list(chain.from_iterable(d.warnings for d in self.details))
310
311    def format(
312        self,
313        *,
314        width: Optional[int] = None,
315        include_conda_list: bool = False,
316    ):
317        """Format summary as Markdown string"""
318        return self._format(
319            width=width, target="md", include_conda_list=include_conda_list
320        )
321
322    format_md = format
323
324    def format_html(
325        self,
326        *,
327        width: Optional[int] = None,
328        include_conda_list: bool = False,
329    ):
330        md_with_html = self._format(
331            target="html", width=width, include_conda_list=include_conda_list
332        )
333        return markdown.markdown(
334            md_with_html, extensions=["tables", "fenced_code", "nl2br"]
335        )
336
337    # TODO: fix bug which casuses extensive white space between the info table and details table
338    # (the generated markdown seems fine)
339    @no_type_check
340    def display(
341        self,
342        *,
343        width: Optional[int] = None,
344        include_conda_list: bool = False,
345        tab_size: int = 4,
346        soft_wrap: bool = True,
347    ) -> None:
348        try:  # render as HTML in Jupyter notebook
349            from IPython.core.getipython import get_ipython
350            from IPython.display import display_html
351        except ImportError:
352            pass
353        else:
354            if get_ipython() is not None:
355                _ = display_html(
356                    self.format_html(
357                        width=width, include_conda_list=include_conda_list
358                    ),
359                    raw=True,
360                )
361                return
362
363        # render with rich
364        self._format(
365            target=rich.console.Console(
366                width=width,
367                tab_size=tab_size,
368                soft_wrap=soft_wrap,
369            ),
370            width=width,
371            include_conda_list=include_conda_list,
372        )
373
374    def add_detail(self, detail: ValidationDetail):
375        if detail.status == "failed":
376            self.status = "failed"
377        elif detail.status != "passed":
378            assert_never(detail.status)
379
380        self.details.append(detail)
381
382    def log(
383        self,
384        to: Union[Literal["display"], Path, Sequence[Union[Literal["display"], Path]]],
385    ) -> List[Path]:
386        """Convenience method to display the validation summary in the terminal and/or
387        save it to disk. See `save` for details."""
388        if to == "display":
389            display = True
390            save_to = []
391        elif isinstance(to, Path):
392            display = False
393            save_to = [to]
394        else:
395            display = "display" in to
396            save_to = [p for p in to if p != "display"]
397
398        if display:
399            self.display()
400
401        return self.save(save_to)
402
403    def save(
404        self, path: Union[Path, Sequence[Path]] = Path("{id}_summary_{now}")
405    ) -> List[Path]:
406        """Save the validation/test summary in JSON, Markdown or HTML format.
407
408        Returns:
409            List of file paths the summary was saved to.
410
411        Notes:
412        - Format is chosen based on the suffix: `.json`, `.md`, `.html`.
413        - If **path** has no suffix it is assumed to be a direcotry to which a
414          `summary.json`, `summary.md` and `summary.html` are saved to.
415        """
416        if isinstance(path, (str, Path)):
417            path = [Path(path)]
418
419        # folder to file paths
420        file_paths: List[Path] = []
421        for p in path:
422            if p.suffix:
423                file_paths.append(p)
424            else:
425                file_paths.extend(
426                    [
427                        p / "summary.json",
428                        p / "summary.md",
429                        p / "summary.html",
430                    ]
431                )
432
433        now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
434        for p in file_paths:
435            p = Path(str(p).format(id=self.id or "bioimageio", now=now))
436            if p.suffix == ".json":
437                self.save_json(p)
438            elif p.suffix == ".md":
439                self.save_markdown(p)
440            elif p.suffix == ".html":
441                self.save_html(p)
442            else:
443                raise ValueError(f"Unknown summary path suffix '{p.suffix}'")
444
445        return file_paths
446
447    def save_json(
448        self, path: Path = Path("summary.json"), *, indent: Optional[int] = 2
449    ):
450        """Save validation/test summary as JSON file."""
451        json_str = self.model_dump_json(indent=indent)
452        path.parent.mkdir(exist_ok=True, parents=True)
453        _ = path.write_text(json_str, encoding="utf-8")
454        logger.info("Saved summary to {}", path.absolute())
455
456    def save_markdown(self, path: Path = Path("summary.md")):
457        """Save rendered validation/test summary as Markdown file."""
458        formatted = self.format_md()
459        path.parent.mkdir(exist_ok=True, parents=True)
460        _ = path.write_text(formatted, encoding="utf-8")
461        logger.info("Saved Markdown formatted summary to {}", path.absolute())
462
463    def save_html(self, path: Path = Path("summary.html")) -> None:
464        """Save rendered validation/test summary as HTML file."""
465        path.parent.mkdir(exist_ok=True, parents=True)
466
467        html = self.format_html()
468        _ = path.write_text(html, encoding="utf-8")
469        logger.info("Saved HTML formatted summary to {}", path.absolute())
470
471    @classmethod
472    def load_json(cls, path: Path) -> Self:
473        """Load validation/test summary from a suitable JSON file"""
474        json_str = Path(path).read_text(encoding="utf-8")
475        return cls.model_validate_json(json_str)
476
477    @field_validator("env", mode="before")
478    def _convert_dict(cls, value: List[Union[List[str], Dict[str, str]]]):
479        """convert old env value for backwards compatibility"""
480        if isinstance(value, list):
481            return [
482                (
483                    (v["name"], v["version"], v.get("build", ""), v.get("channel", ""))
484                    if isinstance(v, dict) and "name" in v and "version" in v
485                    else v
486                )
487                for v in value
488            ]
489        else:
490            return value
491
492    def _format(
493        self,
494        *,
495        target: Union[rich.console.Console, Literal["html", "md"]],
496        width: Optional[int],
497        include_conda_list: bool,
498    ):
499        return _format_summary(
500            self,
501            target=target,
502            width=width or 100,
503            include_conda_list=include_conda_list,
504        )
505
506
507def _format_summary(
508    summary: ValidationSummary,
509    *,
510    hide_tracebacks: bool = False,  # TODO: remove?
511    hide_source: bool = False,  # TODO: remove?
512    hide_env: bool = False,  # TODO: remove?
513    target: Union[rich.console.Console, Literal["html", "md"]] = "md",
514    include_conda_list: bool,
515    width: int,
516) -> str:
517    parts: List[str] = []
518    format_table = _format_html_table if target == "html" else _format_md_table
519    details_below: Dict[str, Union[str, Tuple[str, rich.traceback.Traceback]]] = {}
520    left_out_details: int = 0
521    left_out_details_header = "Left out details"
522
523    def add_part(part: str):
524        parts.append(part)
525        if isinstance(target, rich.console.Console):
526            target.print(rich.markdown.Markdown(part))
527
528    def add_section(header: str):
529        if target == "md" or isinstance(target, rich.console.Console):
530            add_part(f"\n### {header}\n")
531        elif target == "html":
532            parts.append(f'<h3 id="{header_to_tag(header)}">{header}</h3>')
533        else:
534            assert_never(target)
535
536    def header_to_tag(header: str):
537        return (
538            header.replace("`", "")
539            .replace("(", "")
540            .replace(")", "")
541            .replace(" ", "-")
542            .lower()
543        )
544
545    def add_as_details_below(
546        title: str, text: Union[str, Tuple[str, rich.traceback.Traceback]]
547    ):
548        """returns a header and its tag to link to details below"""
549
550        def make_link(header: str):
551            tag = header_to_tag(header)
552            if target == "md":
553                return f"[{header}](#{tag})"
554            elif target == "html":
555                return f'<a href="#{tag}">{header}</a>'
556            elif isinstance(target, rich.console.Console):
557                return f"{header} below"
558            else:
559                assert_never(target)
560
561        for n in range(1, 4):
562            header = f"{title} {n}"
563            if header in details_below:
564                if details_below[header] == text:
565                    return make_link(header)
566            else:
567                details_below[header] = text
568                return make_link(header)
569
570        nonlocal left_out_details
571        left_out_details += 1
572        return make_link(left_out_details_header)
573
574    @dataclass
575    class CodeCell:
576        text: str
577
578    @dataclass
579    class CodeRef:
580        text: str
581
582    def format_code(
583        code: str,
584        lang: str = "",
585        title: str = "Details",
586        cell_line_limit: int = 15,
587        cell_width_limit: int = 120,
588    ) -> Union[CodeRef, CodeCell]:
589
590        if not code.strip():
591            return CodeCell("")
592
593        if target == "html":
594            html_lang = f' lang="{lang}"' if lang else ""
595            code = f"<pre{html_lang}>{code}</pre>"
596            put_below = (
597                code.count("\n") > cell_line_limit
598                or max(map(len, code.split("\n"))) > cell_width_limit
599            )
600        else:
601            put_below = True
602            code = f"\n```{lang}\n{code}\n```\n"
603
604        if put_below:
605            link = add_as_details_below(title, code)
606            return CodeRef(f"See {link}.")
607        else:
608            return CodeCell(code)
609
610    def format_traceback(entry: ErrorEntry):
611        if isinstance(target, rich.console.Console):
612            if entry.traceback_rich is None:
613                return format_code(entry.traceback_md, title="Traceback")
614            else:
615                link = add_as_details_below(
616                    "Traceback", (entry.traceback_md, entry.traceback_rich)
617                )
618                return CodeRef(f"See {link}.")
619
620        if target == "md":
621            return format_code(entry.traceback_md, title="Traceback")
622        elif target == "html":
623            return format_code(entry.traceback_html, title="Traceback")
624        else:
625            assert_never(target)
626
627    def format_text(text: str):
628        if target == "html":
629            return [f"<pre>{text}</pre>"]
630        else:
631            return text.split("\n")
632
633    def get_info_table():
634        info_rows = [
635            [summary.status_icon, summary.name.strip(".").strip()],
636            ["status", summary.status],
637        ]
638        if not hide_source:
639            info_rows.append(["source", summary.source_name])
640
641        if summary.id is not None:
642            info_rows.append(["id", summary.id])
643
644        info_rows.append(["format version", f"{summary.type} {summary.format_version}"])
645        if not hide_env:
646            info_rows.extend([[e.name, e.version] for e in summary.env])
647
648        if include_conda_list:
649            info_rows.append(
650                ["conda list", format_code(summary.conda_list, title="Conda List").text]
651            )
652        return format_table(info_rows)
653
654    def get_details_table():
655        details = [["", "Location", "Details"]]
656
657        def append_detail(
658            status: str, loc: Loc, text: str, code: Union[CodeRef, CodeCell, None]
659        ):
660
661            text_lines = format_text(text)
662            status_lines = [""] * len(text_lines)
663            loc_lines = [""] * len(text_lines)
664            status_lines[0] = status
665            loc_lines[0] = format_loc(loc, target)
666            for s_line, loc_line, text_line in zip(status_lines, loc_lines, text_lines):
667                details.append([s_line, loc_line, text_line])
668
669            if code is not None:
670                details.append(["", "", code.text])
671
672        for d in summary.details:
673            details.append([d.status_icon, format_loc(d.loc, target), d.name])
674
675            for entry in d.errors:
676                append_detail(
677                    "❌",
678                    entry.loc,
679                    entry.msg,
680                    None if hide_tracebacks else format_traceback(entry),
681                )
682
683            for entry in d.warnings:
684                append_detail("⚠", entry.loc, entry.msg, None)
685
686            if d.recommended_env is not None:
687                rec_env = StringIO()
688                json_env = d.recommended_env.model_dump(
689                    mode="json", exclude_defaults=True
690                )
691                assert is_yaml_value(json_env)
692                write_yaml(json_env, rec_env)
693                append_detail(
694                    "",
695                    d.loc,
696                    f"recommended conda environment ({d.name})",
697                    format_code(
698                        rec_env.getvalue(),
699                        lang="yaml",
700                        title="Recommended Conda Environment",
701                    ),
702                )
703
704            if d.conda_compare:
705                wrapped_conda_compare = "\n".join(
706                    TextWrapper(width=width - 4).wrap(d.conda_compare)
707                )
708                append_detail(
709                    "",
710                    d.loc,
711                    f"conda compare ({d.name})",
712                    format_code(
713                        wrapped_conda_compare,
714                        title="Conda Environment Comparison",
715                    ),
716                )
717
718        return format_table(details)
719
720    add_part(get_info_table())
721    add_part(get_details_table())
722
723    for header, text in details_below.items():
724        add_section(header)
725        if isinstance(text, tuple):
726            assert isinstance(target, rich.console.Console)
727            text, rich_obj = text
728            target.print(rich_obj)
729            parts.append(f"{text}\n")
730        else:
731            add_part(f"{text}\n")
732
733    if left_out_details:
734        parts.append(
735            f"\n{left_out_details_header}\nLeft out {left_out_details} more details for brevity.\n"
736        )
737
738    return "".join(parts)
739
740
741def _format_md_table(rows: List[List[str]]) -> str:
742    """format `rows` as markdown table"""
743    n_cols = len(rows[0])
744    assert all(len(row) == n_cols for row in rows)
745    col_widths = [max(max(len(row[i]) for row in rows), 3) for i in range(n_cols)]
746
747    # fix new lines in table cell
748    rows = [[line.replace("\n", "<br>") for line in r] for r in rows]
749
750    lines = [" | ".join(rows[0][i].center(col_widths[i]) for i in range(n_cols))]
751    lines.append(" | ".join("---".center(col_widths[i]) for i in range(n_cols)))
752    lines.extend(
753        [
754            " | ".join(row[i].ljust(col_widths[i]) for i in range(n_cols))
755            for row in rows[1:]
756        ]
757    )
758    return "\n| " + " |\n| ".join(lines) + " |\n"
759
760
761def _format_html_table(rows: List[List[str]]) -> str:
762    """format `rows` as HTML table"""
763
764    def get_line(cells: List[str], cell_tag: Literal["th", "td"] = "td"):
765        return (
766            ["  <tr>"]
767            + [f"    <{cell_tag}>{c}</{cell_tag}>" for c in cells]
768            + ["  </tr>"]
769        )
770
771    table = ["<table>"] + get_line(rows[0], cell_tag="th")
772    for r in rows[1:]:
773        table.extend(get_line(r))
774
775    table.append("</table>")
776
777    return "\n".join(table)