Coverage for apio / managers / scons_filter.py: 82%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-10 03:35 +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:
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, patterns: List[Tuple[str, str]], default_color: str = None
190 ) -> Optional[str]:
191 """Assigns a color for a given line using a list of (regex, color)
192 pairs. Returns the color of the first matching regex (case
193 insensitive), or default_color if none match.
194 """
195 for regex, color in patterns:
196 if re.search(regex, line, re.IGNORECASE):
197 return color
198 return default_color
200 def _output_line(
201 self, line: str, style: Optional[str], terminator: str
202 ) -> None:
203 """Output a line. If a style is given, force that style, otherwise,
204 pass on any color information it may have. The implementation takes
205 into consideration progress bars such as when uploading with the
206 iceprog programmer. These progress bar require certain timing between
207 the chars to have sufficient time to display the text before erasing
208 it."""
210 # -- Apply style if needed.
211 if style:
212 line_part = cstyle(cunstyle(line), style=style)
213 else:
214 line_part = line
216 # -- Get line conditions.
217 is_white = len(line.strip()) == 0
218 is_cr = terminator == "\r"
220 if not is_cr: 220 ↛ 225line 220 didn't jump to line 225 because the condition on line 220 was always true
221 # -- Terminator is EOF or \n. We flush everything.
222 self._output_bfr += line_part + terminator
223 leftover = ""
224 flush = True
225 elif is_white:
226 # -- Terminator is \r and line is white space (progress bar
227 # -- eraser). We queue and and wait for the updated text.
228 self._output_bfr += line_part + terminator
229 leftover = ""
230 flush = False
231 else:
232 # -- Terminator is \r and line has actual text, we flush it out
233 # -- but save queue the \r because on windows 10 cmd it clears the
234 # -- line(?)
235 self._output_bfr += line_part
236 leftover = terminator
237 flush = True
239 if flush: 239 ↛ 245line 239 didn't jump to line 245 because the condition on line 239 was always true
240 # -- Flush the buffer and queue the optional leftover terminator.
241 cwrite(self._output_bfr)
242 self._output_bfr = leftover
243 else:
244 # -- We just queued. Should have no leftover here.
245 assert not leftover
247 def _ignore_line(self, line: str) -> None:
248 """Handle an ignored line. It's dumped if in debug mode."""
249 if self._is_debug: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 cout(f"IGNORED: {line}")
252 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None:
253 """A shared handler for stdout/err lines from the scons sub process.
254 The handler writes both stdout and stderr lines to stdout, possibly
255 with modifications such as text deletion, coloring, and cursor
256 directives.
258 For the possible values of terminator, see AsyncPipe.__init__().
260 NOTE: Ideally, the program specific patterns such as for Fumo and
261 Iceprog should should be condition by a range detector for lines that
262 came from that program. That is to minimize the risk of matching lines
263 from other programs. See the PNR detector for an example.
264 """
266 if self._is_verbose_debug: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 cout(
268 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]",
269 style=INFO,
270 )
272 # -- Update the range detectors.
273 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line)
275 # -- If the line match any of the ignore patterns ignore it.
276 for regex in LINE_IGNORE_LIST:
277 if re.search(regex, line, re.IGNORECASE):
278 self._ignore_line(line)
279 return
281 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where
282 # -- each line starts with "Info: "
283 if ( 283 ↛ 288line 283 didn't jump to line 288 because the condition on line 283 was never true
284 pipe_id == PipeId.STDERR
285 and in_pnr_verbose_range
286 and line.startswith("Info: ")
287 ):
288 line = line[6:]
290 # -- Output the line in the appropriate style.
291 line_color = self._assign_line_color(line, LINE_COLORING_TABLE)
292 self._output_line(line, line_color, terminator)