Coverage for apio / managers / scons_filter.py: 82%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-25 02:31 +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]
55class PipeId(Enum):
56 """Represent the two output streams from the scons subprocess."""
58 STDOUT = 1
59 STDERR = 2
62class RangeEvents(Enum):
63 """An stdout/err line can trigger one of these events, when detecting a
64 range of lines."""
66 START_BEFORE = 1 # Range starts before the current line.
67 START_AFTER = 2 # Range starts after the current line.
68 END_BEFORE = 3 # Range ends before the current line.
69 END_AFTER = 4 # Range ends, after the current line.
72class RangeDetector:
73 """Base detector of a range of lines within the sequence of stdout/err
74 lines recieves from the scons subprocess."""
76 def __init__(self):
77 self._in_range = False
79 def update(self, pipe_id: PipeId, line: str) -> bool:
80 """Updates the range detector with the next stdout/err line.
81 return True iff detector classified this line to be within a range."""
83 prev_state = self._in_range
84 event = self.classify_line(pipe_id, line)
86 if event == RangeEvents.START_BEFORE: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 self._in_range = True
88 return self._in_range
90 if event == RangeEvents.START_AFTER:
91 self._in_range = True
92 return prev_state
94 if event == RangeEvents.END_BEFORE: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true
95 self._in_range = False
96 return self._in_range
98 if event == RangeEvents.END_AFTER:
99 self._in_range = False
100 return prev_state
102 assert event is None, event
103 return self._in_range
105 def classify_line(
106 self, pipe_id: PipeId, line: str
107 ) -> Optional[RangeEvents]: # pragma: no cover
108 """Tests if the next stdout/err line affects the range begin/end.
109 Subclasses should implement this with the necessary logic for the
110 range that is being detected.
111 Returns the event of None if no event."""
112 raise NotImplementedError("Should be implemented by a subclass")
115class PnrRangeDetector(RangeDetector):
116 """Implements a RangeDetector for the nextpnr command verbose
117 log lines."""
119 def classify_line(self, pipe_id: PipeId, line: str) -> RangeEvents:
120 # -- Break line into words.
121 tokens = line.split()
123 # -- Range start: A nextpnr command on stdout without
124 # -- the -q (quiet) flag.
125 # --
126 # -- IMPORTANT: Each of the supported architecture has a different
127 # -- nextpnr command, including 'nextpnr', 'nextpnr-ecp5', and
128 # -- 'nextpnr-himbaechel'.
129 if (
130 pipe_id == PipeId.STDOUT
131 and line.startswith("nextpnr")
132 and "-q" not in tokens
133 ):
134 return RangeEvents.START_AFTER
136 # Range end: The end message of nextnpr.
137 if pipe_id == PipeId.STDERR and "Program finished normally." in line:
138 return RangeEvents.END_AFTER
140 return None
143class SconsFilter:
144 """Implements the filtering and printing of the stdout/err streams of the
145 scons subprocess. Accepts a line one at a time, detects lines ranges of
146 interest, mutates and colors the lines where applicable, and print to
147 stdout."""
149 def __init__(self, colors_enabled: bool):
150 self.colors_enabled = colors_enabled
151 self._pnr_detector = PnrRangeDetector()
153 # self._iverilog_detector = IVerilogRangeDetector()
154 # self._iceprog_detector = IceProgRangeDetector()
156 # -- We cache the values to avoid reevaluating sys env.
157 self._is_debug = util.is_debug(1)
158 self._is_verbose_debug = util.is_debug(5)
160 # -- Accumulates string pieces until we write and flush them. This
161 # -- mechanism is used to display progress bar correctly, Writing the
162 # -- erasure string only when a new value is available.
163 self._output_bfr: str = ""
165 # -- The stdout and stderr are called from independent threads, so we
166 # -- protect the handling method with this lock.
167 # --
168 # -- We don't protect the third thread which is main(). We hope that
169 # -- it doesn't any print console output while these two threads are
170 # -- active, otherwise it can mingle the output.
171 self._thread_lock = threading.Lock()
173 def on_stdout_line(self, line: str, terminator: str) -> None:
174 """Stdout pipe calls this on each line. Called from the stdout thread
175 in AsyncPipe."""
176 with self._thread_lock:
177 self.on_line(PipeId.STDOUT, line, terminator)
179 def on_stderr_line(self, line: str, terminator: str) -> None:
180 """Stderr pipe calls this on each line. Called from the stderr thread
181 in AsyncPipe."""
182 with self._thread_lock:
183 self.on_line(PipeId.STDERR, line, terminator)
185 @staticmethod
186 def _assign_line_color(
187 line: str, patterns: List[Tuple[str, str]], default_color: str = None
188 ) -> Optional[str]:
189 """Assigns a color for a given line using a list of (regex, color)
190 pairs. Returns the color of the first matching regex (case
191 insensitive), or default_color if none match.
192 """
193 for regex, color in patterns:
194 if re.search(regex, line, re.IGNORECASE):
195 return color
196 return default_color
198 def _output_line(
199 self, line: str, style: Optional[str], terminator: str
200 ) -> None:
201 """Output a line. If a style is given, force that style, otherwise,
202 pass on any color information it may have. The implementation takes
203 into consideration progress bars such as when uploading with the
204 iceprog programmer. These progress bar require certain timing between
205 the chars to have sufficient time to display the text before erasing
206 it."""
208 # -- Apply style if needed.
209 if style:
210 line_part = cstyle(cunstyle(line), style=style)
211 else:
212 line_part = line
214 # -- Get line conditions.
215 is_white = len(line.strip()) == 0
216 is_cr = terminator == "\r"
218 if not is_cr: 218 ↛ 223line 218 didn't jump to line 223 because the condition on line 218 was always true
219 # -- Terminator is EOF or \n. We flush everything.
220 self._output_bfr += line_part + terminator
221 leftover = ""
222 flush = True
223 elif is_white:
224 # -- Terminator is \r and line is white space (progress bar
225 # -- eraser). We queue and and wait for the updated text.
226 self._output_bfr += line_part + terminator
227 leftover = ""
228 flush = False
229 else:
230 # -- Terminator is \r and line has actual text, we flush it out
231 # -- but save queue the \r because on windows 10 cmd it clears the
232 # -- line(?)
233 self._output_bfr += line_part
234 leftover = terminator
235 flush = True
237 if flush: 237 ↛ 243line 237 didn't jump to line 243 because the condition on line 237 was always true
238 # -- Flush the buffer and queue the optional leftover terminator.
239 cwrite(self._output_bfr)
240 self._output_bfr = leftover
241 else:
242 # -- We just queued. Should have no leftover here.
243 assert not leftover
245 def _ignore_line(self, line: str) -> None:
246 """Handle an ignored line. It's dumped if in debug mode."""
247 if self._is_debug: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 cout(f"IGNORED: {line}")
250 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None:
251 """A shared handler for stdout/err lines from the scons sub process.
252 The handler writes both stdout and stderr lines to stdout, possibly
253 with modifications such as text deletion, coloring, and cursor
254 directives.
256 For the possible values of terminator, see AsyncPipe.__init__().
258 NOTE: Ideally, the program specific patterns such as for Fumo and
259 Iceprog should should be condition by a range detector for lines that
260 came from that program. That is to minimize the risk of matching lines
261 from other programs. See the PNR detector for an example.
262 """
264 if self._is_verbose_debug: 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 cout(
266 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]",
267 style=INFO,
268 )
270 # -- Update the range detectors.
271 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line)
273 # -- If the line match any of the ignore patterns ignore it.
274 for regex in LINE_IGNORE_LIST:
275 if re.search(regex, line, re.IGNORECASE):
276 self._ignore_line(line)
277 return
279 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where
280 # -- each line starts with "Info: "
281 if ( 281 ↛ 286line 281 didn't jump to line 286 because the condition on line 281 was never true
282 pipe_id == PipeId.STDERR
283 and in_pnr_verbose_range
284 and line.startswith("Info: ")
285 ):
286 line = line[6:]
288 # -- Output the line in the appropriate style.
289 line_color = self._assign_line_color(line, LINE_COLORING_TABLE)
290 self._output_line(line, line_color, terminator)