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 CI callers distinguish invalid PEX25D (DIAGNOSTIC_ERRORS) from invocation 

62 or I/O problems (USAGE). NOT_IMPLEMENTED identifies unsupported operations; 

63 INTERNAL_ERROR identifies unexpected plugin failures. 

64 """ 

65 

66 OK = 0 

67 DIAGNOSTIC_ERRORS = 1 

68 USAGE = 2 

69 NOT_IMPLEMENTED = 3 

70 INTERNAL_ERROR = 4 

71 

72 

73class Severity(IntEnum): 

74 NOTE = 1 

75 WARNING = 2 

76 ERROR = 3 

77 

78 @property 

79 def proto_name(self) -> str: 

80 return f"SEVERITY_{self.name}" 

81 

82 

83class Tier(IntEnum): 

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

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

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

87 

88 @property 

89 def proto_name(self) -> str: 

90 return f"TIER_{self.name}" 

91 

92 

93class DiagnosticsFormat(StrEnum): 

94 """ 

95 How validator output is rendered. 

96 

97 ``human`` 

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

99 ``json`` 

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

101 of ``kpex.pex25d.DiagnosticList``. Stable. 

102 ``pb`` 

103 Binary ``kpex.pex25d.DiagnosticList``. Stable. 

104 """ 

105 

106 HUMAN = 'human' 

107 JSON = 'json' 

108 PB = 'pb' 

109 

110 DEFAULT = 'human' 

111 

112 

113@dataclass(frozen=True) 

114class SourceRef: 

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

116 

117 file: Optional[str] = None 

118 line: Optional[int] = None 

119 column: Optional[int] = None 

120 

121 def __str__(self) -> str: 

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

123 if self.line is not None: 

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

125 if self.column is not None: 

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

127 return ':'.join(parts) 

128 

129 

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

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

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

133 return None 

134 try: 

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

136 return None 

137 except ValueError: 

138 return None 

139 source = record.source 

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

141 line=source.line or None, 

142 column=source.column or None) 

143 

144 

145@dataclass(frozen=True) 

146class Diagnostic: 

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

148 

149 code: str 

150 severity: Severity 

151 tier: Tier 

152 message: str 

153 source: Optional[SourceRef] = None 

154 related: Sequence[SourceRef] = () 

155 

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

157 d: Dict[str, Any] = { 

158 'code': self.code, 

159 'severity': self.severity.proto_name, 

160 'tier': self.tier.proto_name, 

161 'message': self.message, 

162 } 

163 if self.source is not None: 

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

165 if self.related: 

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

167 return d 

168 

169 

170@dataclass 

171class DiagnosticsReport: 

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

173 

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

175 warnings_are_errors: bool = False 

176 

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

178 self.diagnostics.append(diagnostic) 

179 

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

181 self.diagnostics.extend(diagnostics) 

182 

183 @property 

184 def num_errors(self) -> int: 

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

186 

187 @property 

188 def num_warnings(self) -> int: 

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

190 

191 @property 

192 def exit_code(self) -> ExitCode: 

193 if self.num_errors: 

194 return ExitCode.DIAGNOSTIC_ERRORS 

195 if self.warnings_are_errors and self.num_warnings: 

196 return ExitCode.DIAGNOSTIC_ERRORS 

197 return ExitCode.OK 

198 

199 # ---------------------------------------------------------------- rendering 

200 

201 def render(self, 

202 format: DiagnosticsFormat, 

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

204 match format: 

205 case DiagnosticsFormat.HUMAN: 

206 self._render_human() 

207 case DiagnosticsFormat.JSON: 

208 self._render_json(stream or sys.stdout) 

209 case DiagnosticsFormat.PB: 

210 self._render_pb(stream) 

211 case _: 

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

213 

214 def _render_human(self) -> None: 

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

216 for d in self.diagnostics: 

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

218 

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

220 for d in by_tier[tier]: 

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

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

223 match d.severity: 

224 case Severity.ERROR: 

225 error(line) 

226 case Severity.WARNING: 

227 warning(line) 

228 case _: 

229 info(line) 

230 for r in d.related: 

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

232 

233 if not self.diagnostics: 

234 info("No diagnostics.") 

235 else: 

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

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

238 

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

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

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

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

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

244 else: 

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

246 stream.flush() 

247 

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

249 from .protobuf import pex25d_diagnostics_pb2 

250 

251 pb2 = pex25d_diagnostics_pb2() 

252 diagnostic_list = pb2.DiagnosticList() 

253 for d in self.diagnostics: 

254 pb = diagnostic_list.diagnostics.add() 

255 pb.code = d.code 

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

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

258 pb.message = d.message 

259 if d.source is not None: 

260 _fill_source_ref(pb.source, d.source) 

261 for r in d.related: 

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

263 

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

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

266 out.write(diagnostic_list.SerializeToString()) 

267 out.flush() 

268 

269 

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

271 if ref.file is not None: 

272 pb.file = ref.file 

273 if ref.line is not None: 

274 pb.line = ref.line 

275 if ref.column is not None: 

276 pb.column = ref.column 

277 

278 

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

280 import io 

281 return (io.TextIOBase,) 

282 

283 

284@contextlib.contextmanager 

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

286 """ 

287 Yield the stream diagnostics should be written to. 

288 

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

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

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

292 """ 

293 yield sys.stderr if writes_to_stdout else sys.stdout