bioimageio.spec.summary

  1import subprocess
  2from io import StringIO
  3from itertools import chain
  4from pathlib import Path
  5from tempfile import TemporaryDirectory
  6from types import MappingProxyType
  7from typing import (
  8    Any,
  9    Dict,
 10    Iterable,
 11    List,
 12    Literal,
 13    Mapping,
 14    NamedTuple,
 15    Optional,
 16    Sequence,
 17    Set,
 18    Tuple,
 19    Union,
 20    no_type_check,
 21)
 22
 23import rich.console
 24import rich.markdown
 25from pydantic import BaseModel, Field, field_validator, model_validator
 26from pydantic_core.core_schema import ErrorType
 27from typing_extensions import TypedDict, assert_never
 28
 29from ._internal.constants import VERSION
 30from ._internal.io import is_yaml_value
 31from ._internal.io_utils import write_yaml
 32from ._internal.type_guards import is_mapping
 33from ._internal.warning_levels import (
 34    ALERT,
 35    ALERT_NAME,
 36    ERROR,
 37    ERROR_NAME,
 38    INFO,
 39    INFO_NAME,
 40    WARNING,
 41    WARNING_NAME,
 42    WarningLevel,
 43    WarningSeverity,
 44)
 45from .conda_env import CondaEnv
 46
 47Loc = Tuple[Union[int, str], ...]
 48"""location of error/warning in a nested data structure"""
 49
 50WarningSeverityName = Literal["info", "warning", "alert"]
 51WarningLevelName = Literal[WarningSeverityName, "error"]
 52
 53WARNING_SEVERITY_TO_NAME: Mapping[WarningSeverity, WarningSeverityName] = (
 54    MappingProxyType({INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME})
 55)
 56WARNING_LEVEL_TO_NAME: Mapping[WarningLevel, WarningLevelName] = MappingProxyType(
 57    {INFO: INFO_NAME, WARNING: WARNING_NAME, ALERT: ALERT_NAME, ERROR: ERROR_NAME}
 58)
 59WARNING_NAME_TO_LEVEL: Mapping[WarningLevelName, WarningLevel] = MappingProxyType(
 60    {v: k for k, v in WARNING_LEVEL_TO_NAME.items()}
 61)
 62
 63
 64class ValidationEntry(BaseModel):
 65    """Base of `ErrorEntry` and `WarningEntry`"""
 66
 67    loc: Loc
 68    msg: str
 69    type: Union[ErrorType, str]
 70
 71
 72class ErrorEntry(ValidationEntry):
 73    """An error in a `ValidationDetail`"""
 74
 75    traceback: List[str] = Field(default_factory=list)
 76
 77
 78class WarningEntry(ValidationEntry):
 79    """A warning in a `ValidationDetail`"""
 80
 81    severity: WarningSeverity = WARNING
 82    severity_name: WarningSeverityName = WARNING_NAME
 83
 84    @model_validator(mode="before")
 85    @classmethod
 86    def sync_severity_with_severity_name(
 87        cls, data: Union[Mapping[Any, Any], Any]
 88    ) -> Any:
 89        if is_mapping(data):
 90            data = dict(data)
 91            if (
 92                "severity" in data
 93                and "severity_name" not in data
 94                and data["severity"] in WARNING_SEVERITY_TO_NAME
 95            ):
 96                data["severity_name"] = WARNING_SEVERITY_TO_NAME[data["severity"]]
 97
 98            if (
 99                "severity" in data
100                and "severity_name" not in data
101                and data["severity"] in WARNING_SEVERITY_TO_NAME
102            ):
103                data["severity"] = WARNING_NAME_TO_LEVEL[data["severity_name"]]
104
105        return data
106
107
108def format_loc(loc: Loc, enclose_in: str = "`") -> str:
109    """helper to format a location tuple `Loc` as Markdown string"""
110    if not loc:
111        loc = ("__root__",)
112
113    loc_str = ".".join(f"({x})" if x[0].isupper() else x for x in map(str, loc))
114
115    # additional field validation can make the location information quite convoluted, e.g.
116    # `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
117    # therefore we remove the `.function-after[validate_url_ok(), url['http','https']]` here
118    brief_loc_str, *_ = loc_str.split(".function-after")
119    return f"{enclose_in}{brief_loc_str}{enclose_in}"
120
121
122class InstalledPackage(NamedTuple):
123    name: str
124    version: str
125    build: str = ""
126    channel: str = ""
127
128
129class ValidationContextSummary(TypedDict):
130    perform_io_checks: bool
131    known_files: Mapping[str, str]
132    root: str
133    warning_level: str
134
135
136class ValidationDetail(BaseModel, extra="allow"):
137    """a detail in a validation summary"""
138
139    name: str
140    status: Literal["passed", "failed"]
141    loc: Loc = ()
142    """location in the RDF that this detail applies to"""
143    errors: List[ErrorEntry] = Field(default_factory=list)
144    warnings: List[WarningEntry] = Field(default_factory=list)
145    context: Optional[ValidationContextSummary] = None
146
147    recommended_env: Optional[CondaEnv] = None
148    """recommended conda environemnt for this validation detail"""
149    conda_compare: Optional[str] = None
150    """output of `conda compare <recommended env>`"""
151
152    def model_post_init(self, __context: Any):
153        """create `conda_compare` default value if needed"""
154        super().model_post_init(__context)
155        if self.recommended_env is None or self.conda_compare is not None:
156            return
157
158        dumped_env = self.recommended_env.model_dump(mode="json")
159        if not is_yaml_value(dumped_env):
160            self.conda_compare = "Failed to dump recommended env to valid yaml"
161            return
162
163        with TemporaryDirectory() as d:
164            path = Path(d) / "env.yaml"
165            with path.open("w", encoding="utf-8") as f:
166                write_yaml(dumped_env, f)
167
168            compare_proc = subprocess.run(
169                ["conda", "compare", str(path)],
170                stdout=subprocess.PIPE,
171                stderr=subprocess.STDOUT,
172                shell=True,
173                text=True,
174            )
175            self.conda_compare = (
176                compare_proc.stdout
177                or f"conda compare exited with {compare_proc.returncode}"
178            )
179
180    def __str__(self):
181        return f"{self.__class__.__name__}:\n" + self.format()
182
183    @property
184    def status_icon(self):
185        if self.status == "passed":
186            return "✔️"
187        else:
188            return "❌"
189
190    def format(self, hide_tracebacks: bool = False, root_loc: Loc = ()) -> str:
191        """format as Markdown string"""
192        indent = "    " if root_loc else ""
193        errs_wrns = self._format_errors_and_warnings(
194            hide_tracebacks=hide_tracebacks, root_loc=root_loc
195        )
196        return f"{indent}{self.status_icon} {self.name.strip('.')}: {self.status}{errs_wrns}"
197
198    def _format_errors_and_warnings(self, hide_tracebacks: bool, root_loc: Loc):
199        indent = "    " if root_loc else ""
200        if hide_tracebacks:
201            tbs = [""] * len(self.errors)
202        else:
203            slim_tracebacks = [
204                [tt.replace("\n", "<br>") for t in e.traceback if (tt := t.strip())]
205                for e in self.errors
206            ]
207            tbs = [
208                ("<br>      Traceback:<br>      " if st else "") + "<br>      ".join(st)
209                for st in slim_tracebacks
210            ]
211
212        def join_parts(parts: Iterable[Tuple[str, str]]):
213            last_loc = None
214            lines: List[str] = []
215            for loc, msg in parts:
216                if loc == last_loc:
217                    lines.append(f"<br>  {loc} {msg}")
218                else:
219                    lines.append(f"<br>- {loc} {msg}")
220
221                last_loc = loc
222
223            return "".join(lines)
224
225        es = join_parts(
226            (format_loc(root_loc + e.loc), f"{e.msg}{tb}")
227            for e, tb in zip(self.errors, tbs)
228        )
229        ws = join_parts((format_loc(root_loc + w.loc), w.msg) for w in self.warnings)
230
231        return (
232            f"\n{indent}errors:\n{es}"
233            if es
234            else "" + f"\n{indent}warnings:\n{ws}" if ws else ""
235        )
236
237
238class ValidationSummary(BaseModel, extra="allow"):
239    """Summarizes output of all bioimageio validations and tests
240    for one specific `ResourceDescr` instance."""
241
242    name: str
243    source_name: str
244    type: str
245    format_version: str
246    status: Literal["passed", "failed"]
247    details: List[ValidationDetail]
248    env: Set[InstalledPackage] = Field(
249        default_factory=lambda: {
250            InstalledPackage(name="bioimageio.spec", version=VERSION)
251        }
252    )
253    """list of selected, relevant package versions"""
254
255    conda_list: Optional[Sequence[InstalledPackage]] = None
256    """parsed output of conda list"""
257
258    @property
259    def status_icon(self):
260        if self.status == "passed":
261            return "✔️"
262        else:
263            return "❌"
264
265    @property
266    def errors(self) -> List[ErrorEntry]:
267        return list(chain.from_iterable(d.errors for d in self.details))
268
269    @property
270    def warnings(self) -> List[WarningEntry]:
271        return list(chain.from_iterable(d.warnings for d in self.details))
272
273    def __str__(self):
274        return f"{self.__class__.__name__}:\n" + self.format()
275
276    @staticmethod
277    def _format_md_table(rows: List[List[str]]) -> str:
278        """format `rows` as markdown table"""
279        n_cols = len(rows[0])
280        assert all(len(row) == n_cols for row in rows)
281        col_widths = [max(max(len(row[i]) for row in rows), 3) for i in range(n_cols)]
282
283        # fix new lines in table cell
284        rows = [[line.replace("\n", "<br>") for line in r] for r in rows]
285
286        lines = [" | ".join(rows[0][i].center(col_widths[i]) for i in range(n_cols))]
287        lines.append(" | ".join("---".center(col_widths[i]) for i in range(n_cols)))
288        lines.extend(
289            [
290                " | ".join(row[i].ljust(col_widths[i]) for i in range(n_cols))
291                for row in rows[1:]
292            ]
293        )
294        return "\n| " + " |\n| ".join(lines) + " |\n"
295
296    def format(
297        self,
298        hide_tracebacks: bool = False,
299        hide_source: bool = False,
300        hide_env: bool = False,
301        root_loc: Loc = (),
302    ) -> str:
303        """Format summary as Markdown string
304
305        Suitable to embed in HTML using '<br>' instead of '\n'.
306        """
307        info = self._format_md_table(
308            [[self.status_icon, f"{self.name.strip('.').strip()} {self.status}"]]
309            + ([] if hide_source else [["source", self.source_name]])
310            + [
311                ["format version", f"{self.type} {self.format_version}"],
312            ]
313            + ([] if hide_env else [[e.name, e.version] for e in self.env])
314        )
315
316        def format_loc(loc: Loc):
317            return "`" + (".".join(map(str, root_loc + loc)) or ".") + "`"
318
319        details = [["❓", "location", "detail"]]
320        for d in self.details:
321            details.append([d.status_icon, format_loc(d.loc), d.name])
322            if d.context is not None:
323                details.append(
324                    [
325                        "🔍",
326                        "context.perform_io_checks",
327                        str(d.context["perform_io_checks"]),
328                    ]
329                )
330                if d.context["perform_io_checks"]:
331                    details.append(["🔍", "context.root", d.context["root"]])
332                    for kfn, sha in d.context["known_files"].items():
333                        details.append(["🔍", f"context.known_files.{kfn}", sha])
334
335                details.append(
336                    ["🔍", "context.warning_level", d.context["warning_level"]]
337                )
338
339            if d.recommended_env is not None:
340                rec_env = StringIO()
341                json_env = d.recommended_env.model_dump(
342                    mode="json", exclude_defaults=True
343                )
344                assert is_yaml_value(json_env)
345                write_yaml(json_env, rec_env)
346                rec_env_code = rec_env.getvalue().replace("\n", "</code><br><code>")
347                details.append(
348                    [
349                        "🐍",
350                        format_loc(d.loc),
351                        f"recommended conda env ({d.name})<br>"
352                        + f"<pre><code>{rec_env_code}</code></pre>",
353                    ]
354                )
355
356            if d.conda_compare:
357                details.append(
358                    [
359                        "🐍",
360                        format_loc(d.loc),
361                        f"conda compare ({d.name}):<br>"
362                        + d.conda_compare.replace("\n", "<br>"),
363                    ]
364                )
365
366            for entry in d.errors:
367                details.append(
368                    [
369                        "❌",
370                        format_loc(entry.loc),
371                        entry.msg.replace("\n\n", "<br>").replace("\n", "<br>"),
372                    ]
373                )
374                if hide_tracebacks:
375                    continue
376
377                formatted_tb_lines: List[str] = []
378                for tb in entry.traceback:
379                    if not (tb_stripped := tb.strip()):
380                        continue
381
382                    first_tb_line, *tb_lines = tb_stripped.split("\n")
383                    if (
384                        first_tb_line.startswith('File "')
385                        and '", line' in first_tb_line
386                    ):
387                        path, where = first_tb_line[len('File "') :].split('", line')
388                        try:
389                            p = Path(path)
390                        except Exception:
391                            file_name = path
392                        else:
393                            path = p.as_posix()
394                            file_name = p.name
395
396                        where = ", line" + where
397                        first_tb_line = f'[{file_name}]({file_name} "{path}"){where}'
398
399                    if tb_lines:
400                        tb_rest = "<br>`" + "`<br>`".join(tb_lines) + "`"
401                    else:
402                        tb_rest = ""
403
404                    formatted_tb_lines.append(first_tb_line + tb_rest)
405
406                details.append(["", "", "<br>".join(formatted_tb_lines)])
407
408            for entry in d.warnings:
409                details.append(["⚠", format_loc(entry.loc), entry.msg])
410
411        return f"{info}{self._format_md_table(details)}"
412
413    # TODO: fix bug which casuses extensive white space between the info table and details table
414    @no_type_check
415    def display(self) -> None:
416        formatted = self.format()
417        try:
418            from IPython.core.getipython import get_ipython
419            from IPython.display import Markdown, display
420        except ImportError:
421            pass
422        else:
423            if get_ipython() is not None:
424                _ = display(Markdown(formatted))
425                return
426
427        rich_markdown = rich.markdown.Markdown(formatted)
428        console = rich.console.Console()
429        console.print(rich_markdown)
430
431    def add_detail(self, detail: ValidationDetail):
432        if detail.status == "failed":
433            self.status = "failed"
434        elif detail.status != "passed":
435            assert_never(detail.status)
436
437        self.details.append(detail)
438
439    @field_validator("env", mode="before")
440    def _convert_dict(cls, value: List[Union[List[str], Dict[str, str]]]):
441        """convert old env value for backwards compatibility"""
442        if isinstance(value, list):
443            return [
444                (
445                    (v["name"], v["version"], v.get("build", ""), v.get("channel", ""))
446                    if isinstance(v, dict) and "name" in v and "version" in v
447                    else v
448                )
449                for v in value
450            ]
451        else:
452            return value