Coverage for src/backoffice/compatibility.py: 0%

1"""data models for compatibility reports""" 

2 

3import warnings 

4from typing import Any, List, Literal, Mapping, Optional, Sequence, Union 

5 

6from annotated_types import Interval 

7from packaging.version import Version 

8 

9try: 

10 from pydantic import BaseModel, Field, HttpUrl, computed_field, model_validator 

11except ImportError as e: 

12 raise ImportError( 

13 "pydantic is required for backoffice.compatibility. " 

14 "Please install `backoffice[dev]` or use backoffice.compatibility_pure instead." 

15 ) from e 

16 

17from typing_extensions import Annotated 

18 

19PartnerToolName = Literal[ 

20 "ilastik", 

21 "deepimagej", 

22 "icy", 

23 "biapy", 

24 "careamics", 

25] 

26ToolName = Literal["bioimageio.core", PartnerToolName] 

27 

28PARTNER_TOOL_NAMES = ( 

29 "ilastik", 

30 "deepimagej", 

31 "icy", 

32 "biapy", 

33 "careamics", 

34) 

35TOOL_NAMES = ("bioimageio.core", *PARTNER_TOOL_NAMES) 

36 

37ToolNameVersioned = str 

38 

39 

40class Node(BaseModel): 

41 """Base data model with common config""" 

42 

43 pass 

44 

45 

46class Badge(Node): 

47 icon: HttpUrl 

48 label: str 

49 url: HttpUrl 

50 

51 

52class ToolReportDetails(Node, extra="allow"): 

53 traceback: Optional[Sequence[str]] = None 

54 warnings: Optional[Mapping[str, Any]] = None 

55 metadata_completeness: Optional[float] = None 

56 status: Union[Literal["passed", "valid-format", "failed"], Any] = None 

57 

58 

59class ToolCompatibilityReport(Node, extra="allow"): 

60 """Used to report on the compatibility of resource description 

61 in the bioimageio collection for a version specific tool. 

62 """ 

63 

64 tool: Annotated[ToolName, Field(exclude=True, pattern=r"^[a-zA-Z0-9-\.]+$")] 

65 """tool name""" 

66 

67 tool_version: Annotated[str, Field(exclude=True, pattern=r"^[a-z0-9\.-]+$")] 

68 """tool version, ideally in SemVer 2.0 format""" 

69 

70 @property 

71 def report_name(self) -> str: 

72 return f"{self.tool}_{self.tool_version}" 

73 

74 status: Literal["passed", "failed", "not-applicable"] 

75 """status of this tool for this resource""" 

76 

77 score: Annotated[float, Interval(ge=0, le=1.0)] 

78 """score for the compatibility of this tool with the resource""" 

79 

80 @model_validator(mode="before") 

81 @classmethod 

82 def _set_default_score(cls, values: dict[str, Any]) -> dict[str, Any]: 

83 if isinstance(values, dict) and "score" not in values: 

84 values["score"] = 1.0 if values.get("status") == "passed" else 0.0 

85 

86 return values 

87 

88 error: Optional[str] 

89 """error message if `status`=='failed'""" 

90 

91 details: Union[ToolReportDetails, str, List[str], None] = None 

92 """details to explain the `status`""" 

93 

94 badge: Optional[Badge] = None 

95 """status badge with a resource specific link to the tool""" 

96 

97 links: Sequence[str] = () 

98 """the checked resource should link these other bioimage.io resources""" 

99 

100 

101class CompatibilityScores(Node): 

102 tool_compatibility_version_specific: Mapping[ 

103 ToolNameVersioned, Annotated[float, Interval(ge=0, le=1.0)] 

104 ] 

105 

106 metadata_completeness: Annotated[float, Interval(ge=0, le=1.0)] = 0.0 

107 """Score for metadata completeness. 

108 

109 A measure of how many optional fields in the resource RDF are filled out. 

110 """ 

111 

112 metadata_format: Annotated[float, Interval(ge=0, le=1.0)] = 0.0 

113 """Score for metadata formatting. 

114 

115 - 1.0: resource RDF conforms to the latest spec version 

116 - 0.5: resource RDF conforms to an older spec version 

117 - 0.0: resource RDF does not conform to any known spec version 

118""" 

119 

120 @computed_field 

121 @property 

122 def core_compatibility(self) -> float: 

123 return self.tool_compatibility.get("bioimageio.core", 0.0) 

124 

125 @computed_field 

126 @property 

127 def tool_compatibility( 

128 self, 

129 ) -> Mapping[ToolName, Annotated[float, Interval(ge=0, le=1.0)]]: 

130 """Aggregated tool compatibility score""" 

131 grouped: dict[ToolName, dict[Version, float]] = {} 

132 for tool, value in self.tool_compatibility_version_specific.items(): 

133 assert value <= 1.0, f"Tool {tool} has a compatibility score > 1.0: {value}" 

134 tool_name, tool_version = tool.split("_", 1) 

135 if tool_name not in TOOL_NAMES: 

136 warnings.warn(f"Tool {tool_name} is not a valid ToolName") 

137 continue 

138 

139 malus = 0.0 

140 try: 

141 version = Version(tool_version) 

142 except Exception: 

143 version = Version("0.0.0") 

144 malus += 0.1 # penalize non-semver versions 

145 

146 grouped.setdefault(tool_name, {})[version] = value - malus 

147 

148 for tool in list(grouped): 

149 if not grouped[tool]: 

150 del grouped[tool] 

151 

152 agglomerated: dict[ToolName, float] = {} 

153 for tool, version_scores in grouped.items(): 

154 latest_version = max(version_scores.keys()) 

155 

156 if version_scores[latest_version] >= 0.8: 

157 # if the latest version is compatible use it as the score 

158 score = version_scores[latest_version] 

159 else: 

160 # average the top 4 scores to score max 0.8 

161 # as penalty if the last_version isn't fully compatible 

162 top4 = sorted(version_scores.values(), reverse=True)[:4] 

163 score = min(0.8, sum(top4) / len(top4)) 

164 

165 agglomerated[tool] = score 

166 

167 return agglomerated 

168 

169 @computed_field 

170 @property 

171 def overall_partner_tool_compatibility( 

172 self, 

173 ) -> Annotated[float, Interval(ge=0, le=1.0)]: 

174 """Overall partner tool compatibility score. 

175 Note: 

176 - Currently implemented as: Average of the top 3 partner tool compatibility scores. 

177 - Implementation is subject to change in the future. 

178 """ 

179 top3 = sorted( 

180 [v for k, v in self.tool_compatibility.items() if k in PARTNER_TOOL_NAMES], 

181 reverse=True, 

182 )[:3] 

183 if not top3: 

184 return 0.0 

185 else: 

186 return sum(top3) / 3 

187 

188 @computed_field 

189 @property 

190 def overall_compatibility(self) -> Annotated[float, Interval(ge=0, le=1.0)]: 

191 """Weighted, overall score between 0 and 1. 

192 Note: The scoring scheme is subject to change in the future. 

193 """ 

194 return ( 

195 0.25 * self.metadata_format 

196 + 0.25 * self.metadata_completeness 

197 + 0.25 * self.core_compatibility 

198 + 0.25 * self.overall_partner_tool_compatibility 

199 ) 

200 

201 

202class InitialSummary(Node): 

203 rdf_content: dict[str, Any] 

204 """The RDF content of the original rdf.yaml file.""" 

205 

206 rdf_yaml_sha256: str 

207 """SHA-256 of the original RDF YAML file.""" 

208 

209 status: Literal["passed", "failed", "untested"] 

210 """status of the bioimageio.core reproducibility tests.""" 

211 

212 

213class CompatibilitySummary(InitialSummary): 

214 scores: CompatibilityScores 

215 """Scores for compatibility with the bioimage.io community tools.""" 

216 

217 tests: Mapping[ToolName, Mapping[str, ToolCompatibilityReport]] 

218 """Compatibility reports for each tool for each version."""