Coverage for pyEDAA / OutputFilter / Xilinx / Common2.py: 88%
689 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-23 22:13 +0000
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-23 22:13 +0000
1# ==================================================================================================================== #
2# _____ ____ _ _ ___ _ _ _____ _ _ _ #
3# _ __ _ _| ____| _ \ / \ / \ / _ \ _ _| |_ _ __ _ _| |_| ___(_) | |_ ___ _ __ #
4# | '_ \| | | | _| | | | |/ _ \ / _ \ | | | | | | | __| '_ \| | | | __| |_ | | | __/ _ \ '__| #
5# | |_) | |_| | |___| |_| / ___ \ / ___ \ | |_| | |_| | |_| |_) | |_| | |_| _| | | | || __/ | #
6# | .__/ \__, |_____|____/_/ \_\/_/ \_(_)___/ \__,_|\__| .__/ \__,_|\__|_| |_|_|\__\___|_| #
7# |_| |___/ |_| #
8# ==================================================================================================================== #
9# Authors: #
10# Patrick Lehmann #
11# #
12# License: #
13# ==================================================================================================================== #
14# Copyright 2025-2026 Electronic Design Automation Abstraction (EDA²) #
15# #
16# Licensed under the Apache License, Version 2.0 (the "License"); #
17# you may not use this file except in compliance with the License. #
18# You may obtain a copy of the License at #
19# #
20# http://www.apache.org/licenses/LICENSE-2.0 #
21# #
22# Unless required by applicable law or agreed to in writing, software #
23# distributed under the License is distributed on an "AS IS" BASIS, #
24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
25# See the License for the specific language governing permissions and #
26# limitations under the License. #
27# #
28# SPDX-License-Identifier: Apache-2.0 #
29# ==================================================================================================================== #
30#
31"""Basic classes for outputs from AMD/Xilinx Vivado."""
32from datetime import datetime
33from re import Pattern, compile as re_compile
34from typing import ClassVar, Optional as Nullable, Generator, List, Dict, Tuple, Type
36from pyTooling.Decorators import export, readonly
37from pyTooling.MetaClasses import ExtendedType
38from pyTooling.Versioning import YearReleaseVersion
39from pyTooling.Warning import WarningCollector, Warning, CriticalWarning
41from pyEDAA.OutputFilter import OutputFilterException
42from pyEDAA.OutputFilter.Xilinx import Line, LineKind, VivadoMessage
43from pyEDAA.OutputFilter.Xilinx import VivadoInfoMessage, VivadoWarningMessage, VivadoCriticalWarningMessage, VivadoErrorMessage
44from pyEDAA.OutputFilter.Xilinx.Exception import ProcessorException
47MAJOR = r"(?P<major>\d+)"
48MAJOR_MINOR = r"(?P<major>\d+)\.(?P<minor>\d+)"
49MAJOR_MINOR_MICRO = r"(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)"
50MAJOR_MINOR_MICRO_NANO = r"(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)\.(?P<nano>\d+)"
53@export
54class UndetectedEnd(CriticalWarning):
55 _line: Line
57 def __init__(self, message: str, line: Line) -> None:
58 super().__init__(message)
60 self._line = line
62 @readonly
63 def Line(self) -> Line:
64 return self._line
67@export
68class UnknownLine(Warning):
69 _line: Line
71 def __init__(self, message: str, line: Line) -> None:
72 super().__init__(message)
74 self._line = line
76 @readonly
77 def Line(self) -> Line:
78 return self._line
81@export
82class UnknownTask(UnknownLine):
83 pass
86@export
87class UnknownSection(UnknownLine):
88 pass
91@export
92class UnknownPhase(UnknownLine):
93 pass
96@export
97class UnknownSubPhase(UnknownLine):
98 pass
101@export
102class VivadoMessagesMixin(metaclass=ExtendedType, mixin=True):
103 _infoMessages: List[VivadoInfoMessage]
104 _warningMessages: List[VivadoWarningMessage]
105 _criticalWarningMessages: List[VivadoCriticalWarningMessage]
106 _errorMessages: List[VivadoErrorMessage]
107 _toolIDs: Dict[int, str]
108 _toolNames: Dict[str, int]
109 _messagesByID: Dict[int, Dict[int, List[VivadoMessage]]]
111 def __init__(self) -> None:
112 self._infoMessages = []
113 self._warningMessages = []
114 self._criticalWarningMessages = []
115 self._errorMessages = []
116 self._toolIDs = {}
117 self._toolNames = {}
118 self._messagesByID = {}
120 @readonly
121 def ToolIDs(self) -> Dict[int, str]:
122 return self._toolIDs
124 @readonly
125 def ToolNames(self) -> Dict[str, int]:
126 return self._toolNames
128 @readonly
129 def MessagesByID(self) -> Dict[int, Dict[int, List[VivadoMessage]]]:
130 return self._messagesByID
132 @readonly
133 def InfoMessages(self) -> List[VivadoInfoMessage]:
134 return self._infoMessages
136 @readonly
137 def WarningMessages(self) -> List[VivadoWarningMessage]:
138 return self._warningMessages
140 @readonly
141 def CriticalWarningMessages(self) -> List[VivadoCriticalWarningMessage]:
142 return self._criticalWarningMessages
144 @readonly
145 def ErrorMessages(self) -> List[VivadoErrorMessage]:
146 return self._errorMessages
148 def _AddMessage(self, message: VivadoMessage) -> None:
149 if isinstance(message, VivadoInfoMessage):
150 self._infoMessages.append(message)
151 elif isinstance(message, VivadoWarningMessage):
152 self._warningMessages.append(message)
153 elif isinstance(message, VivadoCriticalWarningMessage):
154 self._criticalWarningMessages.append(message)
155 elif isinstance(message, VivadoErrorMessage):
156 self._errorMessages.append(message)
158 if message._toolID in self._messagesByID:
159 sub = self._messagesByID[message._toolID]
160 if message._messageKindID in sub:
161 sub[message._messageKindID].append(message)
162 else:
163 sub[message._messageKindID] = [message]
164 else:
165 self._toolIDs[message._toolID] = message._toolName
166 self._toolNames[message._toolName] = message._toolID
167 self._messagesByID[message._toolID] = {message._messageKindID: [message]}
170@export
171class BaseParser(VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
172 def __init__(self) -> None:
173 super().__init__()
176@export
177class Parser(BaseParser):
178 _processor: "Processor"
180 def __init__(self, processor: "Processor") -> None:
181 super().__init__()
183 self._processor = processor
185 @readonly
186 def Processor(self) -> "Processor":
187 return self._processor
190@export
191class Preamble(Parser):
192 _toolVersion: Nullable[YearReleaseVersion]
193 _startDatetime: Nullable[datetime]
195 _VERSION: ClassVar[Pattern] = re_compile(r"""# Vivado v(\d+\.\d(\.\d)?) \(64-bit\)""")
196 _STARTTIME: ClassVar[Pattern] = re_compile(r"""# Start of session at: (\w+ \w+ \d+ \d+:\d+:\d+ \d+)""")
198 def __init__(self, processor: "BaseProcessor") -> None:
199 super().__init__(processor)
201 self._toolVersion = None
202 self._startDatetime = None
204 @readonly
205 def ToolVersion(self) -> YearReleaseVersion:
206 return self._toolVersion
208 @readonly
209 def StartDatetime(self) -> datetime:
210 return self._startDatetime
212 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
213 if line.StartsWith("#----"): 213 ↛ 216line 213 didn't jump to line 216 because the condition on line 213 was always true
214 line._kind = LineKind.SectionDelimiter
215 else:
216 line._kind |= LineKind.ProcessorError
218 line = yield line
220 while True:
221 if (match := self._VERSION.match(line._message)) is not None:
222 self._toolVersion = YearReleaseVersion.Parse(match[1])
223 line._kind = LineKind.Normal
224 elif (match := self._STARTTIME.match(line._message)) is not None:
225 self._startDatetime = datetime.strptime(match[1], "%a %b %d %H:%M:%S %Y")
226 line._kind = LineKind.Normal
227 elif line.StartsWith("#----"):
228 line._kind = LineKind.SectionDelimiter
229 break
230 else:
231 line._kind = LineKind.Verbose
233 line = yield line
235 nextLine = yield line
236 return nextLine
239@export
240class Task(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
241 # _START: ClassVar[str]
242 # _FINISH: ClassVar[str]
243 _TIME: ClassVar[str] = "Time (s):"
245 _command: "Command"
246 _duration: float
248 def __init__(self, command: "Command") -> None:
249 super().__init__()
250 VivadoMessagesMixin.__init__(self)
252 self._command = command
254 @readonly
255 def Command(self) -> "Command":
256 return self._command
258 def _TaskStart(self, line: Line) -> Generator[Line, Line, Line]:
259 if not line.StartsWith(self._START): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 raise ProcessorException(f"{self.__class__.__name__}._TaskStart(): Expected '{self._START}' at line {line._lineNumber}.")
262 line._kind = LineKind.TaskStart
263 nextLine = yield line
264 return nextLine
266 def _TaskFinish(self, line: Line) -> Generator[Line, Line, Line]:
267 if not line.StartsWith(self._FINISH): 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true
268 raise ProcessorException(f"{self.__class__.__name__}._TaskFinish(): Expected '{self._FINISH}' at line {line._lineNumber}.")
270 line._kind = LineKind.TaskEnd
271 line = yield line
272 while self._TIME is not None: 272 ↛ 279line 272 didn't jump to line 279 because the condition on line 272 was always true
273 if line.StartsWith(self._TIME):
274 line._kind = LineKind.TaskTime
275 break
277 line = yield line
279 line = yield line
280 return line
282 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
283 line = yield from self._TaskStart(line)
285 while True:
286 if line._kind is LineKind.Empty:
287 line = yield line
288 continue
289 elif self._FINISH is not None and line.StartsWith("Ending"):
290 break
291 elif isinstance(line, VivadoMessage):
292 self._AddMessage(line)
293 elif line.StartsWith(self._TIME):
294 line._kind = LineKind.TaskTime
295 nextLine = yield line
296 return nextLine
298 line = yield line
300 nextLine = yield from self._TaskFinish(line)
301 return nextLine
303 def __str__(self) -> str:
304 return f"{self.__class__.__name__}: {self._START}"
307@export
308class TaskWithSubTasks(Task):
309 # _START: ClassVar[str]
310 # _FINISH: ClassVar[str]
311 # _TIME: ClassVar[str] = "Time (s):"
313 _PARSERS: ClassVar[Dict[YearReleaseVersion,Tuple[Type["SubTask"], ...]]] = dict()
315 _subtasks: Dict[Type["SubTask"], "SubTask"]
317 def __init__(self, command: "Command") -> None:
318 super().__init__(command)
320 self._subtasks = {p: p(self) for p in self._PARSERS}
322 @readonly
323 def SubTasks(self) -> Dict[Type["SubTask"], "SubTask"]:
324 return self._subtasks
326 def __getitem__(self, key: Type["SubTask"]) -> "SubTask":
327 return self._subtasks[key]
329 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
330 line = yield from self._TaskStart(line)
332 activeParsers: List[Phase] = list(self._subtasks.values())
334 while True:
335 while True:
336 if line._kind is LineKind.Empty:
337 line = yield line
338 continue
339 elif isinstance(line, VivadoMessage):
340 self._AddMessage(line)
341 elif line.StartsWith("Starting "):
342 for parser in activeParsers: # type: SubTask 342 ↛ 347line 342 didn't jump to line 347 because the loop on line 342 didn't complete
343 if line.StartsWith(parser._START): 343 ↛ 342line 343 didn't jump to line 342 because the condition on line 343 was always true
344 line = yield next(subtask := parser.Generator(line))
345 break
346 else:
347 raise Exception(f"Unknown subtask: {line!r}")
348 break
349 elif line.StartsWith("Ending"):
350 nextLine = yield from self._TaskFinish(line)
351 return nextLine
352 elif line.StartsWith(self._TIME): 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 line._kind = LineKind.TaskTime
354 nextLine = yield line
355 return nextLine
357 line = yield line
359 while subtask is not None: 359 ↛ 334line 359 didn't jump to line 334 because the condition on line 359 was always true
360 # print(line)
361 # if line.StartsWith("Ending"):
362 # line = yield task.send(line)
363 # break
365 if isinstance(line, VivadoMessage):
366 self._AddMessage(line)
368 try:
369 line = yield subtask.send(line)
370 except StopIteration as ex:
371 activeParsers.remove(parser)
372 line = ex.value
373 break
375@export
376class SubTask(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
377 # _START: ClassVar[str]
378 # _FINISH: ClassVar[str]
379 _TIME: ClassVar[str] = "Time (s):"
381 _task: TaskWithSubTasks
382 _duration: float
384 def __init__(self, task: TaskWithSubTasks) -> None:
385 super().__init__()
386 VivadoMessagesMixin.__init__(self)
388 self._task = task
390 @readonly
391 def Task(self) -> TaskWithSubTasks:
392 return self._task
394 def _TaskStart(self, line: Line) -> Generator[Line, Line, Line]:
395 if not line.StartsWith(self._START): 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true
396 raise ProcessorException(f"{self.__class__.__name__}._TaskStart(): Expected '{self._START}' at line {line._lineNumber}.")
398 line._kind = LineKind.TaskStart
399 nextLine = yield line
400 return nextLine
402 def _TaskFinish(self, line: Line) -> Generator[Line, Line, Line]:
403 if not line.StartsWith(self._FINISH): 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 raise ProcessorException(f"{self.__class__.__name__}._TaskFinish(): Expected '{self._FINISH}' at line {line._lineNumber}.")
406 line._kind = LineKind.TaskEnd
407 line = yield line
408 while self._TIME is not None: 408 ↛ 415line 408 didn't jump to line 415 because the condition on line 408 was always true
409 if line.StartsWith(self._TIME):
410 line._kind = LineKind.TaskTime
411 break
413 line = yield line
415 line = yield line
416 return line
418 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
419 line = yield from self._TaskStart(line)
421 while True:
422 if line._kind is LineKind.Empty:
423 line = yield line
424 continue
425 elif self._FINISH is not None and line.StartsWith("Ending"):
426 break
427 elif isinstance(line, VivadoMessage):
428 self._AddMessage(line)
429 elif line.StartsWith(self._TIME):
430 line._kind = LineKind.TaskTime
431 nextLine = yield line
432 return nextLine
434 line = yield line
436 nextLine = yield from self._TaskFinish(line)
437 return nextLine
439 def __str__(self) -> str:
440 return self._NAME
443@export
444class TaskWithPhases(Task):
445 # _START: ClassVar[str]
446 # _FINISH: ClassVar[str]
447 # _TIME: ClassVar[str] = "Time (s):"
449 _PARSERS: ClassVar[Dict[YearReleaseVersion,Tuple[Type["Phase"], ...]]] = tuple()
451 _phases: Dict[Type["Phase"], "Phase"]
453 def __init__(self, command: "Command") -> None:
454 super().__init__(command)
456 self._phases = {p: p(self) for p in self._PARSERS}
458 @readonly
459 def Phases(self) -> Dict[Type["Phase"], "Phase"]:
460 return self._phases
462 def __getitem__(self, key: Type["Phase"]) -> "Phase":
463 return self._phases[key]
465 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
466 line = yield from self._TaskStart(line)
468 activeParsers: List[Phase] = list(self._phases.values())
470 while True:
471 while True:
472 if line._kind is LineKind.Empty:
473 line = yield line
474 continue
475 elif isinstance(line, VivadoMessage):
476 self._AddMessage(line)
477 elif line.StartsWith("Phase "):
478 for parser in activeParsers: # type: Phase 478 ↛ 483line 478 didn't jump to line 483 because the loop on line 478 didn't complete
479 if (match := parser._START.match(line._message)) is not None:
480 line = yield next(phase := parser.Generator(line))
481 break
482 else:
483 raise Exception(f"Unknown phase: {line!r}")
484 break
485 elif line.StartsWith("Ending"):
486 nextLine = yield from self._TaskFinish(line)
487 return nextLine
488 elif line.StartsWith(self._TIME):
489 line._kind = LineKind.TaskTime
490 nextLine = yield line
491 return nextLine
493 line = yield line
495 while phase is not None: 495 ↛ 470line 495 didn't jump to line 470 because the condition on line 495 was always true
496 if isinstance(line, VivadoMessage):
497 self._AddMessage(line)
499 isFinish = line.StartsWith("Ending")
501 try:
502 line = yield phase.send(line)
503 if isFinish:
504 previousLine = line._previousLine
505 WarningCollector.Raise(UndetectedEnd(
506 f"Didn't detect finish: '{previousLine!r}'",
507 previousLine
508 ))
509 break
510 except StopIteration as ex:
511 activeParsers.remove(parser)
512 line = ex.value
513 break
516@export
517class Phase(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
518 # _START: ClassVar[str]
519 # _FINISH: ClassVar[str]
520 _TIME: ClassVar[str] = "Time (s):"
521 _FINAL: ClassVar[Nullable[str]] = None
523 _task: TaskWithPhases
524 _phaseIndex: int
525 _duration: float
527 def __init__(self, task: TaskWithPhases) -> None:
528 super().__init__()
529 VivadoMessagesMixin.__init__(self)
531 self._task = task
532 self._phaseIndex = None
534 @readonly
535 def Task(self) -> TaskWithPhases:
536 return self._task
538 def _PhaseStart(self, line: Line) -> Generator[Line, Line, Line]:
539 if (match := self._START.match(line._message)) is None: 539 ↛ 540line 539 didn't jump to line 540 because the condition on line 539 was never true
540 raise ProcessorException(f"{self.__class__.__name__}._PhaseStart(): Expected '{self._START}' at line {line._lineNumber}.")
542 self._phaseIndex = int(match["major"])
544 line._kind = LineKind.PhaseStart
545 nextLine = yield line
546 return nextLine
548 def _PhaseFinish(self, line: Line) -> Generator[Line, Line, None]:
549 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)
550 if not line.StartsWith(FINISH): 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 raise ProcessorException(f"{self.__class__.__name__}._PhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")
553 line._kind = LineKind.PhaseEnd
554 line = yield line
556 if self._TIME is not None: 556 ↛ 566line 556 didn't jump to line 566 because the condition on line 556 was always true
557 while self._TIME is not None: 557 ↛ 564line 557 didn't jump to line 564 because the condition on line 557 was always true
558 if line.StartsWith(self._TIME):
559 line._kind = LineKind.PhaseTime
560 break
562 line = yield line
564 line = yield line
566 if self._FINAL is not None: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 while self._FINAL is not None:
568 if line.StartsWith(self._FINAL):
569 line._kind = LineKind.PhaseFinal
570 break
572 line = yield line
574 line = yield line
576 return line
578 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
579 line = yield from self._PhaseStart(line)
581 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)
583 while True:
584 if line._kind is LineKind.Empty:
585 line = yield line
586 continue
587 elif isinstance(line, VivadoMessage):
588 self._AddMessage(line)
589 elif line.StartsWith(FINISH):
590 break
592 line = yield line
594 nextLine = yield from self._PhaseFinish(line)
595 return nextLine
597 def __str__(self) -> str:
598 return f"{self.__class__.__name__}: {self._START.pattern}"
601@export
602class PhaseWithChildren(Phase):
603 _SUBPHASE_PREFIX: ClassVar[str] = "Phase {phaseIndex}."
605 _subPhases: Dict[Type["SubPhase"], "SubPhase"]
607 def __init__(self, task: TaskWithPhases) -> None:
608 super().__init__(task)
610 self._subPhases = {p: p(self) for p in self._PARSERS}
612 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
613 line = yield from self._PhaseStart(line)
615 activeParsers: List[SubPhase] = list(self._subPhases.values())
617 SUBPHASE_PREFIX = self._SUBPHASE_PREFIX.format(phaseIndex=self._phaseIndex)
618 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)
620 while True:
621 while True:
622 if line._kind is LineKind.Empty:
623 line = yield line
624 continue
625 elif isinstance(line, VivadoMessage):
626 self._AddMessage(line)
627 elif line.StartsWith(SUBPHASE_PREFIX):
628 for parser in activeParsers: # type: Section 628 ↛ 633line 628 didn't jump to line 633 because the loop on line 628 didn't complete
629 if (match := parser._START.match(line._message)) is not None:
630 line = yield next(phase := parser.Generator(line))
631 break
632 else:
633 WarningCollector.Raise(UnknownSubPhase(f"Unknown subphase: '{line!r}'", line))
634 ex = Exception(f"How to recover from here? Unknown subphase: '{line!r}'")
635 ex.add_note(f"Current phase: start pattern='{self}'")
636 ex.add_note(f"Current task: start pattern='{self._task}'")
637 ex.add_note(f"Current command: {self._task._command}")
638 raise ex
639 break
640 elif line.StartsWith(FINISH):
641 nextLine = yield from self._PhaseFinish(line)
642 return nextLine
644 line = yield line
646 while phase is not None: 646 ↛ 620line 646 didn't jump to line 620 because the condition on line 646 was always true
647 if isinstance(line, VivadoMessage):
648 self._AddMessage(line)
650 isFinish = False # line.StartsWith(SUBPHASE_PREFIX)
652 try:
653 line = yield phase.send(line)
654 if isFinish: 654 ↛ 655line 654 didn't jump to line 655 because the condition on line 654 was never true
655 previousLine = line._previousLine
656 WarningCollector.Raise(UndetectedEnd(
657 f"Didn't detect finish: '{previousLine!r}'",
658 previousLine
659 ))
660 break
661 except StopIteration as ex:
662 activeParsers.remove(parser)
663 line = ex.value
664 break
667@export
668class SubPhase(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
669 # _START: ClassVar[str]
670 # _FINISH: ClassVar[str]
672 _phase: Phase
673 _phaseIndex: int
674 _subPhaseIndex: int
675 _duration: float
677 def __init__(self, phase: Phase) -> None:
678 super().__init__()
679 VivadoMessagesMixin.__init__(self)
681 self._phase = phase
682 self._phaseIndex = None
683 self._subPhaseIndex = None
685 def _SubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:
686 if (match := self._START.match(line._message)) is None: 686 ↛ 687line 686 didn't jump to line 687 because the condition on line 686 was never true
687 raise ProcessorException(f"{self.__class__.__name__}._SubPhaseStart(): Expected '{self._START}' at line {line._lineNumber}.")
689 self._phaseIndex = int(match["major"])
690 self._subPhaseIndex = int(match["minor"])
692 line._kind = LineKind.SubPhaseStart
693 nextLine = yield line
694 return nextLine
696 def _SubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:
697 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)
699 if line.StartsWith(FINISH) is None: 699 ↛ 700line 699 didn't jump to line 700 because the condition on line 699 was never true
700 raise ProcessorException(f"{self.__class__.__name__}._SubPhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")
702 if self._TIME is None: 702 ↛ 703line 702 didn't jump to line 703 because the condition on line 702 was never true
703 line._kind = LineKind.SubPhaseTime
704 else:
705 line._kind = LineKind.SubPhaseEnd
707 line = yield line
708 while self._TIME is not None: 708 ↛ 715line 708 didn't jump to line 715 because the condition on line 708 was always true
709 if line.StartsWith(self._TIME):
710 line._kind = LineKind.SubPhaseTime
711 break
713 line = yield line
715 nextLine = yield line
716 return nextLine
718 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
719 line = yield from self._SubPhaseStart(line)
721 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)
723 while True:
724 if line._kind is LineKind.Empty:
725 line = yield line
726 continue
727 elif line.StartsWith(FINISH):
728 break
729 elif isinstance(line, VivadoMessage):
730 self._AddMessage(line)
732 line = yield line
734 nextLine = yield from self._SubPhaseFinish(line)
735 return nextLine
737 def __str__(self) -> str:
738 return f"{self.__class__.__name__}: {self._START.pattern}"
741@export
742class SubPhaseWithChildren(SubPhase):
743 _subSubPhases: Dict[Type["SubSubPhase"], "SubSubPhase"]
745 def __init__(self, phase: Phase) -> None:
746 super().__init__(phase)
748 self._subSubPhases = {p: p(self) for p in self._PARSERS}
750 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
751 line = yield from self._SubPhaseStart(line)
753 activeParsers: List["SubSubPhase"] = list(self._subSubPhases.values())
755 START_PREFIX = f"Phase {self._phaseIndex}.{self._subPhaseIndex}."
756 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)
758 while True:
759 while True:
760 if line._kind is LineKind.Empty:
761 line = yield line
762 continue
763 elif isinstance(line, VivadoMessage):
764 self._AddMessage(line)
765 elif line.StartsWith(START_PREFIX):
766 for parser in activeParsers: # type: SubSubPhase 766 ↛ 771line 766 didn't jump to line 771 because the loop on line 766 didn't complete
767 if (match := parser._START.match(line._message)) is not None:
768 line = yield next(phase := parser.Generator(line))
769 break
770 else:
771 WarningCollector.Raise(UnknownSubPhase(f"Unknown subsubphase: '{line!r}'", line))
772 ex = Exception(f"How to recover from here? Unknown subsubphase: '{line!r}'")
773 ex.add_note(f"Current task: start pattern='{self._task}'")
774 ex.add_note(f"Current cmd: {self._task._command}")
775 raise ex
776 break
777 elif line.StartsWith(FINISH): 777 ↛ 781line 777 didn't jump to line 781 because the condition on line 777 was always true
778 nextLine = yield from self._SubPhaseFinish(line)
779 return nextLine
781 line = yield line
783 while phase is not None: 783 ↛ 758line 783 didn't jump to line 758 because the condition on line 783 was always true
784 # if line.StartsWith("Ending"):
785 # line = yield task.send(line)
786 # break
788 if isinstance(line, VivadoMessage):
789 self._AddMessage(line)
791 try:
792 line = yield phase.send(line)
793 except StopIteration as ex:
794 activeParsers.remove(parser)
795 line = ex.value
796 break
799@export
800class SubSubPhase(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
801 # _START: ClassVar[str]
802 # _FINISH: ClassVar[str]
804 _subphase: SubPhase
805 _phaseIndex: int
806 _subPhaseIndex: int
807 _subSubPhaseIndex: int
808 _duration: float
810 def __init__(self, subphase: SubPhase) -> None:
811 super().__init__()
812 VivadoMessagesMixin.__init__(self)
814 self._subphase = subphase
815 self._phaseIndex = None
816 self._subPhaseIndex = None
817 self._subSubPhaseIndex = None
819 def _SubSubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:
820 if (match := self._START.match(line._message)) is None: 820 ↛ 821line 820 didn't jump to line 821 because the condition on line 820 was never true
821 raise ProcessorException()
823 self._phaseIndex = int(match["major"])
824 self._subPhaseIndex = int(match["minor"])
825 self._subSubPhaseIndex = int(match["micro"])
827 line._kind = LineKind.SubSubPhaseStart
828 nextLine = yield line
829 return nextLine
831 def _SubSubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:
832 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex)
834 if line.StartsWith(FINISH) is None: 834 ↛ 835line 834 didn't jump to line 835 because the condition on line 834 was never true
835 raise ProcessorException()
837 line._kind = LineKind.SubSubPhaseEnd
838 line = yield line
840 while self._TIME is not None: 840 ↛ 847line 840 didn't jump to line 847 because the condition on line 840 was always true
841 if line.StartsWith(self._TIME):
842 line._kind = LineKind.SubSubPhaseTime
843 break
845 line = yield line
847 nextLine = yield line
848 return nextLine
850 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
851 line = yield from self._SubSubPhaseStart(line)
853 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex)
855 while True:
856 if line._kind is LineKind.Empty:
857 line = yield line
858 continue
859 elif line.StartsWith(FINISH):
860 break
861 elif isinstance(line, VivadoMessage):
862 self._AddMessage(line)
864 line = yield line
866 nextLine = yield from self._SubSubPhaseFinish(line)
867 return nextLine
869 def __str__(self) -> str:
870 return f"{self.__class__.__name__}: {self._START.pattern}"
873@export
874class SubSubPhaseWithChildren(SubSubPhase):
875 _subSubSubPhases: Dict[Type["SubSubSubPhase"], "SubSubSubPhase"]
877 def __init__(self, subphase: SubPhase) -> None:
878 super().__init__(subphase)
880 self._subSubSubPhases = {p: p(self) for p in self._PARSERS}
882 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
883 line = yield from self._SubSubPhaseStart(line)
885 activeParsers: List["SubSubSubPhase"] = list(self._subSubSubPhases.values())
887 START_PREFIX = f"Phase {self._phaseIndex}.{self._subPhaseIndex}.{self._subSubPhaseIndex}."
889 while True:
890 while True:
891 if line._kind is LineKind.Empty:
892 line = yield line
893 continue
894 elif isinstance(line, VivadoMessage): 894 ↛ 895line 894 didn't jump to line 895 because the condition on line 894 was never true
895 self._AddMessage(line)
896 elif line.StartsWith(START_PREFIX):
897 for parser in activeParsers: # type: SubSubSubPhase 897 ↛ 902line 897 didn't jump to line 902 because the loop on line 897 didn't complete
898 if (match := parser._START.match(line._message)) is not None: 898 ↛ 897line 898 didn't jump to line 897 because the condition on line 898 was always true
899 line = yield next(phase := parser.Generator(line))
900 break
901 else:
902 raise Exception(f"Unknown subsubsubphase: {line!r}")
903 break
904 elif line.StartsWith(self._TIME):
905 line._kind = LineKind.SubSubPhaseTime
906 nextLine = yield line
907 return nextLine
909 line = yield line
911 while phase is not None: 911 ↛ 889line 911 didn't jump to line 889 because the condition on line 911 was always true
912 # if line.StartsWith("Ending"):
913 # line = yield task.send(line)
914 # break
916 if isinstance(line, VivadoMessage):
917 self._AddMessage(line)
919 try:
920 line = yield phase.send(line)
921 except StopIteration as ex:
922 activeParsers.remove(parser)
923 line = ex.value
924 break
927@export
928class SubSubSubPhase(BaseParser, VivadoMessagesMixin, metaclass=ExtendedType, slots=True):
929 # _START: ClassVar[str]
930 # _FINISH: ClassVar[str]
932 _subsubphase: SubSubPhase
933 _phaseIndex: int
934 _subPhaseIndex: int
935 _subSubPhaseIndex: int
936 _subSubSubPhaseIndex: int
937 _duration: float
939 def __init__(self, subsubphase: SubSubPhase) -> None:
940 super().__init__()
941 VivadoMessagesMixin.__init__(self)
943 self._subsubphase = subsubphase
944 self._phaseIndex = None
945 self._subPhaseIndex = None
946 self._subSubPhaseIndex = None
947 self._subSubSubPhaseIndex = None
949 def _SubSubSubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:
950 if (match := self._START.match(line._message)) is None: 950 ↛ 951line 950 didn't jump to line 951 because the condition on line 950 was never true
951 raise ProcessorException()
953 self._phaseIndex = int(match["major"])
954 self._subPhaseIndex = int(match["minor"])
955 self._subSubPhaseIndex = int(match["micro"])
956 self._subSubSubPhaseIndex = int(match["nano"])
958 line._kind = LineKind.SubSubSubPhaseStart
959 nextLine = yield line
960 return nextLine
962 def _SubSubSubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:
963 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex, subSubSubPhaseIndex=self._subSubSubPhaseIndex)
965 if line.StartsWith(FINISH) is None: 965 ↛ 966line 965 didn't jump to line 966 because the condition on line 965 was never true
966 raise ProcessorException()
968 line._kind = LineKind.SubSubSubPhaseEnd
969 line = yield line
971 while self._TIME is not None: 971 ↛ 978line 971 didn't jump to line 978 because the condition on line 971 was always true
972 if line.StartsWith(self._TIME):
973 line._kind = LineKind.SubSubSubPhaseTime
974 break
976 line = yield line
978 nextLine = yield line
979 return nextLine
981 def Generator(self, line: Line) -> Generator[Line, Line, Line]:
982 line = yield from self._SubSubSubPhaseStart(line)
984 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex, subSubSubPhaseIndex=self._subSubSubPhaseIndex)
986 while True:
987 if line._kind is LineKind.Empty:
988 line = yield line
989 continue
990 elif line.StartsWith(FINISH):
991 break
992 elif isinstance(line, VivadoMessage):
993 self._AddMessage(line)
995 line = yield line
997 nextLine = yield from self._SubSubSubPhaseFinish(line)
998 return nextLine
1000 def __str__(self) -> str:
1001 return f"{self.__class__.__name__}: {self._START.pattern}"