Coverage for python/lsst/images/schema_docs.py: 96%

142 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-25 09:03 +0000

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11"""Generate Sphinx pages for the frozen JSON schemas. 

12 

13Called from ``doc/conf.py`` at the start of every documentation build. Each 

14frozen ``schemas/{name}/{name}-{version}.json`` file becomes a page at 

15``doc/schemas/{name}-{version}/index.rst`` so that the canonical schema URL 

16``https://images.lsst.io/schemas/{name}-{version}`` resolves (LSST the Docs 

17redirects extensionless directory paths to their ``index.html``), and the raw 

18JSON is staged for publication alongside it via ``html_extra_path``. 

19""" 

20 

21from __future__ import annotations 

22 

23__all__ = ("generate_schema_docs",) 

24 

25import json 

26import re 

27import shutil 

28from pathlib import Path 

29from typing import Any 

30 

31from .diagram import build_graph, make_policy, render 

32from .serialization import ( 

33 SCHEMA_URL_BASE, 

34 ArrayReferenceModel, 

35 class_for_schema, 

36 is_development_version, 

37 parameterize_tree, 

38) 

39 

40 

41def _link_map(schema: dict[str, Any], available_stems: set[str]) -> dict[str, str]: 

42 """Map ``$defs`` keys to rst cross-references for definitions that are 

43 themselves published schemas. 

44 

45 Nested `~lsst.images.serialization.ArchiveTree` models carry their 

46 canonical URL in the ``x-lsst-schema-url`` key of their ``$defs`` entry 

47 (``$id`` would start a new resolution scope), so a composite schema's 

48 page can hyperlink directly to its sub-schemas (e.g. ``visit_image`` → 

49 ``mask``): a relative page reference when the sub-schema is hosted on 

50 this site, or its absolute URL when it is published elsewhere. 

51 """ 

52 links: dict[str, str] = {} 

53 for key, definition in schema.get("$defs", {}).items(): 

54 if not isinstance(definition, dict): 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true

55 continue 

56 schema_id = definition.get("x-lsst-schema-url", "") 

57 if not schema_id: 

58 continue 

59 stem = schema_id.rstrip("/").rsplit("/", 1)[-1] 

60 title = definition.get("title", stem) 

61 version = stem.rpartition("-")[2] 

62 if stem in available_stems: 

63 links[key] = f":doc:`{title} <../{stem}/index>`" 

64 elif is_development_version(version): 

65 # A development sub-schema is unpublished and has no page. 

66 links[key] = title 

67 else: 

68 links[key] = f"`{title} <{schema_id}>`__" 

69 return links 

70 

71 

72def _type_summary(prop: dict[str, Any], links: dict[str, str]) -> str: 

73 """Return a short human-readable type description for a property, 

74 hyperlinking types that have their own schema page. 

75 """ 

76 if "$ref" in prop: 

77 key = prop["$ref"].rsplit("/", 1)[-1] 

78 return links.get(key, key) 

79 for key in ("anyOf", "oneOf", "allOf"): 

80 if key in prop: 

81 return " | ".join(_type_summary(sub, links) for sub in prop[key]) 

82 if prop.get("type") == "array": 

83 items = prop.get("items") 

84 if isinstance(items, dict): 84 ↛ 86line 84 didn't jump to line 86 because the condition on line 84 was always true

85 return f"array of {_type_summary(items, links)}" 

86 return "array" 

87 if "type" in prop: 

88 return str(prop["type"]) 

89 if "const" in prop: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true

90 return repr(prop["const"]) 

91 return "any" 

92 

93 

94_DEFAULT_ROLE_RE = re.compile(r"(?<!`)`([^`]+)`(?!`)") 

95"""Single-backtick (default-role) references in docstring-derived text.""" 

96 

97 

98def _reference_to_literal(match: re.Match[str]) -> str: 

99 """Rewrite a default-role Python reference as an inline literal. 

100 

101 Model docstrings use `X`, `~pkg.mod.X`, and `.X` references that cannot 

102 resolve as py:obj targets on the generated schema pages; render them as 

103 code, applying the ``~`` convention of showing only the last component. 

104 """ 

105 target = match.group(1) 

106 if target.startswith("~"): 

107 target = target[1:].rsplit(".", 1)[-1] 

108 return f"``{target.lstrip('.')}``" 

109 

110 

111def _one_line(text: str) -> str: 

112 """Collapse ``text`` to a single line safe for a list-table cell.""" 

113 return " ".join(text.split()) 

114 

115 

116def _description_text(text: str) -> str: 

117 """Return docstring-derived description text as a single rst line with 

118 Python references neutralized. 

119 """ 

120 return _DEFAULT_ROLE_RE.sub(_reference_to_literal, _one_line(text)) 

121 

122 

123def _root_body(schema: dict[str, Any]) -> dict[str, Any]: 

124 """Return the mapping holding the schema's description and properties. 

125 

126 A recursive model's document root is just a ``$ref`` into ``$defs``, so 

127 the referenced definition holds the interesting content. 

128 """ 

129 if "$ref" in schema and "properties" not in schema: 

130 key = schema["$ref"].rsplit("/", 1)[-1] 

131 target = schema.get("$defs", {}).get(key) 

132 if isinstance(target, dict): 132 ↛ 134line 132 didn't jump to line 134 because the condition on line 132 was always true

133 return target 

134 return schema 

135 

136 

137def _field_table(body: dict[str, Any], links: dict[str, str]) -> list[str]: 

138 """Return rst list-table lines for the schema's top-level properties.""" 

139 properties: dict[str, Any] = body.get("properties", {}) 

140 if not properties: 

141 return ["This schema has no top-level properties."] 

142 required = set(body.get("required", [])) 

143 lines = [ 

144 ".. list-table::", 

145 " :header-rows: 1", 

146 " :widths: 20 25 10 45", 

147 "", 

148 " * - Field", 

149 " - Type", 

150 " - Required", 

151 " - Description", 

152 ] 

153 for name, prop in properties.items(): 

154 lines.append(f" * - ``{name}``") 

155 lines.append(f" - {_one_line(_type_summary(prop, links))}") 

156 lines.append(f" - {'yes' if name in required else 'no'}") 

157 lines.append(f" - {_description_text(prop.get('description', ''))}") 

158 return lines 

159 

160 

161def _mermaid_lines(name: str, version: str) -> list[str]: 

162 """Return a mermaid composition-diagram block for the schema, or a 

163 superseded-version note when the live model no longer matches. 

164 """ 

165 cls = class_for_schema(name) 

166 if cls is None or cls.SCHEMA_VERSION != version: 

167 note = "This schema version has been superseded." 

168 if cls is not None: 

169 note = f"This schema version has been superseded; the current version is {cls.SCHEMA_VERSION}." 

170 return [ 

171 ".. note::", 

172 "", 

173 f" {note}", 

174 ] 

175 graph = build_graph(parameterize_tree(cls, ArrayReferenceModel), policy=make_policy()) 

176 lines = ["Composition", "===========", "", ".. mermaid::", ""] 

177 lines.extend(f" {line}" for line in render(graph, "mermaid").splitlines()) 

178 return lines 

179 

180 

181def _version_key(version: str) -> tuple[int, ...]: 

182 """Return a numeric sort key for a ``major.minor.patch`` version string, 

183 so that e.g. ``1.0.10`` sorts after ``1.0.2``. 

184 """ 

185 return tuple(int(part) for part in version.split(".")) 

186 

187 

188def _family_page( 

189 name: str, versions: list[str], latest_schema: dict[str, Any], available_stems: set[str] 

190) -> str: 

191 """Return the rst source for a schema family page. 

192 

193 The family page lists every published version of one schema (newest 

194 first, with the current version marked), renders the latest version's 

195 content inline, and owns the nav toctree for the version pages, so the 

196 top-level schema index only grows when a new schema is added, not on 

197 every version bump. It is published at the versionless URL 

198 ``{SCHEMA_URL_BASE}/{name}``. 

199 

200 Parameters 

201 ---------- 

202 name 

203 Schema name. 

204 versions 

205 Published versions of the schema, newest first. 

206 latest_schema 

207 Parsed frozen schema of the newest version, rendered inline. 

208 available_stems 

209 ``{name}-{version}`` stems of every published schema, for 

210 cross-linking sub-schemas that have their own pages. 

211 """ 

212 cls = class_for_schema(name) 

213 current = cls.SCHEMA_VERSION if cls is not None else None 

214 lines = [ 

215 "#" * len(name), 

216 name, 

217 "#" * len(name), 

218 "", 

219 _description_text(_root_body(latest_schema).get("description", "")), 

220 "", 

221 "Versions", 

222 "========", 

223 "", 

224 ".. list-table::", 

225 " :header-rows: 1", 

226 "", 

227 " * - Version", 

228 " - Status", 

229 ] 

230 for version in versions: 

231 lines.append(f" * - :doc:`{version} <../{name}-{version}/index>`") 

232 lines.append(f" - {'current' if version == current else 'superseded'}") 

233 # Render the latest version's content inline so the landing page shows the 

234 # schema itself, not just a list of links to click through. 

235 lines.append("") 

236 lines.extend(_content_lines(name, versions[0], latest_schema, available_stems)) 

237 lines.extend(["", ".. toctree::", " :hidden:", ""]) 

238 lines.extend(f" {version} <../{name}-{version}/index>" for version in versions) 

239 lines.append("") 

240 return "\n".join(lines) 

241 

242 

243def _content_lines(name: str, version: str, schema: dict[str, Any], available_stems: set[str]) -> list[str]: 

244 """Return the rst body shared by the version page and the family page: 

245 canonical URL, raw-JSON link, composition diagram, and field table. 

246 """ 

247 links = _link_map(schema, available_stems) 

248 body = _root_body(schema) 

249 canonical_url = schema.get("$id", f"{SCHEMA_URL_BASE}/{name}-{version}") 

250 lines = [ 

251 f"- Canonical URL: ``{canonical_url}``", 

252 f"- `Raw JSON schema <../{name}-{version}.json>`__", 

253 "", 

254 ] 

255 lines.extend(_mermaid_lines(name, version)) 

256 lines.extend(["", "Fields", "======", ""]) 

257 lines.extend(_field_table(body, links)) 

258 return lines 

259 

260 

261def _schema_page(name: str, version: str, schema: dict[str, Any], available_stems: set[str]) -> str: 

262 """Return the rst source for one schema version page.""" 

263 title = f"{name} {version}" 

264 lines = [ 

265 "#" * len(title), 

266 title, 

267 "#" * len(title), 

268 "", 

269 _description_text(_root_body(schema).get("description", "")), 

270 "", 

271 ] 

272 lines.extend(_content_lines(name, version, schema, available_stems)) 

273 lines.append("") 

274 return "\n".join(lines) 

275 

276 

277def generate_schema_docs(schema_dir: Path, page_dir: Path, extra_dir: Path) -> None: 

278 """Generate one Sphinx page per frozen schema file. 

279 

280 Parameters 

281 ---------- 

282 schema_dir 

283 Directory of frozen ``{name}/{name}-{version}.json`` files. 

284 page_dir 

285 Output directory for generated rst; wiped and fully regenerated. 

286 extra_dir 

287 Staging directory for ``html_extra_path``; receives a ``schemas/`` 

288 subdirectory with copies of the raw JSON files. 

289 """ 

290 if page_dir.exists(): 

291 shutil.rmtree(page_dir) 

292 page_dir.mkdir(parents=True) 

293 extra_schemas = extra_dir / "schemas" 

294 if extra_schemas.exists(): 

295 shutil.rmtree(extra_schemas) 

296 extra_schemas.mkdir(parents=True) 

297 paths = sorted(schema_dir.glob("*/*.json")) 

298 available_stems = {path.stem for path in paths} 

299 families: dict[str, list[str]] = {} 

300 schemas_by_stem: dict[str, dict[str, Any]] = {} 

301 for path in paths: 

302 name, _, version = path.stem.rpartition("-") 

303 schema = json.loads(path.read_text()) 

304 entry_dir = page_dir / path.stem 

305 entry_dir.mkdir() 

306 (entry_dir / "index.rst").write_text(_schema_page(name, version, schema, available_stems)) 

307 shutil.copyfile(path, extra_schemas / path.name) 

308 families.setdefault(name, []).append(version) 

309 schemas_by_stem[path.stem] = schema 

310 for name, versions in families.items(): 

311 versions.sort(key=_version_key, reverse=True) 

312 family_dir = page_dir / name 

313 family_dir.mkdir() 

314 (family_dir / "index.rst").write_text( 

315 _family_page(name, versions, schemas_by_stem[f"{name}-{versions[0]}"], available_stems) 

316 ) 

317 index_lines = [ 

318 "#######", 

319 "Schemas", 

320 "#######", 

321 "", 

322 "JSON schemas for the ``lsst.images`` serialization data models.", 

323 "Each schema's page lists its published versions; every version page links the raw JSON schema.", 

324 "See :ref:`lsst.images-schema-versioning` for the versioning rules.", 

325 "", 

326 ".. toctree::", 

327 " :maxdepth: 1", 

328 "", 

329 *(f" {name}/index" for name in sorted(families)), 

330 "", 

331 ] 

332 (page_dir / "index.rst").write_text("\n".join(index_lines))