Coverage for apio/managers/scons_filter.py: 83%

1# This file is part of the Apio project 

2# (C) 2016-present FPGAwars 

3# Project authors: 

4# * Jesús Arroyo (Jesus89) (2016-2019) 

5# * Juan Gonzalez (obijuan) (2019-2024) 

6# * The Apio Team (2024-present) 

7# License: GPLv2 

8 

9"""DOC: TODO""" 

10 

11 

12import re 

13import threading 

14from enum import Enum, unique 

15from typing_extensions import override 

16from apio.common.debug_util import is_debug 

17from apio.common.apio_console import cout, cunstyle, cwrite, cstyle 

18from apio.common.apio_styles import INFO, WARNING, SUCCESS, ERROR 

19 

20 

21# -- A table with line coloring rules. If a line matches any regex, it gets 

22# -- The style of the first regex it matches. Patterns are case insensitive. 

23LINE_COLORING_TABLE = [ 

24 # -- Info patterns 

25 (r"^info:", INFO), 

26 # -- Warning patterns 

27 (r"^warning:", WARNING), 

28 (r"%warning-", WARNING), # Lint/Verilator 

29 # -- Error patterns. 

30 (r"^error:", ERROR), 

31 (r"^%Error:", ERROR), 

32 (r" error: ", ERROR), 

33 (r"fail: ", ERROR), 

34 (r"fatal: ", ERROR), 

35 (r"^fatal error:", ERROR), 

36 (r"assertion failed", ERROR), 

37 # -- Success patterns 

38 (r"is up to date", SUCCESS), 

39 (r"[$]finish called", SUCCESS), 

40 (r"^verify ok$", SUCCESS), 

41 (r"^done$", SUCCESS), 

42] 

43 

44# -- Lines that contain a substring that match any of these regex's are 

45# -- ignored. Regexs are case insensitive. 

46LINE_IGNORE_LIST = [ 

47 # -- Per https://github.com/fpgawars/apio/issues/824 

48 # -- TODO: Remove when fixed. 

49 r"Warning: define gw1n not used in the library", 

50 # -- Per https://github.com/YosysHQ/oss-cad-suite-build/issues/194 

51 r"Numpy is not available, performance will be degraded", 

52 r"Msgspec is not available, performance will be degraded", 

53 r"fastcrc is not available, performance will be degraded", 

54 # -- For openFPGAloader 

55 r"Verifying write [(]May take time[)]", 

56] 

57 

58 

59@unique 

60class PipeId(Enum): 

61 """Represent the two output streams from the scons subprocess.""" 

62 

63 STDOUT = 1 

64 STDERR = 2 

65 

66 

67@unique 

68class RangeEvents(Enum): 

69 """An stdout/err line can trigger one of these events, when detecting a 

70 range of lines.""" 

71 

72 START_BEFORE = 1 # Range starts before the current line. 

73 START_AFTER = 2 # Range starts after the current line. 

74 END_BEFORE = 3 # Range ends before the current line. 

75 END_AFTER = 4 # Range ends, after the current line. 

76 

77 

78class RangeDetector: 

79 """Base detector of a range of lines within the sequence of stdout/err 

80 lines recieves from the scons subprocess.""" 

81 

82 def __init__(self): 

83 self._in_range = False 

84 

85 def update(self, pipe_id: PipeId, line: str) -> bool: 

86 """Updates the range detector with the next stdout/err line. 

87 return True iff detector classified this line to be within a range.""" 

88 

89 prev_state = self._in_range 

90 event = self.classify_line(pipe_id, line) 

91 

92 if event == RangeEvents.START_BEFORE: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 self._in_range = True 

94 return self._in_range 

95 

96 if event == RangeEvents.START_AFTER: 

97 self._in_range = True 

98 return prev_state 

99 

100 if event == RangeEvents.END_BEFORE: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 self._in_range = False 

102 return self._in_range 

103 

104 if event == RangeEvents.END_AFTER: 

105 self._in_range = False 

106 return prev_state 

107 

108 assert event is None, event 

109 return self._in_range 

110 

111 def classify_line( 

112 self, pipe_id: PipeId, line: str 

113 ) -> RangeEvents | None: # pragma: no cover 

114 """Tests if the next stdout/err line affects the range begin/end. 

115 Subclasses should implement this with the necessary logic for the 

116 range that is being detected. 

117 Returns the event of None if no event.""" 

118 raise NotImplementedError("Should be implemented by a subclass") 

119 

120 

121class PnrRangeDetector(RangeDetector): 

122 """Implements a RangeDetector for the nextpnr command verbose 

123 log lines.""" 

124 

125 @override 

126 def classify_line(self, pipe_id: PipeId, line: str) -> RangeEvents | None: 

127 # -- Break line into words. 

128 tokens = line.split() 

129 

130 # -- Range start: A nextpnr command on stdout without 

131 # -- the -q (quiet) flag. 

132 # -- 

133 # -- IMPORTANT: Each of the supported architecture has a different 

134 # -- nextpnr command, including 'nextpnr', 'nextpnr-ecp5', and 

135 # -- 'nextpnr-himbaechel'. 

136 if ( 

137 pipe_id == PipeId.STDOUT 

138 and line.startswith("nextpnr") 

139 and "-q" not in tokens 

140 ): 

141 return RangeEvents.START_AFTER 

142 

143 # Range end: The end message of nextpnr. 

144 if pipe_id == PipeId.STDERR and "Program finished normally." in line: 

145 return RangeEvents.END_AFTER 

146 

147 return None 

148 

149 

150class SconsFilter: 

151 """Implements the filtering and printing of the stdout/err streams of the 

152 scons subprocess. Accepts a line one at a time, detects lines ranges of 

153 interest, mutates and colors the lines where applicable, and print to 

154 stdout.""" 

155 

156 def __init__(self, colors_enabled: bool): 

157 self.colors_enabled = colors_enabled 

158 self._pnr_detector = PnrRangeDetector() 

159 

160 # self._iverilog_detector = IVerilogRangeDetector() 

161 # self._iceprog_detector = IceProgRangeDetector() 

162 

163 # -- We cache the values to avoid reevaluating sys env. 

164 self._is_debug = is_debug(3) 

165 self._is_verbose_debug = is_debug(5) 

166 

167 # -- Accumulates string pieces until we write and flush them. This 

168 # -- mechanism is used to display progress bar correctly, Writing the 

169 # -- erasure string only when a new value is available. 

170 self._output_bfr: str = "" 

171 

172 # -- The stdout and stderr are called from independent threads, so we 

173 # -- protect the handling method with this lock. 

174 # -- 

175 # -- We don't protect the third thread which is main(). We hope that 

176 # -- it doesn't any print console output while these two threads are 

177 # -- active, otherwise it can mingle the output. 

178 self._thread_lock = threading.Lock() 

179 

180 def on_stdout_line(self, line: str, terminator: str) -> None: 

181 """Stdout pipe calls this on each line. Called from the stdout thread 

182 in AsyncPipe.""" 

183 with self._thread_lock: 

184 self.on_line(PipeId.STDOUT, line, terminator) 

185 

186 def on_stderr_line(self, line: str, terminator: str) -> None: 

187 """Stderr pipe calls this on each line. Called from the stderr thread 

188 in AsyncPipe.""" 

189 with self._thread_lock: 

190 self.on_line(PipeId.STDERR, line, terminator) 

191 

192 @staticmethod 

193 def _assign_line_color( 

194 line: str, 

195 patterns: list[tuple[str, str]], 

196 default_color: str | None = None, 

197 ) -> str | None: 

198 """Assigns a color for a given line using a list of (regex, color) 

199 pairs. Returns the color of the first matching regex (case 

200 insensitive), or default_color if none match. 

201 """ 

202 for regex, color in patterns: 

203 if re.search(regex, line, re.IGNORECASE): 

204 return color 

205 return default_color 

206 

207 def _output_line( 

208 self, line: str, style: str | None, terminator: str 

209 ) -> None: 

210 """Output a line. If a style is given, force that style, otherwise, 

211 pass on any color information it may have. The implementation takes 

212 into consideration progress bars such as when uploading with the 

213 iceprog programmer. These progress bar require certain timing between 

214 the chars to have sufficient time to display the text before erasing 

215 it.""" 

216 

217 # -- Apply style if needed. 

218 if style: 

219 line_part = cstyle(cunstyle(line), style=style) 

220 else: 

221 line_part = line 

222 

223 # -- Get line conditions. 

224 is_white = len(line.strip()) == 0 

225 is_cr = terminator == "\r" 

226 

227 if not is_cr: 227 ↛ 232line 227 didn't jump to line 232 because the condition on line 227 was always true

228 # -- Terminator is EOF or \n. We flush everything. 

229 self._output_bfr += line_part + terminator 

230 leftover = "" 

231 flush = True 

232 elif is_white: 

233 # -- Terminator is \r and line is white space (progress bar 

234 # -- eraser). We queue and and wait for the updated text. 

235 self._output_bfr += line_part + terminator 

236 leftover = "" 

237 flush = False 

238 else: 

239 # -- Terminator is \r and line has actual text, we flush it out 

240 # -- but save queue the \r because on windows 10 cmd it clears the 

241 # -- line(?) 

242 self._output_bfr += line_part 

243 leftover = terminator 

244 flush = True 

245 

246 if flush: 246 ↛ 252line 246 didn't jump to line 252 because the condition on line 246 was always true

247 # -- Flush the buffer and queue the optional leftover terminator. 

248 cwrite(self._output_bfr) 

249 self._output_bfr = leftover 

250 else: 

251 # -- We just queued. Should have no leftover here. 

252 assert not leftover 

253 

254 def _ignore_line(self, line: str) -> None: 

255 """Handle an ignored line. It's dumped if in debug mode.""" 

256 if self._is_debug: 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true

257 cout(f"IGNORED: {line}") 

258 

259 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None: 

260 """A shared handler for stdout/err lines from the scons sub process. 

261 The handler writes both stdout and stderr lines to stdout, possibly 

262 with modifications such as text deletion, coloring, and cursor 

263 directives. 

264 

265 For the possible values of terminator, see AsyncPipe.__init__(). 

266 

267 NOTE: Ideally, the program specific patterns such as for Fumo and 

268 Iceprog should should be condition by a range detector for lines that 

269 came from that program. That is to minimize the risk of matching lines 

270 from other programs. See the PNR detector for an example. 

271 """ 

272 

273 if self._is_verbose_debug: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true

274 cout( 

275 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]", 

276 style=INFO, 

277 ) 

278 

279 # -- Update the range detectors. 

280 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line) 

281 

282 # -- If the line match any of the ignore patterns ignore it. 

283 for regex in LINE_IGNORE_LIST: 

284 if re.search(regex, line, re.IGNORECASE): 

285 self._ignore_line(line) 

286 return 

287 

288 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where 

289 # -- each line starts with "Info: " 

290 if ( 290 ↛ 295line 290 didn't jump to line 295 because the condition on line 290 was never true

291 pipe_id == PipeId.STDERR 

292 and in_pnr_verbose_range 

293 and line.startswith("Info: ") 

294 ): 

295 line = line[6:] 

296 

297 # -- Output the line in the appropriate style. 

298 line_color = self._assign_line_color(line, LINE_COLORING_TABLE) 

299 self._output_line(line, line_color, terminator)