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

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

2 

3from __future__ import annotations 

4 

5import warnings 

6from collections.abc import Mapping, Sequence 

7from typing import Any, Literal 

8 

9from annotated_types import Interval 

10from packaging.version import Version 

11 

12try: 

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

14except ImportError as e: 

15 raise ImportError( 

16 "pydantic is required for backoffice.compatibility. " 

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

18 ) from e 

19 

20from typing import Annotated 

21 

22from .compatibility_pure import ( 

23 PARTNER_TOOL_NAMES, 

24 TOOL_NAMES, 

25 ToolName, 

26 ToolNameVersioned, 

27) 

28 

29 

30class Node(BaseModel): 

31 """Base data model with common config""" 

32 

33 

34class Badge(Node): 

35 icon: HttpUrl 

36 label: str 

37 url: HttpUrl 

38 

39 

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

41 traceback: Sequence[str] | None = None 

42 warnings: Mapping[str, Any] | None = None 

43 metadata_completeness: float | None = None 

44 status: Literal["passed", "valid-format", "failed"] | Any = None 

45 

46 

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

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

49 in the bioimageio collection for a version specific tool. 

50 """ 

51 

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

53 """status of this tool for this resource""" 

54 

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

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

57 

58 @model_validator(mode="before") 

59 @classmethod 

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

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

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

63 

64 return values 

65 

66 error: str | None 

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

68 

69 details: ToolReportDetails | str | list[str] | None = None 

70 """details to explain the `status`""" 

71 

72 badge: Badge | None = None 

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

74 

75 links: Sequence[str] = () 

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

77 

78 

79class ToolCompatibilityReportWithToolInfo(ToolCompatibilityReport): 

80 tool: ToolName 

81 """tool name""" 

82 

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

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

85 

86 @property 

87 def report_name(self) -> str: 

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

89 

90 

91class CompatibilityScores(Node): 

92 tool_compatibility_version_specific: Mapping[ 

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

94 ] 

95 

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

97 """Score for metadata completeness. 

98 

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

100 """ 

101 

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

103 """Score for metadata formatting. 

104 

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

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

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

108""" 

109 

110 @computed_field 

111 @property 

112 def core_compatibility(self) -> float: 

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

114 

115 @computed_field 

116 @property 

117 def tool_compatibility( 

118 self, 

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

120 """Aggregated tool compatibility score""" 

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

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

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

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

125 if tool_name not in TOOL_NAMES: 

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

127 continue 

128 

129 malus = 0.0 

130 try: 

131 version = Version(tool_version) 

132 except Exception: 

133 version = Version("0.0.0") 

134 malus += 0.1 # penalize non-semver versions 

135 

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

137 

138 for tool in list(grouped): 

139 if not grouped[tool]: 

140 del grouped[tool] 

141 

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

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

144 latest_version = max(version_scores.keys()) 

145 

146 if version_scores[latest_version] >= 0.8: 

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

148 score = version_scores[latest_version] 

149 else: 

150 # average the top 4 scores to score max 0.8 

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

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

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

154 # however, this score cannot be lower than the latest version score 

155 score = max(score, version_scores[latest_version]) 

156 

157 agglomerated[tool] = score 

158 

159 return agglomerated 

160 

161 @computed_field 

162 @property 

163 def overall_partner_tool_compatibility( 

164 self, 

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

166 """Overall partner tool compatibility score. 

167 Note: 

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

169 - Implementation is subject to change in the future. 

170 """ 

171 top3 = sorted( 

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

173 reverse=True, 

174 )[:3] 

175 if not top3: 

176 return 0.0 

177 else: 

178 return sum(top3) / 3 

179 

180 @computed_field 

181 @property 

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

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

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

185 """ 

186 return ( 

187 0.25 * self.metadata_format 

188 + 0.25 * self.metadata_completeness 

189 + 0.25 * self.core_compatibility 

190 + 0.25 * self.overall_partner_tool_compatibility 

191 ) 

192 

193 

194class InitialSummary(Node): 

195 rdf_content: dict[str, Any] 

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

197 

198 rdf_yaml_sha256: str 

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

200 

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

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

203 

204 

205class CompatibilitySummary(InitialSummary): 

206 scores: CompatibilityScores 

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

208 

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

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