Coverage for klayout_pex/pex25d/diagnostics.py: 71%

1# 

2# -------------------------------------------------------------------------------- 

3# SPDX-FileCopyrightText: 2024-2026 Martin Jan Köhler and Harald Pretl 

4# Johannes Kepler University, Institute for Integrated Circuits. 

5# 

6# This file is part of KPEX  

7# (see https://github.com/iic-jku/klayout-pex). 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21# SPDX-License-Identifier: GPL-3.0-or-later 

22# -------------------------------------------------------------------------------- 

23# 

24 

25""" 

26Validator output: coded, positioned diagnostics, and how they reach the user. 

27 

28This can be used to check a custom PEX25D *writer* against the reference validator without depending on kpex internals 

29or on the exact wording of a message — so ``--diagnostics json`` and ``--diagnostics pb`` are first-class 

30outputs here, not a debugging afterthought. 

31 

32Diagnostics are modelled as a plain dataclass rather than as the generated protobuf message, for two reasons: 

33 1) rendering must work on a checkout where the ``*_pb2`` modules have not been generated yet, and 

34 2) the human renderer wants to group and sort in ways the wire format has no opinion about. 

35 The conversion to``kpex.pex25d.Diagnostic`` happens only for ``--diagnostics pb``. 

36""" 

37 

38from __future__ import annotations 

39 

40import contextlib 

41import dataclasses 

42import enum 

43import json 

44import sys 

45from dataclasses import dataclass, field 

46from enum import IntEnum, StrEnum 

47from typing import * 

48 

49from ..log import ( 

50 error, 

51 warning, 

52 info, 

53 subproc, 

54) 

55 

56 

57class ExitCode(IntEnum): 

58 """ 

59 Process exit codes, shared by ``kpex`` and ``pex25d``. 

60 

61 Distinguishing OK from ERRORS from USAGE matters for the CI use case: 

62 a writer-conformance run wants to distinguish 

63 1) "your file is wrong" 

64 2) "you called me wrong" 

65 3) a crash 

66 """ 

67 

68 OK = 0 

69 DIAGNOSTIC_ERRORS = 1 

70 USAGE = 2 

71 NOT_IMPLEMENTED = 3 

72 

73 

74class Severity(IntEnum): 

75 NOTE = 1 

76 WARNING = 2 

77 ERROR = 3 

78 

79 @property 

80 def proto_name(self) -> str: 

81 return f"SEVERITY_{self.name}" 

82 

83 

84class Tier(IntEnum): 

85 SYNTAX = 1 # arity, clause order, literal not a grid multiple, unknown record 

86 SEMANTIC = 2 # name resolution, acyclic WRAPS, depth ties, ... 

87 GEOMETRIC = 3 # ring and box shape, hole containment, overlap (--strict) 

88 

89 @property 

90 def proto_name(self) -> str: 

91 return f"TIER_{self.name}" 

92 

93 

94class DiagnosticsFormat(StrEnum): 

95 """ 

96 How validator output is rendered. 

97 

98 ``human`` 

99 Rich text for a terminal. Not stable; never parse it. 

100 ``json`` 

101 One JSON object with a ``diagnostics`` array, mirroring the field names 

102 of ``kpex.pex25d.DiagnosticList``. Stable. 

103 ``pb`` 

104 Binary ``kpex.pex25d.DiagnosticList``. Stable. 

105 """ 

106 

107 HUMAN = 'human' 

108 JSON = 'json' 

109 PB = 'pb' 

110 

111 DEFAULT = 'human' 

112 

113 

114@dataclass(frozen=True) 

115class SourceRef: 

116 """Position in a PEX25D source file. Mirrors ``kpex.pex25d.SourceRef``.""" 

117 

118 file: Optional[str] = None 

119 line: Optional[int] = None 

120 column: Optional[int] = None 

121 

122 def __str__(self) -> str: 

123 parts = [self.file or '<input>'] 

124 if self.line is not None: 

125 parts.append(str(self.line)) 

126 if self.column is not None: 

127 parts.append(str(self.column)) 

128 return ':'.join(parts) 

129 

130 

131def source_ref(record: Any) -> Optional[SourceRef]: 

132 """The ``SourceRef`` of a protobuf record, or ``None`` if it carries none.""" 

133 if record is None or not hasattr(record, 'HasField'): 

134 return None 

135 try: 

136 if not record.HasField('source'): 

137 return None 

138 except ValueError: 

139 return None 

140 source = record.source 

141 return SourceRef(file=source.file or None, 

142 line=source.line or None, 

143 column=source.column or None) 

144 

145 

146@dataclass(frozen=True) 

147class Diagnostic: 

148 """One validator finding. Mirrors ``kpex.pex25d.Diagnostic``.""" 

149 

150 code: str 

151 severity: Severity 

152 tier: Tier 

153 message: str 

154 source: Optional[SourceRef] = None 

155 related: Sequence[SourceRef] = () 

156 

157 def as_json_dict(self) -> Dict[str, Any]: 

158 d: Dict[str, Any] = { 

159 'code': self.code, 

160 'severity': self.severity.proto_name, 

161 'tier': self.tier.proto_name, 

162 'message': self.message, 

163 } 

164 if self.source is not None: 

165 d['source'] = dataclasses.asdict(self.source) 

166 if self.related: 

167 d['related'] = [dataclasses.asdict(r) for r in self.related] 

168 return d 

169 

170 

171@dataclass 

172class DiagnosticsReport: 

173 """Accumulates diagnostics and decides the process exit code.""" 

174 

175 diagnostics: List[Diagnostic] = field(default_factory=list) 

176 warnings_are_errors: bool = False 

177 

178 def add(self, diagnostic: Diagnostic) -> None: 

179 self.diagnostics.append(diagnostic) 

180 

181 def extend(self, diagnostics: Iterable[Diagnostic]) -> None: 

182 self.diagnostics.extend(diagnostics) 

183 

184 @property 

185 def num_errors(self) -> int: 

186 return sum(1 for d in self.diagnostics if d.severity == Severity.ERROR) 

187 

188 @property 

189 def num_warnings(self) -> int: 

190 return sum(1 for d in self.diagnostics if d.severity == Severity.WARNING) 

191 

192 @property 

193 def exit_code(self) -> ExitCode: 

194 if self.num_errors: 

195 return ExitCode.DIAGNOSTIC_ERRORS 

196 if self.warnings_are_errors and self.num_warnings: 

197 return ExitCode.DIAGNOSTIC_ERRORS 

198 return ExitCode.OK 

199 

200 # ---------------------------------------------------------------- rendering 

201 

202 def render(self, 

203 format: DiagnosticsFormat, 

204 stream: Optional[IO[Any]] = None) -> None: 

205 match format: 

206 case DiagnosticsFormat.HUMAN: 

207 self._render_human() 

208 case DiagnosticsFormat.JSON: 

209 self._render_json(stream or sys.stdout) 

210 case DiagnosticsFormat.PB: 

211 self._render_pb(stream) 

212 case _: 

213 raise ValueError(f"Unknown diagnostics format {format}") 

214 

215 def _render_human(self) -> None: 

216 by_tier: Dict[Tier, List[Diagnostic]] = {} 

217 for d in self.diagnostics: 

218 by_tier.setdefault(d.tier, []).append(d) 

219 

220 for tier in sorted(by_tier.keys()): 

221 for d in by_tier[tier]: 

222 where = f"{d.source} " if d.source else '' 

223 line = f"{where}{d.code}: {d.message}" 

224 match d.severity: 

225 case Severity.ERROR: 

226 error(line) 

227 case Severity.WARNING: 

228 warning(line) 

229 case _: 

230 info(line) 

231 for r in d.related: 

232 subproc(f" … see also {r}") 

233 

234 if not self.diagnostics: 

235 info("No diagnostics.") 

236 else: 

237 info(f"{self.num_errors} error(s), {self.num_warnings} warning(s), " 

238 f"{len(self.diagnostics)} diagnostic(s) total") 

239 

240 def _render_json(self, stream: IO[Any]) -> None: 

241 payload = {'diagnostics': [d.as_json_dict() for d in self.diagnostics]} 

242 text = json.dumps(payload, indent=2, ensure_ascii=False) 

243 if hasattr(stream, 'buffer') or isinstance(stream, io_text_types()): 

244 stream.write(text + '\n') 

245 else: 

246 stream.write((text + '\n').encode('utf-8')) 

247 stream.flush() 

248 

249 def _render_pb(self, stream: Optional[IO[Any]]) -> None: 

250 from .protobuf import pex25d_diagnostics_pb2 

251 

252 pb2 = pex25d_diagnostics_pb2() 

253 diagnostic_list = pb2.DiagnosticList() 

254 for d in self.diagnostics: 

255 pb = diagnostic_list.diagnostics.add() 

256 pb.code = d.code 

257 pb.severity = pb2.Diagnostic.Severity.Value(d.severity.proto_name) 

258 pb.tier = pb2.Diagnostic.Tier.Value(d.tier.proto_name) 

259 pb.message = d.message 

260 if d.source is not None: 

261 _fill_source_ref(pb.source, d.source) 

262 for r in d.related: 

263 _fill_source_ref(pb.related.add(), r) 

264 

265 out = stream if stream is not None else sys.stdout.buffer 

266 out = getattr(out, 'buffer', out) 

267 out.write(diagnostic_list.SerializeToString()) 

268 out.flush() 

269 

270 

271def _fill_source_ref(pb: Any, ref: SourceRef) -> None: 

272 if ref.file is not None: 

273 pb.file = ref.file 

274 if ref.line is not None: 

275 pb.line = ref.line 

276 if ref.column is not None: 

277 pb.column = ref.column 

278 

279 

280def io_text_types() -> Tuple[type, ...]: 

281 import io 

282 return (io.TextIOBase,) 

283 

284 

285@contextlib.contextmanager 

286def diagnostics_stream(writes_to_stdout: bool) -> Iterator[IO[Any]]: 

287 """ 

288 Yield the stream diagnostics should be written to. 

289 

290 When the verb's *artifact* output goes to stdout, diagnostics must not: 

291 a PEX25D file with a JSON diagnostics blob stapled to the front is not a PEX25D file. 

292 In that case diagnostics go to stderr, which is also where the rich logger already writes. 

293 """ 

294 yield sys.stderr if writes_to_stdout else sys.stdout