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

1from __future__ import annotations 

2 

3import json 

4import warnings 

5from concurrent.futures import Future, ThreadPoolExecutor, as_completed 

6from typing import Any 

7 

8from loguru import logger 

9from packaging.version import Version 

10from tqdm import tqdm 

11 

12from backoffice.compatibility import ( 

13 TOOL_NAMES, 

14 CompatibilityScores, 

15 CompatibilitySummary, 

16 ToolCompatibilityReportWithToolInfo, 

17 ToolName, 

18 ToolNameVersioned, 

19 ToolReportDetails, 

20) 

21from backoffice.index import IndexItem, IndexItemVersion, load_index 

22from backoffice.utils import get_summary, get_summary_file_path 

23from backoffice.utils_pure import get_all_tool_report_paths 

24 

25 

26def summarize_reports(): 

27 index = load_index() 

28 for item in tqdm(index.items): 

29 for v in item.versions: 

30 _summarize(item, v) 

31 

32 

33def summarize_reports_parallel(max_workers: int | None = None): 

34 index = load_index() 

35 with ThreadPoolExecutor(max_workers=max_workers) as executor: 

36 futures: list[Future[Any]] = [] 

37 for item in index.items: 

38 for v in item.versions: 

39 futures.append(executor.submit(_summarize, item, v)) 

40 

41 for _ in tqdm(as_completed(futures), total=len(futures)): 

42 pass 

43 

44 

45def _summarize(item: IndexItem, v: IndexItemVersion): 

46 """Conflate all summaries for a given item version.""" 

47 

48 initial_summary = get_summary(item.id, v.version) 

49 

50 reports: list[ToolCompatibilityReportWithToolInfo] = [] 

51 scores: dict[ToolNameVersioned, float] = {} 

52 metadata_completeness = 0.0 

53 metadata_format_score = 0.0 

54 metadata_format_version = Version( 

55 "0.0.0" 

56 ) # to track the latest core version with valid format 

57 for report_path in get_all_tool_report_paths(item.id, v.version): 

58 tool, tool_version = report_path.stem.split("_", 1) 

59 tool = tool.lower() 

60 if tool not in TOOL_NAMES: 

61 warnings.warn(f"Report {report_path} has unknown tool name '{tool}'.") 

62 continue 

63 try: 

64 data = json.loads(report_path.read_text(encoding="utf-8")) 

65 if "tool" in data: 

66 if data["tool"] != tool: 

67 warnings.warn( 

68 f"Report {report_path} has inconsistent tool name '{data['tool']}' != '{tool}'." 

69 ) 

70 del data["tool"] 

71 

72 if "tool_version" in data: 

73 if data["tool_version"] != tool_version: 

74 warnings.warn( 

75 f"Report {report_path} has inconsistent tool version '{data['tool_version']}' != '{tool_version}'." 

76 ) 

77 del data["tool_version"] 

78 

79 report = ToolCompatibilityReportWithToolInfo( 

80 tool=tool, tool_version=tool_version, **data 

81 ) 

82 except Exception as e: 

83 report = ToolCompatibilityReportWithToolInfo( 

84 tool=tool, 

85 tool_version=tool_version, 

86 status="failed", 

87 error=str(e), 

88 score=0.0, 

89 details="Failed to parse compatibility report.", 

90 ) 

91 

92 scores[f"{tool}_{tool_version}"] = report.score 

93 reports.append(report) 

94 if report.tool == "bioimageio.core" and isinstance( 

95 report.details, ToolReportDetails 

96 ): 

97 # select the best completeness score among core reports 

98 metadata_completeness = max( 

99 metadata_completeness, report.details.metadata_completeness or 0.0 

100 ) 

101 # determine metadata format score 

102 # - valid-format for latest core report: 1.0 

103 # - valid-format for older core report: 0.5 

104 # - invalid format for all core reports: 0.0 

105 core_version = Version(tool_version) 

106 if core_version >= metadata_format_version: 

107 metadata_format_version = core_version 

108 if report.details.status in ("passed", "valid-format"): 

109 metadata_format_score = 1.0 

110 else: 

111 metadata_format_score = 0.5 if metadata_format_score else 0.0 

112 

113 elif not metadata_format_score and report.details.status in ( 

114 "passed", 

115 "valid-format", 

116 ): 

117 metadata_format_score = 0.5 

118 

119 tests: dict[ToolName, dict[str, ToolCompatibilityReportWithToolInfo]] = {} 

120 for r in reports: 

121 tests.setdefault(r.tool, {})[r.tool_version] = r 

122 

123 compatibility_scores = CompatibilityScores( 

124 tool_compatibility_version_specific=scores, 

125 metadata_completeness=metadata_completeness, 

126 metadata_format=metadata_format_score, 

127 ) 

128 

129 compatibility_status = ( 

130 "passed" 

131 if compatibility_scores.tool_compatibility 

132 and max(compatibility_scores.tool_compatibility.values()) >= 0.5 

133 else "failed" 

134 ) 

135 summary = CompatibilitySummary( 

136 rdf_content=initial_summary.rdf_content, 

137 rdf_yaml_sha256=initial_summary.rdf_yaml_sha256, 

138 status=compatibility_status, 

139 scores=compatibility_scores, 

140 tests=tests, 

141 ) 

142 

143 json_dict = summary.model_dump(mode="json") 

144 with get_summary_file_path(item.id, v.version).open("wt", encoding="utf-8") as f: 

145 json.dump(json_dict, f, indent=4, sort_keys=True, ensure_ascii=False) 

146 # TODO: use .model_dump_json once it supports 'sort_keys' argument for a potential speed gain 

147 # _ = get_summary_file_path(item.id, v.version).write_text( 

148 # summary.model_dump_json(indent=4), encoding="utf-8" 

149 # ) 

150 

151 logger.info( 

152 "summarized {} version {} with {} reports, status: {}, metadata completeness: {:.2f}", 

153 item.id, 

154 v.version, 

155 len(reports), 

156 compatibility_status, 

157 metadata_completeness, 

158 ) 

159 

160 

161if __name__ == "__main__": 

162 summarize_reports()