add recommended_env and conda_compare

bioimageio/spec/conda_env.py

@@ -6,11 +6,11 @@
 from ruyaml import YAML
 from typing_extensions import assert_never
 
-from bioimageio.spec._internal.gh_utils import set_github_warning
-from bioimageio.spec.common import RelativeFilePath
-from bioimageio.spec.model import v0_4, v0_5
-from bioimageio.spec.model.v0_5 import Version
-from bioimageio.spec.utils import download
+from ._internal.gh_utils import set_github_warning
+from .common import RelativeFilePath
+from .model import v0_4, v0_5
+from .model.v0_5 import Version
+from .utils import download
 
 yaml = YAML(typ="safe")
 
@@ -43,7 +43,7 @@ class CondaEnv(BaseModel):
     dependencies: List[Union[str, PipDeps]] = Field(default_factory=list)
 
     @field_validator("name", mode="after")
-    def _ensure_valid_conda_env_name(self, value: Optional[str]) -> Optional[str]:
+    def _ensure_valid_conda_env_name(cls, value: Optional[str]) -> Optional[str]:
         if value is None:
             return None
 

bioimageio/spec/summary.py

@@ -1,25 +1,35 @@
+import subprocess
+from io import StringIO
 from itertools import chain
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 from types import MappingProxyType
 from typing import (
+    TYPE_CHECKING,
     Any,
+    Dict,
     Iterable,
     List,
     Literal,
     Mapping,
+    NamedTuple,
     Optional,
+    Sequence,
+    Set,
     Tuple,
     Union,
     no_type_check,
 )
 
 import rich.console
 import rich.markdown
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 from pydantic_core.core_schema import ErrorType
 from typing_extensions import TypedDict, assert_never
 
 from ._internal.constants import VERSION
+from ._internal.io import is_yaml_value
+from ._internal.io_utils import write_yaml
 from ._internal.warning_levels import (
     ALERT,
     ALERT_NAME,
@@ -33,6 +43,9 @@
     WarningSeverity,
 )
 
+if TYPE_CHECKING:
+    from .conda_env import CondaEnv
+
 Loc = Tuple[Union[int, str], ...]
 """location of error/warning in a nested data structure"""
 
@@ -102,9 +115,11 @@ def format_loc(loc: Loc, enclose_in: str = "`") -> str:
     return f"{enclose_in}{brief_loc_str}{enclose_in}"
 
 
-class InstalledPackage(TypedDict):
+class InstalledPackage(NamedTuple):
     name: str
     version: str
+    build: str = ""
+    channel: str = ""
 
 
 class ValidationContextSummary(TypedDict):
@@ -117,10 +132,30 @@ class ValidationContextSummary(TypedDict):
 class ValidationDetail(BaseModel, extra="allow"):
     name: str
     status: Literal["passed", "failed"]
+    loc: Loc = ()
     errors: List[ErrorEntry] = Field(default_factory=list)
     warnings: List[WarningEntry] = Field(default_factory=list)
     context: Optional[ValidationContextSummary] = None
 
+    recommended_env: Optional["CondaEnv"] = None
+    """recommended conda environemnt for this validation detail"""
+    conda_compare: Optional[str] = None
+    """output of `conda compare <recommended env>`"""
+
+    def model_post_init(self, __context: Any):
+        """create `conda_compare` default value if needed"""
+        super().model_post_init(__context)
+        if self.recommended_env is not None and self.conda_compare is None:
+            dumped_env = self.recommended_env.model_dump(mode="json")
+            if is_yaml_value(dumped_env):
+                with NamedTemporaryFile(encoding="utf-8") as f:
+                    write_yaml(dumped_env, f)
+                    self.conda_compare = subprocess.run(
+                        ["conda", "compare", f.name], capture_output=True, text=True
+                    ).stdout
+            else:
+                self.conda_compare = "Failed to dump recommended env to valid yaml"
+
     def __str__(self):
         return f"{self.__class__.__name__}:\n" + self.format()
 
@@ -185,13 +220,16 @@ class ValidationSummary(BaseModel, extra="allow"):
     format_version: str
     status: Literal["passed", "failed"]
     details: List[ValidationDetail]
-    env: List[InstalledPackage] = Field(
-        default_factory=lambda: [
+    env: Set[InstalledPackage] = Field(
+        default_factory=lambda: {
             InstalledPackage(name="bioimageio.spec", version=VERSION)
-        ]
+        }
     )
     """list of selected, relevant package versions"""
 
+    conda_list: Optional[Sequence[InstalledPackage]] = None
+    """parsed output of conda list"""
+
     @property
     def status_icon(self):
         if self.status == "passed":
@@ -243,15 +281,15 @@ def format(
             + [
                 ["format version", f"{self.type} {self.format_version}"],
             ]
-            + ([] if hide_env else [[e["name"], e["version"]] for e in self.env])
+            + ([] if hide_env else [[e.name, e.version] for e in self.env])
         )
 
         def format_loc(loc: Loc):
             return "`" + (".".join(map(str, root_loc + loc)) or ".") + "`"
 
         details = [["❓", "location", "detail"]]
         for d in self.details:
-            details.append([d.status_icon, "", d.name])
+            details.append([d.status_icon, format_loc(d.loc), d.name])
             if d.context is not None:
                 details.append(
                     [
@@ -269,6 +307,24 @@ def format_loc(loc: Loc):
                     ["🔍", "context.warning_level", d.context["warning_level"]]
                 )
 
+            if d.recommended_env is not None:
+                rec_env = StringIO()
+                json_env = d.recommended_env.model_dump(mode="json")
+                assert is_yaml_value(json_env)
+                write_yaml(json_env, rec_env)
+                details.append(
+                    [
+                        "🐍",
+                        "recommended conda env",
+                        f"```yaml\n{rec_env.read()}\n```".replace("\n", "<br>"),
+                    ]
+                )
+
+            if d.conda_compare is not None:
+                details.append(
+                    ["🐍", "actual conda env", d.conda_compare.replace("\n", "<br>")]
+                )
+
             for entry in d.errors:
                 details.append(
                     [
@@ -341,3 +397,18 @@ def add_detail(self, detail: ValidationDetail):
             assert_never(detail.status)
 
         self.details.append(detail)
+
+    @field_validator("env", mode="before")
+    def _convert_dict(cls, value: List[Union[List[str], Dict[str, str]]]):
+        """convert old env value for backwards compatibility"""
+        if isinstance(value, list):
+            return [
+                (
+                    (v["name"], v["version"], v.get("build", ""), v.get("channel", ""))
+                    if isinstance(v, dict) and "name" in v and "version" in v
+                    else v
+                )
+                for v in value
+            ]
+        else:
+            return value