Coverage for pyEDAA/OutputFilter/Xilinx/Common2.py: 78%

1# ==================================================================================================================== #

2# _____ ____ _ _ ___ _ _ _____ _ _ _ #

3# _ __ _ _| ____| _ \ / \ / \ / _ \ _ _| |_ _ __ _ _| |_| ___(_) | |_ ___ _ __ #

4# | '_ \| | | | _| | | | |/ _ \ / _ \ | | | | | | | __| '_ \| | | | __| |_ | | | __/ _ \ '__| #

5# | |_) | |_| | |___| |_| / ___ \ / ___ \ | |_| | |_| | |_| |_) | |_| | |_| _| | | | || __/ | #

6# | .__/ \__, |_____|____/_/ \_\/_/ \_(_)___/ \__,_|\__| .__/ \__,_|\__|_| |_|_|\__\___|_| #

7# |_| |___/ |_| #

8# ==================================================================================================================== #

9# Authors: #

10# Patrick Lehmann #

11# #

12# License: #

13# ==================================================================================================================== #

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, Any

36from pyTooling.Common import getFullyQualifiedName

37from pyTooling.Decorators import export, readonly

38from pyTooling.MetaClasses import ExtendedType

39from pyTooling.Versioning import YearReleaseVersion

40from pyTooling.Warning import WarningCollector, Warning, CriticalWarning

42from pyEDAA.OutputFilter.Xilinx import Line, LineKind, VivadoMessage, InfoMessage, WarningMessage, CriticalWarningMessage, ErrorMessage

43from pyEDAA.OutputFilter.Xilinx import VivadoInfoMessage, VivadoWarningMessage, VivadoCriticalWarningMessage, VivadoErrorMessage

44from pyEDAA.OutputFilter.Xilinx.Exception import ProcessorException, NotPresentException

46MAJOR = r"(?P<major>\d+)"

47MAJOR_MINOR = r"(?P<major>\d+)\.(?P<minor>\d+)"

48MAJOR_MINOR_MICRO = r"(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)"

49MAJOR_MINOR_MICRO_NANO = r"(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)\.(?P<nano>\d+)"

52@export

53class UndetectedEnd(CriticalWarning):

54 _line: Line

56 def __init__(self, message: str, line: Line) -> None:

57 super().__init__(message)

59 self._line = line

61 @readonly

62 def Line(self) -> Line:

63 return self._line

66@export

67class UnknownLine(Warning):

68 _line: Line

70 def __init__(self, message: str, line: Line) -> None:

71 super().__init__(message)

73 self._line = line

75 @readonly

76 def Line(self) -> Line:

77 return self._line

80@export

81class UnknownTask(UnknownLine):

82 pass

85@export

86class UnknownSubTask(UnknownLine):

87 pass

90@export

91class UnknownSection(UnknownLine):

92 pass

95@export

96class UnknownPhase(UnknownLine):

97 pass

100@export

101class UnknownSubPhase(UnknownLine):

102 pass

103

104

105@export

106class SubTaskNotPresentException(NotPresentException):

107 pass

108

109

110@export

111class PhaseNotPresentException(NotPresentException):

112 pass

113

114

115@export

116class NestedTaskNotPresentException(NotPresentException):

117 pass

118

119

120@export

121class VivadoMessagesMixin(metaclass=ExtendedType, mixin=True):

122 _infoMessages: List[VivadoInfoMessage]

123 _warningMessages: List[VivadoWarningMessage]

124 _criticalWarningMessages: List[VivadoCriticalWarningMessage]

125 _errorMessages: List[VivadoErrorMessage]

126 _toolIDs: Dict[int, str]

127 _toolNames: Dict[str, int]

128 _messagesByID: Dict[int, Dict[int, List[VivadoMessage]]]

129

130 def __init__(self) -> None:

131 self._infoMessages = []

132 self._warningMessages = []

133 self._criticalWarningMessages = []

134 self._errorMessages = []

135 self._toolIDs = {}

136 self._toolNames = {}

137 self._messagesByID = {}

138

139 @readonly

140 def ToolIDs(self) -> Dict[int, str]:

141 return self._toolIDs

142

143 @readonly

144 def ToolNames(self) -> Dict[str, int]:

145 return self._toolNames

146

147 @readonly

148 def MessagesByID(self) -> Dict[int, Dict[int, List[VivadoMessage]]]:

149 return self._messagesByID

150

151 @readonly

152 def InfoMessages(self) -> List[VivadoInfoMessage]:

153 return self._infoMessages

154

155 @readonly

156 def WarningMessages(self) -> List[VivadoWarningMessage]:

157 return self._warningMessages

158

159 @readonly

160 def CriticalWarningMessages(self) -> List[VivadoCriticalWarningMessage]:

161 return self._criticalWarningMessages

162

163 @readonly

164 def ErrorMessages(self) -> List[VivadoErrorMessage]:

165 return self._errorMessages

166

167 def _AddMessage(self, message: VivadoMessage) -> None:

168 if isinstance(message, InfoMessage):

169 self._infoMessages.append(message)

170 elif isinstance(message, WarningMessage):

171 self._warningMessages.append(message)

172 elif isinstance(message, CriticalWarningMessage):

173 self._criticalWarningMessages.append(message)

174 elif isinstance(message, ErrorMessage): 174 ↛ 177line 174 didn't jump to line 177 because the condition on line 174 was always true

175 self._errorMessages.append(message)

176

177 if message._toolID in self._messagesByID:

178 sub = self._messagesByID[message._toolID]

179 if message._messageKindID in sub:

180 sub[message._messageKindID].append(message)

181 else:

182 sub[message._messageKindID] = [message]

183 else:

184 if message._toolID is not None:

185 self._toolIDs[message._toolID] = message._toolName

186 self._toolNames[message._toolName] = message._toolID

187

188 self._messagesByID[message._toolID] = {message._messageKindID: [message]}

189

190

191@export

192class BaseParser(VivadoMessagesMixin, metaclass=ExtendedType, slots=True):

193 def __init__(self) -> None:

194 super().__init__()

195

196

197@export

198class Parser(BaseParser):

199 _processor: "Processor"

200

201 def __init__(self, processor: "Processor") -> None:

202 super().__init__()

203

204 self._processor = processor

205

206 @readonly

207 def Processor(self) -> "Processor":

208 return self._processor

209

210

211@export

212class Preamble(Parser):

213 """

214 A parser for the preamble emitted by Vivado at session start.

215

216 .. rubric:: Extracted information

217

218 * Vivado tool version. |br|

219 See :data:`ToolVersion`

220 * Session start timestamp (date and time). |br|

221 See :data:`StartDatetime`

222

223 .. rubric:: Example

224

225 .. code-block::

226

227 #-----------------------------------------------------------

228 # Vivado v2025.1 (64-bit)

229 # SW Build 6140274 on Thu May 22 00:12:29 MDT 2025

230 # IP Build 6138677 on Thu May 22 03:10:11 MDT 2025

231 # SharedData Build 6139179 on Tue May 20 17:58:58 MDT 2025

232 # Start of session at: Thu Jun 12 18:39:05 2025

233 # Process ID : 28856

234 # Current directory : C:/Git/.../StopWatch/project/4_WithTiming.runs/impl_1

235 # Command line : vivado.exe -log toplevel.vdi -applog -product Vivado -messageDb vivado.pb -mode batch -source toplevel.tcl -notrace

236 # Log file : C:/Git/.../StopWatch/project/4_WithTiming.runs/impl_1/toplevel.vdi

237 # Journal file : C:/Git/.../StopWatch/project/4_WithTiming.runs/impl_1\vivado.jou

238 # Running On : Paebbels

239 # Platform : Windows Server 2016 or Windows 10

240 # Operating System : 26100

241 # Processor Detail : 11th Gen Intel(R) Core(TM) i9-11950H @ 2.60GHz

242 # CPU Frequency : 2611 MHz

243 # CPU Physical cores : 8

244 # CPU Logical cores : 16

245 # Host memory : 34048 MB

246 # Swap memory : 28991 MB

247 # Total Virtual : 63039 MB

248 # Available Virtual : 29246 MB

249 #-----------------------------------------------------------

250 """

251 _VERSION: ClassVar[Pattern] = re_compile(r"""# Vivado v(\d+\.\d(\.\d)?) \(64-bit\)""")

252 _STARTTIME: ClassVar[Pattern] = re_compile(r"""# Start of session at: (\w+\s+\w+\s+\d+\s+\d+:\d+:\d+\s+\d+)""")

253

254 _toolVersion: Nullable[YearReleaseVersion] #: Used Vivado version.

255 _startDatetime: Nullable[datetime] #: Session start timestamp.

256

257 def __init__(self, processor: "BaseProcessor") -> None:

258 """

259 Initializes a Vivado preamble parser.

260

261 :param processor: Reference to the Vivado log processor.

262 """

263 super().__init__(processor)

264

265 self._toolVersion = None

266 self._startDatetime = None

267

268 @readonly

269 def ToolVersion(self) -> YearReleaseVersion:

270 """

271 Read-only property to access the extracted Vivado tool version.

272

273 :returns: The used Vivado version as reported in the Vivado log messages.

274 """

275 return self._toolVersion

276

277 @readonly

278 def StartDatetime(self) -> datetime:

279 """

280 Read-only property to access the date and time when the Vivado session was started.

281

282 :returns: Datatime when the session was started.

283 :raises ProcessorException: When start timestamp wasn't extracted from preamble.

284 """

285 if self._startDatetime is None: 285 ↛ 286line 285 didn't jump to line 286 because the condition on line 285 was never true

286 raise ProcessorException("No start timestamp extracted from preamble.")

287

288 return self._startDatetime

289

290 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

291 """

292 A generator for processing the Vivado session preamble line-by-line.

293

294 :param line: First line to process.

295 :returns: A generator processing log messages.

296 """

297 if line.StartsWith("#----"): 297 ↛ 300line 297 didn't jump to line 300 because the condition on line 297 was always true

298 line._kind = LineKind.SectionDelimiter

299 else:

300 line._kind |= LineKind.ProcessorError # TODO: throw / return error

301

302 line = yield line

303

304 # a normal preamble has up to 23 lines including both delimiter lines.

305 for _ in range(30): 305 ↛ 320line 305 didn't jump to line 320 because the loop on line 305 didn't complete

306 if (match := self._VERSION.match(line._message)) is not None:

307 self._toolVersion = YearReleaseVersion.Parse(match[1])

308 line._kind = LineKind.Normal

309 elif (match := self._STARTTIME.match(line._message)) is not None:

310 self._startDatetime = datetime.strptime(match[1], "%a %b %d %H:%M:%S %Y")

311 line._kind = LineKind.Normal

312 elif line.StartsWith("#----"):

313 line._kind = LineKind.SectionDelimiter

314 break

315 else:

316 line._kind = LineKind.Verbose

317

318 line = yield line

319 else:

320 line._kind |= LineKind.ProcessorError # TODO: throw / return error

321

322 nextLine = yield line

323 return nextLine

324

325

326@export

327class Postamble(Parser, VivadoMessagesMixin):

328 """

329 A parser for the postamble emitted by Vivado at session end.

330

331 .. rubric:: Extracted information

332

333 * Session exit timestamp (date and time). |br|

334 See :data:`ExitDatetime`

335

336 .. rubric:: Example

337

338 .. code-block::

339

340 INFO: [Common 17-206] Exiting Vivado at Tue Sep 2 08:46:23 2025...

341

342 """

343 _INFO: Tuple[int, int] = (17, 206)

344 _ENDTIME: ClassVar[Pattern] = re_compile(r"""Exiting Vivado at (\w+\s+\w+\s+\d+\s+\d+:\d+:\d+\s+\d+)""")

345

346 _exitDatetime: Nullable[datetime] #: Session exit timestamp.

347

348 def __init__(self, processor: "BaseProcessor") -> None:

349 """

350 Initializes a Vivado postamble parser.

351

352 :param processor: Reference to the Vivado log processor.

353 """

354 super().__init__(processor)

355 VivadoMessagesMixin.__init__(self)

356

357 self._exitDatetime = None

358

359 @readonly

360 def ExitDatetime(self) -> Nullable[datetime]:

361 """

362 Read-only property to access the date and time when the Vivado session was exited.

363

364 :returns: Datatime when the session was exited.

365 :raises ProcessorException: When exit timestamp wasn't extracted from postamble.

366 """

367 if self._exitDatetime is None: 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true

368 raise ProcessorException("No exit timestamp extracted from postamble.")

369

370 return self._exitDatetime

371

372 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

373 """

374 A generator for processing the Vivado session preamble line-by-line.

375

376 :param line: First line to process.

377 :returns: A generator processing log messages.

378 """

379 if isinstance(line, VivadoMessage): 379 ↛ 385line 379 didn't jump to line 385 because the condition on line 379 was always true

380 self._AddMessage(line)

381

382 if not isinstance(line, VivadoInfoMessage): 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true

383 raise ProcessorException(f"{self.__class__.__name__}.Generator(): Expected '{self._ENDTIME}' at line {line._lineNumber}.")

384

385 if (match := self._ENDTIME.match(line._message)) is not None:

386 self._exitDatetime = datetime.strptime(match[1], "%a %b %d %H:%M:%S %Y")

387 else:

388 pass

389

390 line = yield line

391

392 # todo: should we receive and expect an ned-token like None?

393 return line

394

395

396@export

397class Task(BaseParser, VivadoMessagesMixin):

398 """

399 A task's output emitted by a Vivado command.

400

401 .. rubric:: Extracted information

402

403 * Vivado messages (info, warning, critical warning, error).

404

405 .. rubric:: Example

406

407 .. code-block::

408

409 Starting Cache Timing Information Task

410 INFO: [Timing 38-35] 79-Done setting XDC timing constraints.

411 Ending Cache Timing Information Task | Checksum: 19fe8cb97

412

413 Time (s): cpu = 00:00:09 ; elapsed = 00:00:09 . Memory (MB): peak = 1370.594 ; gain = 493.266

414

415 """

416 # _NAME: ClassVar[str]

417 # _START: ClassVar[str]

418 # _FINISH: ClassVar[str]

419 _TIME: ClassVar[str] = "Time (s):"

420

421 _command: "Command" #: Reference to the command (parent).

422 _duration: float #: Duration of a task according to reported times by Vivado.

423

424 def __init__(self, command: "Command") -> None:

425 """

426 Initializes a task (without child elements).

427

428 :param command: Reference to the command.

429 """

430 super().__init__()

431 VivadoMessagesMixin.__init__(self)

432

433 self._command = command

434

435 @readonly

436 def Command(self) -> "Command":

437 """

438 Read-only property to access the command.

439

440 :returns: The command this task's output was logged for.

441 """

442 return self._command

443

444 def _TaskStart(self, line: Line) -> Generator[Line, Line, Line]:

445 """

446 A generator for processing a task start (single line).

447

448 :param line: First line to process (task start).

449 :returns: A generator processing log messages.

450 :raises ProcessorException: If first line doesn't conform to the *task start* pattern.

451 """

452 if not line.StartsWith(self._START): 452 ↛ 453line 452 didn't jump to line 453 because the condition on line 452 was never true

453 raise ProcessorException(f"{self.__class__.__name__}._TaskStart(): Expected '{self._START}' at line {line._lineNumber}.")

454

455 line._kind = LineKind.TaskStart

456 nextLine = yield line

457 return nextLine

458

459 def _TaskFinish(self, line: Line) -> Generator[Line, Line, Line]:

460 """

461 A generator for processing a task finish line-by-line.

462

463 :param line: First line to process (task finish).

464 :returns: A generator processing log messages.

465 :raises ProcessorException: If finish line doesn't conform to the *task finish* pattern.

466 """

467 if not line.StartsWith(self._FINISH): 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true

468 raise ProcessorException(f"{self.__class__.__name__}._TaskFinish(): Expected '{self._FINISH}' at line {line._lineNumber}.")

469

470 line._kind = LineKind.TaskEnd

471 line = yield line

472 while self._TIME is not None: # TODO: limit search for time pattern to XX lines 472 ↛ 479line 472 didn't jump to line 479 because the condition on line 472 was always true

473 if line.StartsWith(self._TIME):

474 line._kind = LineKind.TaskTime

475 break

476

477 line = yield line

478

479 nextLine = yield line

480 return nextLine

481

482 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

483 """

484 A generator for processing a task without child elements line-by-line.

485

486 .. rubric:: Algorithm

487

488 1. Send first line to :meth:`_TaskStart`.

489 2. Process body lines

490

491 * Collect Vivado messages (info, warning, critical warning, error).

492 * Check for *task finish* pattern.

493 * Check for *time* pattern.

494

495 3. Send last lines to :meth:`_TaskFinish`.

496

497 :param line: First line to process.

498 :returns: A generator processing log messages.

499 """

500 line = yield from self._TaskStart(line)

501

502 while True:

503 if line._kind is LineKind.Empty:

504 line = yield line

505 continue

506 elif self._FINISH is not None and line.StartsWith("Ending"):

507 break

508 elif isinstance(line, VivadoMessage):

509 self._AddMessage(line)

510 elif line.StartsWith(self._TIME):

511 line._kind = LineKind.TaskTime

512 nextLine = yield line

513 return nextLine

514

515 line = yield line

516

517 nextLine = yield from self._TaskFinish(line)

518 return nextLine

519

520 def __str__(self) -> str:

521 return f"{self.__class__.__name__}: {self._START}"

522

523

524@export

525class TaskWithSubTasks(Task):

526 """

527 A task's output emitted by a Vivado command.

528

529 .. rubric:: Extracted information

530

531 * Vivado messages (info, warning, critical warning, error).

532 * Subtasks

533

534 .. rubric:: Example

535

536 .. code-block::

537

538 Starting Cache Timing Information Task

539 INFO: [Timing 38-35] 79-Done setting XDC timing constraints.

540 Ending Cache Timing Information Task | Checksum: 19fe8cb97

541

542 Time (s): cpu = 00:00:09 ; elapsed = 00:00:09 . Memory (MB): peak = 1370.594 ; gain = 493.266

543

544 """

545 # _PARSERS: ClassVar[Tuple[Type["SubTask"], ...]]

546

547 _subtasks: Dict[Type["SubTask"], "SubTask"]

548

549 def __init__(self, command: "Command") -> None:

550 super().__init__(command)

551

552 self._subtasks = {p: p(self) for p in self._PARSERS}

553

554 @readonly

555 def SubTasks(self) -> Dict[Type["SubTask"], "SubTask"]:

556 return self._subtasks

557

558 def __contains__(self, key: Any) -> bool:

559 if not issubclass(key, SubTask): 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true

560 ex = TypeError(f"Parameter 'key' is not a Subtask.")

561 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

562 raise ex

563

564 return key in self._subtasks

565

566 def __getitem__(self, key: Type["SubTask"]) -> "SubTask":

567 try:

568 return self._subtasks[key]

569 except KeyError as ex:

570 raise SubTaskNotPresentException(F"Subtask '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

571

572 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

573 line = yield from self._TaskStart(line)

574

575 activeParsers: List[Phase] = list(self._subtasks.values())

576

577 while True:

578 while True:

579 if line._kind is LineKind.Empty:

580 line = yield line

581 continue

582 elif isinstance(line, VivadoMessage):

583 self._AddMessage(line)

584 elif line.StartsWith("Starting "):

585 for parser in activeParsers: # type: SubTask 585 ↛ 590line 585 didn't jump to line 590 because the loop on line 585 didn't complete

586 if line.StartsWith(parser._START): 586 ↛ 585line 586 didn't jump to line 585 because the condition on line 586 was always true

587 line = yield next(subtask := parser.Generator(line))

588 break

589 else:

590 WarningCollector.Raise(UnknownSubTask(f"Unknown subtask: '{line!r}'", line))

591 ex = Exception(f"How to recover from here? Unknown subtask: '{line!r}'")

592 ex.add_note(f"Current task: start pattern='{self}'")

593 ex.add_note(f"Current command: {self._command}")

594 raise ex

595 break

596 elif line.StartsWith("Ending"):

597 nextLine = yield from self._TaskFinish(line)

598 return nextLine

599 elif line.StartsWith(self._TIME): 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true

600 line._kind = LineKind.TaskTime

601 nextLine = yield line

602 return nextLine

603

604 line = yield line

605

606 while True:

607 isFinish = False # line.StartsWith("Ending") # FIXME: detect end, but time might come later

608

609 try:

610 processedLine = subtask.send(line)

611

612 if isinstance(processedLine, VivadoMessage):

613 self._AddMessage(processedLine)

614

615 if isFinish: 615 ↛ 616line 615 didn't jump to line 616 because the condition on line 615 was never true

616 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

617 line = yield processedLine

618 break

619 except StopIteration as ex:

620 activeParsers.remove(parser)

621 line = ex.value

622 break

623

624 line = yield processedLine

625

626

627@export

628class SubTask(BaseParser, VivadoMessagesMixin):

629 # _NAME: ClassVar[str]

630 # _START: ClassVar[str]

631 # _FINISH: ClassVar[str]

632 _TIME: ClassVar[str] = "Time (s):"

633

634 _task: TaskWithSubTasks

635 _duration: float

636

637 def __init__(self, task: TaskWithSubTasks) -> None:

638 super().__init__()

639 VivadoMessagesMixin.__init__(self)

640

641 self._task = task

642

643 @readonly

644 def Task(self) -> TaskWithSubTasks:

645 return self._task

646

647 def _TaskStart(self, line: Line) -> Generator[Line, Line, Line]:

648 if not line.StartsWith(self._START): 648 ↛ 649line 648 didn't jump to line 649 because the condition on line 648 was never true

649 raise ProcessorException(f"{self.__class__.__name__}._TaskStart(): Expected '{self._START}' at line {line._lineNumber}.")

650

651 line._kind = LineKind.TaskStart

652 nextLine = yield line

653 return nextLine

654

655 def _TaskFinish(self, line: Line) -> Generator[Line, Line, Line]:

656 if not line.StartsWith(self._FINISH): 656 ↛ 657line 656 didn't jump to line 657 because the condition on line 656 was never true

657 raise ProcessorException(f"{self.__class__.__name__}._TaskFinish(): Expected '{self._FINISH}' at line {line._lineNumber}.")

658

659 line._kind = LineKind.TaskEnd

660 line = yield line

661 while self._TIME is not None: 661 ↛ 668line 661 didn't jump to line 668 because the condition on line 661 was always true

662 if line.StartsWith(self._TIME):

663 line._kind = LineKind.TaskTime

664 break

665

666 line = yield line

667

668 nextLine = yield line

669 return nextLine

670

671 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

672 line = yield from self._TaskStart(line)

673

674 while True:

675 if line._kind is LineKind.Empty: 675 ↛ 676line 675 didn't jump to line 676 because the condition on line 675 was never true

676 line = yield line

677 continue

678 elif self._FINISH is not None and line.StartsWith("Ending"):

679 break

680 elif isinstance(line, VivadoMessage):

681 self._AddMessage(line)

682 elif line.StartsWith(self._TIME): 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true

683 line._kind = LineKind.TaskTime

684 nextLine = yield line

685 return nextLine

686

687 line = yield line

688

689 nextLine = yield from self._TaskFinish(line)

690 return nextLine

691

692 def __str__(self) -> str:

693 return self._NAME

694

695

696@export

697class TaskWithPhases(Task):

698 # _PARSERS: ClassVar[Tuple[Type["Phase"], ...]]

699

700 _phases: Dict[Type["Phase"], "Phase"]

701

702 def __init__(self, command: "Command") -> None:

703 super().__init__(command)

704

705 self._phases = {p: p(self) for p in self._PARSERS}

706

707 @readonly

708 def Phases(self) -> Dict[Type["Phase"], "Phase"]:

709 return self._phases

710

711 def __contains__(self, key: Any) -> bool:

712 if not issubclass(key, Phase): 712 ↛ 713line 712 didn't jump to line 713 because the condition on line 712 was never true

713 ex = TypeError(f"Parameter 'key' is not a Phase.")

714 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

715 raise ex

716

717 return key in self._phases

718

719 def __getitem__(self, key: Type["Phase"]) -> "Phase":

720 try:

721 return self._phases[key]

722 except KeyError as ex:

723 raise PhaseNotPresentException(F"Phase '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

724

725 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

726 line = yield from self._TaskStart(line)

727

728 activeParsers: List[Phase] = list(self._phases.values())

729

730 while True:

731 while True:

732 if line._kind is LineKind.Empty:

733 line = yield line

734 continue

735 elif isinstance(line, VivadoMessage):

736 self._AddMessage(line)

737 elif line.StartsWith("Phase "):

738 for parser in activeParsers: # type: Phase 738 ↛ 743line 738 didn't jump to line 743 because the loop on line 738 didn't complete

739 if (match := parser._START.match(line._message)) is not None:

740 line = yield next(phase := parser.Generator(line))

741 break

742 else:

743 WarningCollector.Raise(UnknownPhase(f"Unknown phase: '{line!r}'", line))

744 ex = Exception(f"How to recover from here? Unknown phase: '{line!r}'")

745 ex.add_note(f"Current task: start pattern='{self}'")

746 ex.add_note(f"Current command: {self._command}")

747 raise ex

748 break

749 elif line.StartsWith("Ending"):

750 nextLine = yield from self._TaskFinish(line)

751 return nextLine

752 elif line.StartsWith(self._TIME):

753 line._kind = LineKind.TaskTime

754 nextLine = yield line

755 return nextLine

756

757 line = yield line

758

759 while True:

760 isFinish = False #line.StartsWith("Ending")

761

762 try:

763 processedLine = phase.send(line)

764

765 if isinstance(processedLine, VivadoMessage):

766 self._AddMessage(processedLine)

767

768 if isFinish: 768 ↛ 769line 768 didn't jump to line 769 because the condition on line 768 was never true

769 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

770 line = yield processedLine

771 break

772 except StopIteration as ex:

773 activeParsers.remove(parser)

774 line = ex.value

775 break

776

777 line = yield processedLine

778

779

780@export

781class Phase(BaseParser, VivadoMessagesMixin):

782 # _NAME: ClassVar[str]

783 # _START: ClassVar[str]

784 # _FINISH: ClassVar[str]

785 _TIME: ClassVar[str] = "Time (s):"

786 _FINAL: ClassVar[Nullable[str]] = None

787

788 _task: TaskWithPhases

789 _phaseIndex: int

790 _duration: float

791

792 def __init__(self, task: TaskWithPhases) -> None:

793 super().__init__()

794 VivadoMessagesMixin.__init__(self)

795

796 self._task = task

797 self._phaseIndex = None

798

799 @readonly

800 def Task(self) -> TaskWithPhases:

801 return self._task

802

803 def _PhaseStart(self, line: Line) -> Generator[Line, Line, Line]:

804 if (match := self._START.match(line._message)) is None: 804 ↛ 805line 804 didn't jump to line 805 because the condition on line 804 was never true

805 raise ProcessorException(f"{self.__class__.__name__}._PhaseStart(): Expected '{self._START}' at line {line._lineNumber}.")

806

807 self._phaseIndex = int(match["major"])

808

809 line._kind = LineKind.PhaseStart

810 nextLine = yield line

811 return nextLine

812

813 def _PhaseFinish(self, line: Line) -> Generator[Line, Line, None]:

814 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)

815 if not line.StartsWith(FINISH): 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true

816 raise ProcessorException(f"{self.__class__.__name__}._PhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")

817

818 line._kind = LineKind.PhaseEnd

819 line = yield line

820

821 if self._TIME is not None: 821 ↛ 833line 821 didn't jump to line 833 because the condition on line 821 was always true

822 while True:

823 if line.StartsWith(self._TIME):

824 line._kind = LineKind.PhaseTime

825 break

826 elif isinstance(line, VivadoMessage): 826 ↛ 827line 826 didn't jump to line 827 because the condition on line 826 was never true

827 self._AddMessage(line)

828

829 line = yield line

830

831 line = yield line

832

833 if self._FINAL is not None and self._task._command._processor._preamble._toolVersion >= "2023.2":

834 while True:

835 if line.StartsWith(self._FINAL): 835 ↛ 838line 835 didn't jump to line 838 because the condition on line 835 was always true

836 line._kind = LineKind.PhaseFinal

837 break

838 elif isinstance(line, VivadoMessage):

839 self._AddMessage(line)

840

841 line = yield line

842

843 line = yield line

844

845 # TODO: optionally collect following INFO messages like 31-389, 31-1021, 31-662

846

847 return line

848

849 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

850 line = yield from self._PhaseStart(line)

851

852 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)

853

854 while True:

855 if line._kind is LineKind.Empty:

856 line = yield line

857 continue

858 elif isinstance(line, VivadoMessage):

859 self._AddMessage(line)

860 elif line.StartsWith(FINISH):

861 break

862

863 line = yield line

864

865 nextLine = yield from self._PhaseFinish(line)

866 return nextLine

867

868 def __str__(self) -> str:

869 return f"{self.__class__.__name__}: {self._START.pattern}"

870

871

872@export

873class PhaseWithChildren(Phase):

874 _SUBPHASE_PREFIX: ClassVar[str] = "Phase {phaseIndex}."

875

876 _subPhases: Dict[Type["SubPhase"], "SubPhase"]

877

878 def __init__(self, task: TaskWithPhases) -> None:

879 super().__init__(task)

880

881 self._subPhases = {p: p(self) for p in self._PARSERS}

882

883 def __contains__(self, key: Any) -> bool:

884 if not issubclass(key, SubPhase):

885 ex = TypeError(f"Parameter 'item' is not a SubPhase.")

886 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

887 raise ex

888

889 return key in self._subPhases

890

891 def __getitem__(self, key: Type["SubPhase"]) -> "SubPhase":

892 try:

893 return self._subPhases[key]

894 except KeyError as ex:

895 raise PhaseNotPresentException(F"SubPhase '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

896

897 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

898 line = yield from self._PhaseStart(line)

899

900 activeParsers: List[SubPhase] = list(self._subPhases.values())

901

902 SUBPHASE_PREFIX = self._SUBPHASE_PREFIX.format(phaseIndex=self._phaseIndex)

903 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)

904

905 while True:

906 while True:

907 if line._kind is LineKind.Empty:

908 line = yield line

909 continue

910 elif isinstance(line, VivadoMessage):

911 self._AddMessage(line)

912 elif line.StartsWith(SUBPHASE_PREFIX):

913 for parser in activeParsers: # type: Section 913 ↛ 918line 913 didn't jump to line 918 because the loop on line 913 didn't complete

914 if (match := parser._START.match(line._message)) is not None:

915 line = yield next(phase := parser.Generator(line))

916 break

917 else:

918 WarningCollector.Raise(UnknownSubPhase(f"Unknown subphase: '{line!r}'", line))

919 ex = Exception(f"How to recover from here? Unknown subphase: '{line!r}'")

920 ex.add_note(f"Current phase: start pattern='{self}'")

921 ex.add_note(f"Current task: start pattern='{self._task}'")

922 ex.add_note(f"Current command: {self._task._command}")

923 raise ex

924 break

925 elif line.StartsWith(FINISH):

926 nextLine = yield from self._PhaseFinish(line)

927 return nextLine

928

929 line = yield line

930

931 while True:

932 isFinish = False # line.StartsWith(SUBPHASE_PREFIX) # FIXME: detect end, but end (e.g. time) is later then ending text

933

934 try:

935 processedLine = phase.send(line)

936

937 if isinstance(processedLine, VivadoMessage):

938 self._AddMessage(processedLine)

939

940 if isFinish: 940 ↛ 941line 940 didn't jump to line 941 because the condition on line 940 was never true

941 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

942 line = yield processedLine

943 break

944 except StopIteration as ex:

945 activeParsers.remove(parser)

946 line = ex.value

947 break

948

949 line = yield processedLine

950

951

952@export

953class SubPhase(BaseParser, VivadoMessagesMixin):

954 # _NAME: ClassVar[str]

955 # _START: ClassVar[str]

956 # _FINISH: ClassVar[str]

957

958 _phase: Phase

959 _phaseIndex: int

960 _subPhaseIndex: int

961 _duration: float

962

963 def __init__(self, phase: Phase) -> None:

964 super().__init__()

965 VivadoMessagesMixin.__init__(self)

966

967 self._phase = phase

968 self._phaseIndex = None

969 self._subPhaseIndex = None

970

971 def _SubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:

972 if (match := self._START.match(line._message)) is None: 972 ↛ 973line 972 didn't jump to line 973 because the condition on line 972 was never true

973 raise ProcessorException(f"{self.__class__.__name__}._SubPhaseStart(): Expected '{self._START}' at line {line._lineNumber}.")

974

975 self._phaseIndex = int(match["major"])

976 self._subPhaseIndex = int(match["minor"])

977

978 line._kind = LineKind.SubPhaseStart

979 nextLine = yield line

980 return nextLine

981

982 def _SubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:

983 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)

984

985 if not line.StartsWith(FINISH): 985 ↛ 986line 985 didn't jump to line 986 because the condition on line 985 was never true

986 raise ProcessorException(f"{self.__class__.__name__}._SubPhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")

987

988 if self._TIME is None:

989 line._kind = LineKind.SubPhaseTime

990 else:

991 line._kind = LineKind.SubPhaseEnd

992

993 line = yield line

994 while self._TIME is not None: 994 ↛ 1001line 994 didn't jump to line 1001 because the condition on line 994 was always true

995 if line.StartsWith(self._TIME):

996 line._kind = LineKind.SubPhaseTime

997 break

998

999 line = yield line

1000

1001 nextLine = yield line

1002 return nextLine

1003

1004 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1005 line = yield from self._SubPhaseStart(line)

1006

1007 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)

1008

1009 while True:

1010 if line._kind is LineKind.Empty:

1011 line = yield line

1012 continue

1013 elif line.StartsWith(FINISH):

1014 break

1015 elif isinstance(line, VivadoMessage):

1016 self._AddMessage(line)

1017

1018 line = yield line

1019

1020 nextLine = yield from self._SubPhaseFinish(line)

1021 return nextLine

1022

1023 def __str__(self) -> str:

1024 return f"{self.__class__.__name__}: {self._START.pattern}"

1025

1026

1027@export

1028class SubPhaseWithChildren(SubPhase):

1029 _subSubPhases: Dict[Type["SubSubPhase"], "SubSubPhase"]

1030

1031 def __init__(self, phase: Phase) -> None:

1032 super().__init__(phase)

1033

1034 self._subSubPhases = {p: p(self) for p in self._PARSERS}

1035

1036 def __contains__(self, key: Any) -> bool:

1037 if not issubclass(key, SubSubPhase):

1038 ex = TypeError(f"Parameter 'item' is not a SubSubPhase.")

1039 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

1040 raise ex

1041

1042 return key in self._subSubPhases

1043

1044 def __getitem__(self, key: Type["SubSubPhase"]) -> "SubSubPhase":

1045 try:

1046 return self._subSubPhases[key]

1047 except KeyError as ex:

1048 raise PhaseNotPresentException(F"SubSubPhase '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

1049

1050 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1051 line = yield from self._SubPhaseStart(line)

1052

1053 activeParsers: List["SubSubPhase"] = list(self._subSubPhases.values())

1054

1055 START_PREFIX = f"Phase {self._phaseIndex}.{self._subPhaseIndex}."

1056 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex)

1057

1058 while True:

1059 while True:

1060 if line._kind is LineKind.Empty:

1061 line = yield line

1062 continue

1063 elif isinstance(line, VivadoMessage):

1064 self._AddMessage(line)

1065 elif line.StartsWith(START_PREFIX):

1066 for parser in activeParsers: # type: SubSubPhase 1066 ↛ 1071line 1066 didn't jump to line 1071 because the loop on line 1066 didn't complete

1067 if (match := parser._START.match(line._message)) is not None:

1068 line = yield next(phase := parser.Generator(line))

1069 break

1070 else:

1071 WarningCollector.Raise(UnknownSubPhase(f"Unknown subsubphase: '{line!r}'", line))

1072 ex = Exception(f"How to recover from here? Unknown subsubphase: '{line!r}'")

1073 ex.add_note(f"Current subphase: start pattern='{self}'")

1074 ex.add_note(f"Current phase: start pattern='{self._phase}'")

1075 ex.add_note(f"Current task: start pattern='{self._phase._task}'")

1076 ex.add_note(f"Current cmd: {self._phase._task._command}")

1077 raise ex

1078 break

1079 elif line.StartsWith(FINISH): 1079 ↛ 1083line 1079 didn't jump to line 1083 because the condition on line 1079 was always true

1080 nextLine = yield from self._SubPhaseFinish(line)

1081 return nextLine

1082

1083 line = yield line

1084

1085 while True:

1086 isFinish = False # line.StartsWith("Ending")

1087

1088 try:

1089 processedLine = phase.send(line)

1090

1091 if isinstance(processedLine, VivadoMessage):

1092 self._AddMessage(processedLine)

1093

1094 if isFinish: 1094 ↛ 1095line 1094 didn't jump to line 1095 because the condition on line 1094 was never true

1095 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

1096 line = yield processedLine

1097 break

1098 except StopIteration as ex:

1099 activeParsers.remove(parser)

1100 line = ex.value

1101 break

1102

1103 line = yield processedLine

1104

1105

1106@export

1107class SubSubPhase(BaseParser, VivadoMessagesMixin):

1108 # _NAME: ClassVar[str]

1109 # _START: ClassVar[str]

1110 # _FINISH: ClassVar[str]

1111

1112 _subphase: SubPhase

1113 _phaseIndex: int

1114 _subPhaseIndex: int

1115 _subSubPhaseIndex: int

1116 _duration: float

1117

1118 def __init__(self, subphase: SubPhase) -> None:

1119 super().__init__()

1120 VivadoMessagesMixin.__init__(self)

1121

1122 self._subphase = subphase

1123 self._phaseIndex = None

1124 self._subPhaseIndex = None

1125 self._subSubPhaseIndex = None

1126

1127 def _SubSubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:

1128 if (match := self._START.match(line._message)) is None: 1128 ↛ 1129line 1128 didn't jump to line 1129 because the condition on line 1128 was never true

1129 raise ProcessorException()

1130

1131 self._phaseIndex = int(match["major"])

1132 self._subPhaseIndex = int(match["minor"])

1133 self._subSubPhaseIndex = int(match["micro"])

1134

1135 line._kind = LineKind.SubSubPhaseStart

1136 nextLine = yield line

1137 return nextLine

1138

1139 def _SubSubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:

1140 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex)

1141

1142 if not line.StartsWith(FINISH): 1142 ↛ 1143line 1142 didn't jump to line 1143 because the condition on line 1142 was never true

1143 raise ProcessorException(f"{self.__class__.__name__}._SubSubPhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")

1144

1145 line._kind = LineKind.SubSubPhaseEnd

1146 line = yield line

1147

1148 while self._TIME is not None: 1148 ↛ 1155line 1148 didn't jump to line 1155 because the condition on line 1148 was always true

1149 if line.StartsWith(self._TIME):

1150 line._kind = LineKind.SubSubPhaseTime

1151 break

1152

1153 line = yield line

1154

1155 nextLine = yield line

1156 return nextLine

1157

1158 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1159 line = yield from self._SubSubPhaseStart(line)

1160

1161 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex)

1162

1163 while True:

1164 if line._kind is LineKind.Empty:

1165 line = yield line

1166 continue

1167 elif line.StartsWith(FINISH):

1168 break

1169 elif isinstance(line, VivadoMessage):

1170 self._AddMessage(line)

1171

1172 line = yield line

1173

1174 nextLine = yield from self._SubSubPhaseFinish(line)

1175 return nextLine

1176

1177 def __str__(self) -> str:

1178 return f"{self.__class__.__name__}: {self._START.pattern}"

1179

1180

1181@export

1182class SubSubPhaseWithChildren(SubSubPhase):

1183 _subSubSubPhases: Dict[Type["SubSubSubPhase"], "SubSubSubPhase"]

1184

1185 def __init__(self, subphase: SubPhase) -> None:

1186 super().__init__(subphase)

1187

1188 self._subSubSubPhases = {p: p(self) for p in self._PARSERS}

1189

1190 def __contains__(self, key: Any) -> bool:

1191 if not issubclass(key, SubSubSubPhase):

1192 ex = TypeError(f"Parameter 'item' is not a SubSubSubPhase.")

1193 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

1194 raise ex

1195

1196 return key in self._subSubSubPhases

1197

1198 def __getitem__(self, key: Type["SubSubSubPhase"]) -> "SubSubSubPhase":

1199 try:

1200 return self._subSubSubPhases[key]

1201 except KeyError as ex:

1202 raise PhaseNotPresentException(F"SubSubSubPhase '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

1203

1204 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1205 line = yield from self._SubSubPhaseStart(line)

1206

1207 activeParsers: List["SubSubSubPhase"] = list(self._subSubSubPhases.values())

1208

1209 START_PREFIX = f"Phase {self._phaseIndex}.{self._subPhaseIndex}.{self._subSubPhaseIndex}."

1210

1211 while True:

1212 while True:

1213 if line._kind is LineKind.Empty:

1214 line = yield line

1215 continue

1216 elif isinstance(line, VivadoMessage):

1217 self._AddMessage(line)

1218 elif line.StartsWith(START_PREFIX):

1219 for parser in activeParsers: # type: SubSubSubPhase 1219 ↛ 1224line 1219 didn't jump to line 1224 because the loop on line 1219 didn't complete

1220 if (match := parser._START.match(line._message)) is not None: 1220 ↛ 1219line 1220 didn't jump to line 1219 because the condition on line 1220 was always true

1221 line = yield next(phase := parser.Generator(line))

1222 break

1223 else:

1224 WarningCollector.Raise(UnknownSubPhase(f"Unknown subsubsubphase: '{line!r}'", line))

1225 ex = Exception(f"How to recover from here? Unknown subsubsubphase: '{line!r}'")

1226 ex.add_note(f"Current subsubphase: start pattern='{self}'")

1227 ex.add_note(f"Current subphase: start pattern='{self._subphase}'")

1228 ex.add_note(f"Current phase: start pattern='{self._subphase._phase}'")

1229 ex.add_note(f"Current task: start pattern='{self._subphase._phase._task}'")

1230 ex.add_note(f"Current cmd: {self._subphase._phase._task._command}")

1231 raise ex

1232 break

1233 elif line.StartsWith(self._TIME):

1234 line._kind = LineKind.SubSubPhaseTime

1235 nextLine = yield line

1236 return nextLine

1237

1238 line = yield line

1239

1240 while True:

1241 isFinish = False # line.StartsWith("Ending")

1242

1243 try:

1244 processedLine = phase.send(line)

1245

1246 if isinstance(processedLine, VivadoMessage):

1247 self._AddMessage(processedLine)

1248

1249 if isFinish: 1249 ↛ 1250line 1249 didn't jump to line 1250 because the condition on line 1249 was never true

1250 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

1251 line = yield processedLine

1252 break

1253 except StopIteration as ex:

1254 activeParsers.remove(parser)

1255 line = ex.value

1256 break

1257

1258 line = yield processedLine

1259

1260

1261@export

1262class SubSubSubPhase(BaseParser, VivadoMessagesMixin):

1263 # _NAME: ClassVar[str]

1264 # _START: ClassVar[str]

1265 # _FINISH: ClassVar[str]

1266

1267 _subsubphase: SubSubPhase

1268 _phaseIndex: int

1269 _subPhaseIndex: int

1270 _subSubPhaseIndex: int

1271 _subSubSubPhaseIndex: int

1272 _duration: float

1273

1274 def __init__(self, subsubphase: SubSubPhase) -> None:

1275 super().__init__()

1276 VivadoMessagesMixin.__init__(self)

1277

1278 self._subsubphase = subsubphase

1279 self._phaseIndex = None

1280 self._subPhaseIndex = None

1281 self._subSubPhaseIndex = None

1282 self._subSubSubPhaseIndex = None

1283

1284 def _SubSubSubPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:

1285 if (match := self._START.match(line._message)) is None: 1285 ↛ 1286line 1285 didn't jump to line 1286 because the condition on line 1285 was never true

1286 raise ProcessorException()

1287

1288 self._phaseIndex = int(match["major"])

1289 self._subPhaseIndex = int(match["minor"])

1290 self._subSubPhaseIndex = int(match["micro"])

1291 self._subSubSubPhaseIndex = int(match["nano"])

1292

1293 line._kind = LineKind.SubSubSubPhaseStart

1294 nextLine = yield line

1295 return nextLine

1296

1297 def _SubSubSubPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:

1298 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex, subSubSubPhaseIndex=self._subSubSubPhaseIndex)

1299

1300 if not line.StartsWith(FINISH): 1300 ↛ 1301line 1300 didn't jump to line 1301 because the condition on line 1300 was never true

1301 raise ProcessorException(f"{self.__class__.__name__}._SubSubSubPhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")

1302

1303 line._kind = LineKind.SubSubSubPhaseEnd

1304 line = yield line

1305

1306 while self._TIME is not None: 1306 ↛ 1313line 1306 didn't jump to line 1313 because the condition on line 1306 was always true

1307 if line.StartsWith(self._TIME):

1308 line._kind = LineKind.SubSubSubPhaseTime

1309 break

1310

1311 line = yield line

1312

1313 nextLine = yield line

1314 return nextLine

1315

1316 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1317 line = yield from self._SubSubSubPhaseStart(line)

1318

1319 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex, subPhaseIndex=self._subPhaseIndex, subSubPhaseIndex=self._subSubPhaseIndex, subSubSubPhaseIndex=self._subSubSubPhaseIndex)

1320

1321 while True:

1322 if line._kind is LineKind.Empty: 1322 ↛ 1323line 1322 didn't jump to line 1323 because the condition on line 1322 was never true

1323 line = yield line

1324 continue

1325 elif line.StartsWith(FINISH):

1326 break

1327 elif isinstance(line, VivadoMessage): 1327 ↛ 1330line 1327 didn't jump to line 1330 because the condition on line 1327 was always true

1328 self._AddMessage(line)

1329

1330 line = yield line

1331

1332 nextLine = yield from self._SubSubSubPhaseFinish(line)

1333 return nextLine

1334

1335 def __str__(self) -> str:

1336 return f"{self.__class__.__name__}: {self._START.pattern}"

1337

1338

1339@export

1340class SubSubSubPhaseWithTasks(SubSubSubPhase):

1341 _nestedTasks: Dict[Type["NestedTask"], "NestedTask"]

1342

1343 def __init__(self, subsubphase: SubSubPhase) -> None:

1344 super().__init__(subsubphase)

1345

1346 self._nestedTasks = {p: p(self) for p in self._PARSERS}

1347

1348 def __contains__(self, key: Any) -> bool:

1349 if not issubclass(key, NestedTask):

1350 ex = TypeError(f"Parameter 'key' is not a NestedTask.")

1351 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

1352 raise ex

1353

1354 return key in self._nestedTasks

1355

1356 def __getitem__(self, key: Type["NestedTask"]) -> "NestedTask":

1357 try:

1358 return self._nestedTasks[key]

1359 except KeyError as ex:

1360 raise NestedTaskNotPresentException(F"NestedTask '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

1361

1362 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1363 line = yield from self._SubSubSubPhaseStart(line)

1364

1365 activeParsers: List["NestedTask"] = list(self._nestedTasks.values())

1366

1367 # START_PREFIX = f"Phase {self._phaseIndex}.{self._subPhaseIndex}.{self._subSubPhaseIndex}."

1368

1369 while True:

1370 while True:

1371 if line._kind is LineKind.Empty:

1372 line = yield line

1373 continue

1374 elif isinstance(line, VivadoMessage):

1375 self._AddMessage(line)

1376 elif line.StartsWith("Starting "):

1377 for parser in activeParsers: # type: NestedTask 1377 ↛ 1382line 1377 didn't jump to line 1382 because the loop on line 1377 didn't complete

1378 if line.StartsWith(parser._START): 1378 ↛ 1377line 1378 didn't jump to line 1377 because the condition on line 1378 was always true

1379 line = yield next(phase := parser.Generator(line))

1380 break

1381 else:

1382 WarningCollector.Raise(UnknownSubPhase(f"Unknown NestedTask: '{line!r}'", line))

1383 ex = Exception(f"How to recover from here? Unknown NestedTask: '{line!r}'")

1384 ex.add_note(f"Current subsubsubphase: start pattern='{self}'")

1385 ex.add_note(f"Current subsubphase: start pattern='{self._subsubphase}'")

1386 ex.add_note(f"Current subphase: start pattern='{self._subsubphase._subphase}'")

1387 ex.add_note(f"Current phase: start pattern='{self._subsubphase._subphase._phase}'")

1388 ex.add_note(f"Current task: start pattern='{self._subsubphase._subphase._phase._task}'")

1389 ex.add_note(f"Current cmd: {self._subsubphase._subphase._phase._task._command}")

1390 raise ex

1391 break

1392 elif line.StartsWith(self._TIME):

1393 line._kind = LineKind.SubSubSubPhaseTime

1394 nextLine = yield line

1395 return nextLine

1396

1397 line = yield line

1398

1399 while True:

1400 isFinish = False # line.StartsWith("Ending")

1401

1402 try:

1403 processedLine = phase.send(line)

1404

1405 if isinstance(processedLine, VivadoMessage):

1406 self._AddMessage(processedLine)

1407

1408 if isFinish: 1408 ↛ 1409line 1408 didn't jump to line 1409 because the condition on line 1408 was never true

1409 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

1410 line = yield processedLine

1411 break

1412 except StopIteration as ex:

1413 activeParsers.remove(parser)

1414 line = ex.value

1415 break

1416

1417 line = yield processedLine

1418

1419

1420@export

1421class NestedTask(BaseParser, VivadoMessagesMixin):

1422 # _NAME: ClassVar[str]

1423 # _START: ClassVar[str]

1424 # _FINISH: ClassVar[str]

1425 _TIME: ClassVar[str] = "Time (s):"

1426

1427 _subsubsubphase: SubSubSubPhaseWithTasks

1428 _duration: float

1429

1430 def __init__(self, subsubsubphase: SubSubSubPhaseWithTasks) -> None:

1431 super().__init__()

1432 VivadoMessagesMixin.__init__(self)

1433

1434 self._subsubsubphase = subsubsubphase

1435

1436 @readonly

1437 def SubSubSubPhase(self) -> SubSubSubPhaseWithTasks:

1438 return self._subsubsubphase

1439

1440 def _NestedTaskStart(self, line: Line) -> Generator[Line, Line, Line]:

1441 if not line.StartsWith(self._START): 1441 ↛ 1442line 1441 didn't jump to line 1442 because the condition on line 1441 was never true

1442 raise ProcessorException(f"{self.__class__.__name__}._TaskStart(): Expected '{self._START}' at line {line._lineNumber}.")

1443

1444 line._kind = LineKind.NestedTaskStart

1445 nextLine = yield line

1446 return nextLine

1447

1448 def _NestedTaskFinish(self, line: Line) -> Generator[Line, Line, Line]:

1449 if not line.StartsWith(self._FINISH): 1449 ↛ 1450line 1449 didn't jump to line 1450 because the condition on line 1449 was never true

1450 raise ProcessorException(f"{self.__class__.__name__}._TaskFinish(): Expected '{self._FINISH}' at line {line._lineNumber}.")

1451

1452 line._kind = LineKind.NestedTaskEnd

1453 line = yield line

1454

1455 if self._TIME is not None: 1455 ↛ 1464line 1455 didn't jump to line 1464 because the condition on line 1455 was always true

1456 while True:

1457 if line.StartsWith(self._TIME):

1458 line._kind = LineKind.TaskTime

1459 line = yield line

1460 break

1461

1462 line = yield line

1463

1464 return line

1465

1466 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1467 line = yield from self._NestedTaskStart(line)

1468

1469 while True:

1470 if line._kind is LineKind.Empty:

1471 line = yield line

1472 continue

1473 elif self._FINISH is not None and line.StartsWith("Ending"):

1474 break

1475 elif isinstance(line, VivadoMessage):

1476 self._AddMessage(line)

1477 elif line.StartsWith(self._TIME):

1478 line._kind = LineKind.TaskTime

1479 nextLine = yield line

1480 return nextLine

1481

1482 line = yield line

1483

1484 nextLine = yield from self._NestedTaskFinish(line)

1485 return nextLine

1486

1487 def __str__(self) -> str:

1488 return self._NAME

1489

1490

1491@export

1492class NestedTaskWithPhases(NestedTask):

1493 """

1494 A task's output emitted by a Vivado command.

1495

1496 .. rubric:: Extracted information

1497

1498 * Vivado messages (info, warning, critical warning, error).

1499 * Nested phases

1500

1501 .. rubric:: Example

1502

1503 .. code-block::

1504

1505 Phase 4.1.1.1 BUFG Insertion

1506

1507 Starting Physical Synthesis Task

1508

1509 Phase 1 Physical Synthesis Initialization

1510 INFO: [Physopt 32-721] Multithreading enabled for phys_opt_design using a maximum of 2 CPUs

1511 INFO: [Physopt 32-619] Estimated Timing Summary | WNS=-0.733 | TNS=-0.936 |

1512 Phase 1 Physical Synthesis Initialization | Checksum: 1818afcc0

1513

1514 Time (s): cpu = 00:00:00 ; elapsed = 00:00:00.014 . Memory (MB): peak = 1865.645 ; gain = 0.000

1515 INFO: [Place 46-56] BUFG insertion identified 0 candidate nets. Inserted BUFG: 0, Replicated BUFG Driver: 0, Skipped due to Placement/Routing Conflicts: 0, Skipped due to Timing Degradation: 0, Skipped due to netlist editing failed: 0.

1516 Ending Physical Synthesis Task | Checksum: 22839c186

1517

1518 Time (s): cpu = 00:00:00 ; elapsed = 00:00:00.016 . Memory (MB): peak = 1865.645 ; gain = 0.000

1519 Phase 4.1.1.1 BUFG Insertion | Checksum: 1a8cbaaf2

1520 """

1521 # _PARSERS: ClassVar[Tuple[Type["SubTask"], ...]]

1522

1523 _nestedPhases: Dict[Type["NestedPhase"], "NestedPhase"]

1524

1525 def __init__(self, subsubsubPhase: SubSubSubPhaseWithTasks) -> None:

1526 super().__init__(subsubsubPhase)

1527

1528 self._nestedPhases = {p: p(self) for p in self._PARSERS}

1529

1530 @readonly

1531 def NestedPhases(self) -> Dict[Type["NestedPhase"], "NestedPhase"]:

1532 return self._nestedPhases

1533

1534 def __contains__(self, key: Any) -> bool:

1535 if not issubclass(key, NestedPhase):

1536 ex = TypeError(f"Parameter 'key' is not a NestedPhase.")

1537 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")

1538 raise ex

1539

1540 return key in self._nestedPhases

1541

1542 def __getitem__(self, key: Type["NestedPhase"]) -> "NestedPhase":

1543 try:

1544 return self._nestedPhases[key]

1545 except KeyError as ex:

1546 raise SubTaskNotPresentException(F"NestedPhase '{key._NAME}' not present in '{self._parent.logfile}'.") from ex

1547

1548 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1549 line = yield from self._NestedTaskStart(line)

1550

1551 activeParsers: List[Phase] = list(self._nestedPhases.values())

1552

1553 while True:

1554 while True:

1555 if line._kind is LineKind.Empty:

1556 line = yield line

1557 continue

1558 elif isinstance(line, VivadoMessage):

1559 self._AddMessage(line)

1560 elif line.StartsWith("Phase "):

1561 for parser in activeParsers: # type: NestedPhase 1561 ↛ 1566line 1561 didn't jump to line 1566 because the loop on line 1561 didn't complete

1562 if (match := parser._START.match(line._message)) is not None: 1562 ↛ 1561line 1562 didn't jump to line 1561 because the condition on line 1562 was always true

1563 line = yield next(phase := parser.Generator(line))

1564 break

1565 else:

1566 WarningCollector.Raise(UnknownSubPhase(f"Unknown NestedPhase: '{line!r}'", line))

1567 ex = Exception(f"How to recover from here? Unknown NestedPhase: '{line!r}'")

1568 ex.add_note(f"Current nestedtask: start pattern='{self}'")

1569 ex.add_note(f"Current subsubsubphase: start pattern='{self._subsubsubphase}'")

1570 ex.add_note(f"Current subsubphase: start pattern='{self._subsubsubphase._subsubphase}'")

1571 ex.add_note(f"Current subphase: start pattern='{self._subsubsubphase._subsubphase._subphase}'")

1572 ex.add_note(f"Current phase: start pattern='{self._subsubsubphase._subsubphase._subphase._phase}'")

1573 ex.add_note(f"Current task: start pattern='{self._subsubsubphase._subsubphase._subphase._phase._task}'")

1574 ex.add_note(f"Current cmd: {self._subsubsubphase._subsubphase._subphase._phase._task._command}")

1575 raise ex

1576 break

1577 elif line.StartsWith("Ending"): 1577 ↛ 1580line 1577 didn't jump to line 1580 because the condition on line 1577 was always true

1578 nextLine = yield from self._NestedTaskFinish(line)

1579 return nextLine

1580 elif line.StartsWith(self._TIME):

1581 line._kind = LineKind.TaskTime

1582 nextLine = yield line

1583 return nextLine

1584

1585 line = yield line

1586

1587 while True:

1588 isFinish = False # line.StartsWith("Ending") # FIXME: detect end, but time might come later

1589

1590 try:

1591 processedLine = phase.send(line)

1592

1593 if isinstance(processedLine, VivadoMessage):

1594 self._AddMessage(processedLine)

1595

1596 if isFinish: 1596 ↛ 1597line 1596 didn't jump to line 1597 because the condition on line 1596 was never true

1597 WarningCollector.Raise(UndetectedEnd(f"Didn't detect finish: '{processedLine!r}'", processedLine))

1598 line = yield processedLine

1599 break

1600 except StopIteration as ex:

1601 activeParsers.remove(parser)

1602 line = ex.value

1603 break

1604

1605 line = yield processedLine

1606

1607

1608@export

1609class NestedPhase(BaseParser, VivadoMessagesMixin):

1610 # _NAME: ClassVar[str]

1611 # _START: ClassVar[str]

1612 # _FINISH: ClassVar[str]

1613 _TIME: ClassVar[str] = "Time (s):"

1614 _FINAL: ClassVar[Nullable[str]] = None

1615

1616 _nestedTask: NestedTaskWithPhases

1617 _phaseIndex: int

1618 _duration: float

1619

1620 def __init__(self, nestedTask: TaskWithPhases) -> None:

1621 super().__init__()

1622 VivadoMessagesMixin.__init__(self)

1623

1624 self._nestedTask = nestedTask

1625 self._phaseIndex = None

1626

1627 @readonly

1628 def NestedTask(self) -> NestedTaskWithPhases:

1629 return self._nestedTask

1630

1631 def _NestedPhaseStart(self, line: Line) -> Generator[Line, Line, Line]:

1632 if (match := self._START.match(line._message)) is None: 1632 ↛ 1633line 1632 didn't jump to line 1633 because the condition on line 1632 was never true

1633 raise ProcessorException(f"{self.__class__.__name__}._PhaseStart(): Expected '{self._START}' at line {line._lineNumber}.")

1634

1635 self._phaseIndex = int(match["major"])

1636

1637 line._kind = LineKind.NestedPhaseStart

1638 nextLine = yield line

1639 return nextLine

1640

1641 def _NestedPhaseFinish(self, line: Line) -> Generator[Line, Line, None]:

1642 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)

1643 if not line.StartsWith(FINISH): 1643 ↛ 1644line 1643 didn't jump to line 1644 because the condition on line 1643 was never true

1644 raise ProcessorException(f"{self.__class__.__name__}._PhaseFinish(): Expected '{FINISH}' at line {line._lineNumber}.")

1645

1646 line._kind = LineKind.NestedPhaseEnd

1647 line = yield line

1648

1649 if self._TIME is not None: 1649 ↛ 1661line 1649 didn't jump to line 1661 because the condition on line 1649 was always true

1650 while True:

1651 if line.StartsWith(self._TIME):

1652 line._kind = LineKind.PhaseTime

1653 break

1654 elif isinstance(line, VivadoMessage): 1654 ↛ 1655line 1654 didn't jump to line 1655 because the condition on line 1654 was never true

1655 self._AddMessage(line)

1656

1657 line = yield line

1658

1659 line = yield line

1660

1661 return line

1662

1663 def Generator(self, line: Line) -> Generator[Line, Line, Line]:

1664 line = yield from self._NestedPhaseStart(line)

1665

1666 FINISH = self._FINISH.format(phaseIndex=self._phaseIndex)

1667

1668 while True:

1669 if line._kind is LineKind.Empty: 1669 ↛ 1670line 1669 didn't jump to line 1670 because the condition on line 1669 was never true

1670 line = yield line

1671 continue

1672 elif isinstance(line, VivadoMessage):

1673 self._AddMessage(line)

1674 elif line.StartsWith(FINISH): 1674 ↛ 1677line 1674 didn't jump to line 1677 because the condition on line 1674 was always true

1675 break

1676

1677 line = yield line

1678

1679 nextLine = yield from self._NestedPhaseFinish(line)

1680 return nextLine

1681

1682 def __str__(self) -> str:

1683 return f"{self.__class__.__name__}: {self._START.pattern}"