Coverage for bioimageio/spec/summary.py: 41%
223 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-02-05 13:53 +0000
« prev ^ index » next coverage.py v7.6.10, created at 2025-02-05 13:53 +0000
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)
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
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
47Loc = Tuple[Union[int, str], ...]
48"""location of error/warning in a nested data structure"""
50WarningSeverityName = Literal["info", "warning", "alert"]
51WarningLevelName = Literal[WarningSeverityName, "error"]
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)
64class ValidationEntry(BaseModel):
65 """Base of `ErrorEntry` and `WarningEntry`"""
67 loc: Loc
68 msg: str
69 type: Union[ErrorType, str]
72class ErrorEntry(ValidationEntry):
73 """An error in a `ValidationDetail`"""
75 traceback: List[str] = Field(default_factory=list)
78class WarningEntry(ValidationEntry):
79 """A warning in a `ValidationDetail`"""
81 severity: WarningSeverity = WARNING
82 severity_name: WarningSeverityName = WARNING_NAME
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"]]
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"]]
105 return data
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__",)
113 loc_str = ".".join(f"({x})" if x[0].isupper() else x for x in map(str, loc))
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}"
122class InstalledPackage(NamedTuple):
123 name: str
124 version: str
125 build: str = ""
126 channel: str = ""
129class ValidationContextSummary(TypedDict):
130 perform_io_checks: bool
131 known_files: Mapping[str, str]
132 root: str
133 warning_level: str
136class ValidationDetail(BaseModel, extra="allow"):
137 """a detail in a validation summary"""
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
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>`"""
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
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
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)
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 )
180 def __str__(self):
181 return f"{self.__class__.__name__}:\n" + self.format()
183 @property
184 def status_icon(self):
185 if self.status == "passed":
186 return "✔️"
187 else:
188 return "❌"
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}"
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 ]
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}")
221 last_loc = loc
223 return "".join(lines)
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)
231 return (
232 f"\n{indent}errors:\n{es}"
233 if es
234 else "" + f"\n{indent}warnings:\n{ws}" if ws else ""
235 )
238class ValidationSummary(BaseModel, extra="allow"):
239 """Summarizes output of all bioimageio validations and tests
240 for one specific `ResourceDescr` instance."""
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"""
255 conda_list: Optional[Sequence[InstalledPackage]] = None
256 """parsed output of conda list"""
258 @property
259 def status_icon(self):
260 if self.status == "passed":
261 return "✔️"
262 else:
263 return "❌"
265 @property
266 def errors(self) -> List[ErrorEntry]:
267 return list(chain.from_iterable(d.errors for d in self.details))
269 @property
270 def warnings(self) -> List[WarningEntry]:
271 return list(chain.from_iterable(d.warnings for d in self.details))
273 def __str__(self):
274 return f"{self.__class__.__name__}:\n" + self.format()
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)]
283 # fix new lines in table cell
284 rows = [[line.replace("\n", "<br>") for line in r] for r in rows]
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"
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
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 )
316 def format_loc(loc: Loc):
317 return "`" + (".".join(map(str, root_loc + loc)) or ".") + "`"
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])
335 details.append(
336 ["🔍", "context.warning_level", d.context["warning_level"]]
337 )
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 )
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 )
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
377 formatted_tb_lines: List[str] = []
378 for tb in entry.traceback:
379 if not (tb_stripped := tb.strip()):
380 continue
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
396 where = ", line" + where
397 first_tb_line = f'[{file_name}]({file_name} "{path}"){where}'
399 if tb_lines:
400 tb_rest = "<br>`" + "`<br>`".join(tb_lines) + "`"
401 else:
402 tb_rest = ""
404 formatted_tb_lines.append(first_tb_line + tb_rest)
406 details.append(["", "", "<br>".join(formatted_tb_lines)])
408 for entry in d.warnings:
409 details.append(["⚠", format_loc(entry.loc), entry.msg])
411 return f"{info}{self._format_md_table(details)}"
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
427 rich_markdown = rich.markdown.Markdown(formatted)
428 console = rich.console.Console()
429 console.print(rich_markdown)
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)
437 self.details.append(detail)
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