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