Coverage for apio/managers/scons_filter.py: 82%
102 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 03:53 +0000
1"""DOC: TODO"""
3# -*- coding: utf-8 -*-
4# -- This file is part of the Apio project
5# -- (C) 2016-2019 FPGAwars
6# -- Author Jesús Arroyo
7# -- License GPLv2
10import re
11import threading
12from enum import Enum, unique
13from apio.common.debug_util import is_debug
14from apio.common.apio_console import cout, cunstyle, cwrite, cstyle
15from apio.common.apio_styles import INFO, WARNING, SUCCESS, ERROR
18# -- A table with line coloring rules. If a line matches any regex, it gets
19# -- The style of the first regex it matches. Patterns are case insensitive.
20LINE_COLORING_TABLE = [
21 # -- Info patterns
22 (r"^info:", INFO),
23 # -- Warning patterns
24 (r"^warning:", WARNING),
25 (r"%warning-", WARNING), # Lint/Verilator
26 # -- Error patterns.
27 (r"^error:", ERROR),
28 (r"^%Error:", ERROR),
29 (r" error: ", ERROR),
30 (r"fail: ", ERROR),
31 (r"fatal: ", ERROR),
32 (r"^fatal error:", ERROR),
33 (r"assertion failed", ERROR),
34 # -- Success patterns
35 (r"is up to date", SUCCESS),
36 (r"[$]finish called", SUCCESS),
37 (r"^verify ok$", SUCCESS),
38 (r"^done$", SUCCESS),
39]
41# -- Lines that contain a substring that match any of these regex's are
42# -- ignored. Regexs are case insensitive.
43LINE_IGNORE_LIST = [
44 # -- Per https://github.com/fpgawars/apio/issues/824
45 # -- TODO: Remove when fixed.
46 r"Warning: define gw1n not used in the library",
47 # -- Per https://github.com/YosysHQ/oss-cad-suite-build/issues/194
48 r"Numpy is not available, performance will be degraded",
49 r"Msgspec is not available, performance will be degraded",
50 r"fastcrc is not available, performance will be degraded",
51 # -- For openFPGAloader
52 r"Verifying write [(]May take time[)]",
53]
56@unique
57class PipeId(Enum):
58 """Represent the two output streams from the scons subprocess."""
60 STDOUT = 1
61 STDERR = 2
64@unique
65class RangeEvents(Enum):
66 """An stdout/err line can trigger one of these events, when detecting a
67 range of lines."""
69 START_BEFORE = 1 # Range starts before the current line.
70 START_AFTER = 2 # Range starts after the current line.
71 END_BEFORE = 3 # Range ends before the current line.
72 END_AFTER = 4 # Range ends, after the current line.
75class RangeDetector:
76 """Base detector of a range of lines within the sequence of stdout/err
77 lines recieves from the scons subprocess."""
79 def __init__(self):
80 self._in_range = False
82 def update(self, pipe_id: PipeId, line: str) -> bool:
83 """Updates the range detector with the next stdout/err line.
84 return True iff detector classified this line to be within a range."""
86 prev_state = self._in_range
87 event = self.classify_line(pipe_id, line)
89 if event == RangeEvents.START_BEFORE: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 self._in_range = True
91 return self._in_range
93 if event == RangeEvents.START_AFTER:
94 self._in_range = True
95 return prev_state
97 if event == RangeEvents.END_BEFORE: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 self._in_range = False
99 return self._in_range
101 if event == RangeEvents.END_AFTER:
102 self._in_range = False
103 return prev_state
105 assert event is None, event
106 return self._in_range
108 def classify_line(
109 self, pipe_id: PipeId, line: str
110 ) -> RangeEvents | None: # pragma: no cover
111 """Tests if the next stdout/err line affects the range begin/end.
112 Subclasses should implement this with the necessary logic for the
113 range that is being detected.
114 Returns the event of None if no event."""
115 raise NotImplementedError("Should be implemented by a subclass")
118class PnrRangeDetector(RangeDetector):
119 """Implements a RangeDetector for the nextpnr command verbose
120 log lines."""
122 def classify_line(self, pipe_id: PipeId, line: str) -> RangeEvents | None:
123 # -- Break line into words.
124 tokens = line.split()
126 # -- Range start: A nextpnr command on stdout without
127 # -- the -q (quiet) flag.
128 # --
129 # -- IMPORTANT: Each of the supported architecture has a different
130 # -- nextpnr command, including 'nextpnr', 'nextpnr-ecp5', and
131 # -- 'nextpnr-himbaechel'.
132 if (
133 pipe_id == PipeId.STDOUT
134 and line.startswith("nextpnr")
135 and "-q" not in tokens
136 ):
137 return RangeEvents.START_AFTER
139 # Range end: The end message of nextpnr.
140 if pipe_id == PipeId.STDERR and "Program finished normally." in line:
141 return RangeEvents.END_AFTER
143 return None
146class SconsFilter:
147 """Implements the filtering and printing of the stdout/err streams of the
148 scons subprocess. Accepts a line one at a time, detects lines ranges of
149 interest, mutates and colors the lines where applicable, and print to
150 stdout."""
152 def __init__(self, colors_enabled: bool):
153 self.colors_enabled = colors_enabled
154 self._pnr_detector = PnrRangeDetector()
156 # self._iverilog_detector = IVerilogRangeDetector()
157 # self._iceprog_detector = IceProgRangeDetector()
159 # -- We cache the values to avoid reevaluating sys env.
160 self._is_debug = is_debug(1)
161 self._is_verbose_debug = is_debug(5)
163 # -- Accumulates string pieces until we write and flush them. This
164 # -- mechanism is used to display progress bar correctly, Writing the
165 # -- erasure string only when a new value is available.
166 self._output_bfr: str = ""
168 # -- The stdout and stderr are called from independent threads, so we
169 # -- protect the handling method with this lock.
170 # --
171 # -- We don't protect the third thread which is main(). We hope that
172 # -- it doesn't any print console output while these two threads are
173 # -- active, otherwise it can mingle the output.
174 self._thread_lock = threading.Lock()
176 def on_stdout_line(self, line: str, terminator: str) -> None:
177 """Stdout pipe calls this on each line. Called from the stdout thread
178 in AsyncPipe."""
179 with self._thread_lock:
180 self.on_line(PipeId.STDOUT, line, terminator)
182 def on_stderr_line(self, line: str, terminator: str) -> None:
183 """Stderr pipe calls this on each line. Called from the stderr thread
184 in AsyncPipe."""
185 with self._thread_lock:
186 self.on_line(PipeId.STDERR, line, terminator)
188 @staticmethod
189 def _assign_line_color(
190 line: str,
191 patterns: list[tuple[str, str]],
192 default_color: str | None = None,
193 ) -> str | None:
194 """Assigns a color for a given line using a list of (regex, color)
195 pairs. Returns the color of the first matching regex (case
196 insensitive), or default_color if none match.
197 """
198 for regex, color in patterns:
199 if re.search(regex, line, re.IGNORECASE):
200 return color
201 return default_color
203 def _output_line(
204 self, line: str, style: str | None, terminator: str
205 ) -> None:
206 """Output a line. If a style is given, force that style, otherwise,
207 pass on any color information it may have. The implementation takes
208 into consideration progress bars such as when uploading with the
209 iceprog programmer. These progress bar require certain timing between
210 the chars to have sufficient time to display the text before erasing
211 it."""
213 # -- Apply style if needed.
214 if style:
215 line_part = cstyle(cunstyle(line), style=style)
216 else:
217 line_part = line
219 # -- Get line conditions.
220 is_white = len(line.strip()) == 0
221 is_cr = terminator == "\r"
223 if not is_cr: 223 ↛ 228line 223 didn't jump to line 228 because the condition on line 223 was always true
224 # -- Terminator is EOF or \n. We flush everything.
225 self._output_bfr += line_part + terminator
226 leftover = ""
227 flush = True
228 elif is_white:
229 # -- Terminator is \r and line is white space (progress bar
230 # -- eraser). We queue and and wait for the updated text.
231 self._output_bfr += line_part + terminator
232 leftover = ""
233 flush = False
234 else:
235 # -- Terminator is \r and line has actual text, we flush it out
236 # -- but save queue the \r because on windows 10 cmd it clears the
237 # -- line(?)
238 self._output_bfr += line_part
239 leftover = terminator
240 flush = True
242 if flush: 242 ↛ 248line 242 didn't jump to line 248 because the condition on line 242 was always true
243 # -- Flush the buffer and queue the optional leftover terminator.
244 cwrite(self._output_bfr)
245 self._output_bfr = leftover
246 else:
247 # -- We just queued. Should have no leftover here.
248 assert not leftover
250 def _ignore_line(self, line: str) -> None:
251 """Handle an ignored line. It's dumped if in debug mode."""
252 if self._is_debug: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 cout(f"IGNORED: {line}")
255 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None:
256 """A shared handler for stdout/err lines from the scons sub process.
257 The handler writes both stdout and stderr lines to stdout, possibly
258 with modifications such as text deletion, coloring, and cursor
259 directives.
261 For the possible values of terminator, see AsyncPipe.__init__().
263 NOTE: Ideally, the program specific patterns such as for Fumo and
264 Iceprog should should be condition by a range detector for lines that
265 came from that program. That is to minimize the risk of matching lines
266 from other programs. See the PNR detector for an example.
267 """
269 if self._is_verbose_debug: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 cout(
271 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]",
272 style=INFO,
273 )
275 # -- Update the range detectors.
276 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line)
278 # -- If the line match any of the ignore patterns ignore it.
279 for regex in LINE_IGNORE_LIST:
280 if re.search(regex, line, re.IGNORECASE):
281 self._ignore_line(line)
282 return
284 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where
285 # -- each line starts with "Info: "
286 if ( 286 ↛ 291line 286 didn't jump to line 291 because the condition on line 286 was never true
287 pipe_id == PipeId.STDERR
288 and in_pnr_verbose_range
289 and line.startswith("Info: ")
290 ):
291 line = line[6:]
293 # -- Output the line in the appropriate style.
294 line_color = self._assign_line_color(line, LINE_COLORING_TABLE)
295 self._output_line(line, line_color, terminator)