bioimageio.spec.summary

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