Coverage for src/somesy/codemeta/writer.py: 80%

190 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-18 08:48 +0000

1"""codemeta.json creation module.""" 

2 

3import json 

4import logging 

5import uuid 

6from collections import OrderedDict 

7from collections.abc import Sequence 

8from copy import deepcopy 

9from pathlib import Path 

10from typing import Any 

11 

12from somesy.codemeta.utils import validate_codemeta 

13from somesy.core.models import Entity, Person, ProjectMetadata 

14from somesy.core.writer import FieldKeyMapping, ProjectMetadataWriter 

15 

16logger = logging.getLogger("somesy") 

17 

18V2_CONTEXT = "https://doi.org/10.5063/schema/codemeta-2.0" 

19V3_CONTEXT = "https://w3id.org/codemeta/3.1" 

20 

21 

22class CodeMeta(ProjectMetadataWriter): 

23 """Codemeta.json parser and saver.""" 

24 

25 def __init__( 

26 self, 

27 path: Path, 

28 merge: bool | None = False, 

29 pass_validation: bool | None = False, 

30 ): 

31 """Codemeta.json parser. 

32 

33 See [somesy.core.writer.ProjectMetadataWriter.__init__][]. 

34 """ 

35 self.merge = merge 

36 self._default_context = V3_CONTEXT 

37 mappings: FieldKeyMapping = { 

38 "repository": ["codeRepository"], 

39 "homepage": ["softwareHelp"], 

40 "documentation": ["buildInstructions"], 

41 "keywords": ["keywords"], 

42 "authors": ["author"], 

43 "maintainers": ["maintainer"], 

44 "contributors": ["contributor"], 

45 } 

46 super().__init__( 

47 path, 

48 create_if_not_exists=True, 

49 direct_mappings=mappings, 

50 merge=merge, 

51 pass_validation=pass_validation, 

52 ) 

53 

54 @property 

55 def authors(self): 

56 """Return the only author of the codemeta.json file as list.""" 

57 return self._get_property(self._get_key("publication_authors")) or [] 

58 

59 @authors.setter 

60 def authors(self, authors: list[Person | Entity]) -> None: 

61 """Set the authors of the project.""" 

62 authors_dict = self._people_with_roles(authors) 

63 self._set_property(self._get_key("authors"), authors_dict) 

64 

65 @property 

66 def maintainers(self): 

67 """Return the maintainers of the codemeta.json file.""" 

68 return self._get_property(self._get_key("maintainers")) 

69 

70 @maintainers.setter 

71 def maintainers(self, maintainers: list[Person | Entity]) -> None: 

72 """Set the maintainers of the project.""" 

73 maintainers_dict = [self._from_person(m) for m in maintainers] 

74 self._set_property(self._get_key("maintainers"), maintainers_dict) 

75 

76 @property 

77 def contributors(self): 

78 """Return the contributors of the codemeta.json file.""" 

79 return self._get_property(self._get_key("contributors")) 

80 

81 @contributors.setter 

82 def contributors(self, contributors: list[Person | Entity]) -> None: 

83 """Set the contributors of the project.""" 

84 contributors_dict = self._people_with_roles(contributors) 

85 self._set_property(self._get_key("contributors"), contributors_dict) 

86 

87 def _load(self) -> None: 

88 """Load codemeta.json file.""" 

89 with self.path.open() as f: 

90 self._data = json.load(f, object_pairs_hook=OrderedDict) 

91 

92 def _upgrade_to_v3(self) -> None: 

93 """Normalize an existing CodeMeta file before v3.1 validation.""" 

94 context = self._data.get("@context", []) 

95 context = context if isinstance(context, list) else [context] 

96 context = [item for item in context if item != V2_CONTEXT] 

97 if V3_CONTEXT not in context: 

98 context.insert(0, V3_CONTEXT) 

99 self._data["@context"] = context 

100 

101 for old, new in ( 

102 ("contIntegration", "continuousIntegration"), 

103 ("embargoDate", "embargoEndDate"), 

104 ): 

105 if old in self._data and new not in self._data: 

106 self._data[new] = self._data[old] 

107 self._data.pop(old, None) 

108 

109 def _validate(self) -> None: 

110 """Validate codemeta.json content using pydantic class.""" 

111 if self.pass_validation: 

112 return 

113 loaded_data = self._data 

114 if self.merge: 

115 self._data = deepcopy(self._data) 

116 self._upgrade_to_v3() 

117 try: 

118 invalid_fields = validate_codemeta(self._data) 

119 finally: 

120 self._data = loaded_data 

121 if invalid_fields and self.merge: 

122 raise ValueError( 

123 f"Invalid fields in codemeta.json: {invalid_fields}. Cannot merge with invalid fields." 

124 ) 

125 

126 def _init_new_file(self) -> None: 

127 """Create a new codemeta.json file with bare minimum generic data.""" 

128 data = self._new_data() 

129 # dump to file 

130 with self.path.open("w+", newline="\n") as f: 

131 json.dump(data, f) 

132 

133 def _new_data(self) -> dict[str, Any]: 

134 """Return the bare minimum generic CodeMeta data.""" 

135 return { 

136 "@context": self._default_context, 

137 "@type": "SoftwareSourceCode", 

138 "author": [], 

139 } 

140 

141 def save(self, path: Path | None = None) -> None: 

142 """Save the codemeta.json file.""" 

143 path = path or self.path 

144 logger.debug(f"Saving codemeta.json to {path}") 

145 

146 # copy the _data 

147 data = self._data.copy() 

148 

149 # set license 

150 if "license" in data: 

151 licenses = data["license"] 

152 licenses = licenses if isinstance(licenses, list) else [licenses] 

153 data["license"] = [ 

154 license 

155 if license.startswith("https://spdx.org/licenses/") 

156 else f"https://spdx.org/licenses/{license}" 

157 for license in licenses 

158 ] 

159 

160 # if softwareHelp is set, set url to softwareHelp 

161 if "softwareHelp" in data: 

162 data["url"] = data["softwareHelp"] 

163 

164 with path.open("w", newline="\n") as f: 

165 # codemeta.json indentation is 2 spaces 

166 json.dump(data, f) 

167 

168 @staticmethod 

169 def _from_person(person: Person | Entity) -> dict: 

170 """Convert project metadata person object to codemeta.json dict for person format.""" 

171 if isinstance(person, Person): 

172 person_dict = { 

173 "@type": "Person", 

174 } 

175 if person.given_names: 

176 person_dict["givenName"] = person.given_names 

177 if person.family_names: 

178 person_dict["familyName"] = person.family_names 

179 if person.email: 

180 person_dict["email"] = person.email 

181 if person.orcid: 

182 person_dict["@id"] = str(person.orcid) 

183 person_dict["identifier"] = str(person.orcid) 

184 if person.address: 

185 person_dict["address"] = person.address 

186 if person.affiliation: 

187 person_dict["affiliation"] = person.affiliation 

188 return person_dict 

189 else: 

190 entity_dict = {"@type": "Organization", "name": person.name} 

191 if person.address: 

192 entity_dict["address"] = person.address 

193 if person.email: 

194 entity_dict["email"] = person.email 

195 if person.date_start: 

196 entity_dict["startDate"] = person.date_start.isoformat() 

197 if person.date_end: 

198 entity_dict["endDate"] = person.date_end.isoformat() 

199 if person.website: 

200 entity_dict["@id"] = str(person.website) 

201 entity_dict["identifier"] = str(person.website) 

202 if person.rorid: 

203 entity_dict["@id"] = str(person.rorid) 

204 entity_dict["identifier"] = str(person.rorid) 

205 return entity_dict 

206 

207 @staticmethod 

208 def _role_identifier(person: Person | Entity, person_dict: dict) -> str: 

209 """Return an identifier that can link a CodeMeta role to its person.""" 

210 if identifier := person_dict.get("@id"): 

211 return identifier 

212 identity = person.email or person.full_name 

213 return f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{type(person).__name__}:{identity}')}" 

214 

215 @staticmethod 

216 def _has_contribution_metadata(person: Person | Entity) -> bool: 

217 """Return whether a person has metadata representable by a CodeMeta role.""" 

218 return any( 

219 ( 

220 person.contribution, 

221 person.contribution_types, 

222 person.contribution_begin, 

223 person.contribution_end, 

224 ) 

225 ) 

226 

227 def _people_with_roles(self, people: Sequence[Person | Entity]) -> list[dict]: 

228 """Serialize people and their granular contribution roles.""" 

229 result = [] 

230 for person in people: 

231 person_dict = self._from_person(person) 

232 if not self._has_contribution_metadata(person): 

233 result.append(person_dict) 

234 continue 

235 

236 identifier = self._role_identifier(person, person_dict) 

237 person_dict["@id"] = identifier 

238 result.append(person_dict) 

239 

240 role_names: list[str | None] = [] 

241 if person.contribution: 

242 role_names.append(person.contribution) 

243 role_names.extend( 

244 contribution_type.value 

245 for contribution_type in person.contribution_types or [] 

246 ) 

247 if not role_names: 

248 role_names.append(None) 

249 for role_name in role_names: 

250 role = {"@type": "Role", "schema:author": identifier} 

251 if role_name: 

252 role["roleName"] = role_name 

253 if person.contribution_begin: 

254 role["startDate"] = person.contribution_begin.isoformat() 

255 if person.contribution_end: 

256 role["endDate"] = person.contribution_end.isoformat() 

257 result.append(role) 

258 return result 

259 

260 @staticmethod 

261 def _to_person(person_obj) -> Person | Entity: 

262 """Convert codemeta.json dict or str for person/entity format to project metadata person object.""" 

263 if "name" in person_obj: 

264 entity_obj = {"name": person_obj["name"]} 

265 return Entity(**entity_obj) 

266 else: 

267 person_data = {} 

268 if "givenName" in person_obj: 

269 person_data["given_names"] = person_obj["givenName"].strip() 

270 if "familyName" in person_obj: 

271 person_data["family_names"] = person_obj["familyName"].strip() 

272 if "email" in person_obj: 

273 person_data["email"] = person_obj["email"].strip() 

274 if "@id" in person_obj: 

275 person_data["orcid"] = person_obj["@id"].strip() 

276 if "address" in person_obj: 

277 person_data["address"] = person_obj["address"].strip() 

278 

279 return Person(**person_data) 

280 

281 def _sync_person_list( 

282 self, old: list[Any], new: Sequence[Person | Entity] 

283 ) -> list[Any]: 

284 """Override the _sync_person_list function from ProjectMetadataWriter. 

285 

286 This method wont care about existing persons in codemeta.json file. 

287 

288 Args: 

289 old (List[Any]): existing persons in codemeta.json file, in this case ignored in the output. However, it is necessary to make the function compatible with the parent class. 

290 new (List[Person]): new persons to add to codemeta.json file 

291 

292 Returns: 

293 List[Any]: list of new persons to add to codemeta.json file 

294 

295 """ 

296 return list(new) 

297 

298 def sync(self, metadata: ProjectMetadata) -> None: 

299 """Sync codemeta.json with project metadata. 

300 

301 Use existing sync function from ProjectMetadataWriter but update repository and contributors. 

302 """ 

303 if not self.merge: 

304 self._data = self._new_data() 

305 else: 

306 self._upgrade_to_v3() 

307 self._data["@type"] = "SoftwareSourceCode" 

308 if metadata.authors(): 

309 self._data["author"] = [] 

310 self._data["maintainer"] = [] 

311 self._data["contributor"] = [] 

312 

313 super().sync(metadata) 

314 if metadata.doi: 

315 self._data["identifier"] = f"https://doi.org/{metadata.doi}" 

316 if licenses := metadata.license: 

317 self.license = [ 

318 f"https://spdx.org/licenses/{license.value}" 

319 for license in (licenses if isinstance(licenses, list) else [licenses]) 

320 ] 

321 self.contributors = metadata.contributors() 

322 

323 if "softwareHelp" in self._data: 

324 self._data["url"] = self._data["softwareHelp"]