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

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 

35 

36from pyTooling.Decorators import export, readonly 

37from pyTooling.MetaClasses import ExtendedType 

38from pyTooling.Versioning import YearReleaseVersion 

39from pyTooling.Warning import WarningCollector, Warning, CriticalWarning 

40 

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 

45 

46 

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+)" 

51 

52 

53@export 

54class UndetectedEnd(CriticalWarning): 

55 _line: Line 

56 

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

58 super().__init__(message) 

59 

60 self._line = line 

61 

62 @readonly 

63 def Line(self) -> Line: 

64 return self._line 

65 

66 

67@export 

68class UnknownLine(Warning): 

69 _line: Line 

70 

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

72 super().__init__(message) 

73 

74 self._line = line 

75 

76 @readonly 

77 def Line(self) -> Line: 

78 return self._line 

79 

80 

81@export 

82class UnknownTask(UnknownLine): 

83 pass 

84 

85 

86@export 

87class UnknownSection(UnknownLine): 

88 pass 

89 

90 

91@export 

92class UnknownPhase(UnknownLine): 

93 pass 

94 

95 

96@export 

97class UnknownSubPhase(UnknownLine): 

98 pass 

99 

100 

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]]] 

110 

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 = {} 

119 

120 @readonly 

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

122 return self._toolIDs 

123 

124 @readonly 

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

126 return self._toolNames 

127 

128 @readonly 

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

130 return self._messagesByID 

131 

132 @readonly 

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

134 return self._infoMessages 

135 

136 @readonly 

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

138 return self._warningMessages 

139 

140 @readonly 

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

142 return self._criticalWarningMessages 

143 

144 @readonly 

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

146 return self._errorMessages 

147 

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) 

157 

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]} 

168 

169 

170@export 

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

172 def __init__(self) -> None: 

173 super().__init__() 

174 

175 

176@export 

177class Parser(BaseParser): 

178 _processor: "Processor" 

179 

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

181 super().__init__() 

182 

183 self._processor = processor 

184 

185 @readonly 

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

187 return self._processor 

188 

189 

190@export 

191class Preamble(Parser): 

192 _toolVersion: Nullable[YearReleaseVersion] 

193 _startDatetime: Nullable[datetime] 

194 

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+)""") 

197 

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

199 super().__init__(processor) 

200 

201 self._toolVersion = None 

202 self._startDatetime = None 

203 

204 @readonly 

205 def ToolVersion(self) -> YearReleaseVersion: 

206 return self._toolVersion 

207 

208 @readonly 

209 def StartDatetime(self) -> datetime: 

210 return self._startDatetime 

211 

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 

217 

218 line = yield line 

219 

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 

232 

233 line = yield line 

234 

235 nextLine = yield line 

236 return nextLine 

237 

238 

239@export 

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

241 # _START: ClassVar[str] 

242 # _FINISH: ClassVar[str] 

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

244 

245 _command: "Command" 

246 _duration: float 

247 

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

249 super().__init__() 

250 VivadoMessagesMixin.__init__(self) 

251 

252 self._command = command 

253 

254 @readonly 

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

256 return self._command 

257 

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}.") 

261 

262 line._kind = LineKind.TaskStart 

263 nextLine = yield line 

264 return nextLine 

265 

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}.") 

269 

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 

276 

277 line = yield line 

278 

279 line = yield line 

280 return line 

281 

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

283 line = yield from self._TaskStart(line) 

284 

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 

297 

298 line = yield line 

299 

300 nextLine = yield from self._TaskFinish(line) 

301 return nextLine 

302 

303 def __str__(self) -> str: 

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

305 

306 

307@export 

308class TaskWithSubTasks(Task): 

309 # _START: ClassVar[str] 

310 # _FINISH: ClassVar[str] 

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

312 

313 _PARSERS: ClassVar[Dict[YearReleaseVersion,Tuple[Type["SubTask"], ...]]] = dict() 

314 

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

316 

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

318 super().__init__(command) 

319 

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

321 

322 @readonly 

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

324 return self._subtasks 

325 

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

327 return self._subtasks[key] 

328 

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

330 line = yield from self._TaskStart(line) 

331 

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

333 

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 

356 

357 line = yield line 

358 

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 

364 

365 if isinstance(line, VivadoMessage): 

366 self._AddMessage(line) 

367 

368 try: 

369 line = yield subtask.send(line) 

370 except StopIteration as ex: 

371 activeParsers.remove(parser) 

372 line = ex.value 

373 break 

374 

375@export 

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

377 # _START: ClassVar[str] 

378 # _FINISH: ClassVar[str] 

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

380 

381 _task: TaskWithSubTasks 

382 _duration: float 

383 

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

385 super().__init__() 

386 VivadoMessagesMixin.__init__(self) 

387 

388 self._task = task 

389 

390 @readonly 

391 def Task(self) -> TaskWithSubTasks: 

392 return self._task 

393 

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}.") 

397 

398 line._kind = LineKind.TaskStart 

399 nextLine = yield line 

400 return nextLine 

401 

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}.") 

405 

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 

412 

413 line = yield line 

414 

415 line = yield line 

416 return line 

417 

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

419 line = yield from self._TaskStart(line) 

420 

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 

433 

434 line = yield line 

435 

436 nextLine = yield from self._TaskFinish(line) 

437 return nextLine 

438 

439 def __str__(self) -> str: 

440 return self._NAME 

441 

442 

443@export 

444class TaskWithPhases(Task): 

445 # _START: ClassVar[str] 

446 # _FINISH: ClassVar[str] 

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

448 

449 _PARSERS: ClassVar[Dict[YearReleaseVersion,Tuple[Type["Phase"], ...]]] = tuple() 

450 

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

452 

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

454 super().__init__(command) 

455 

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

457 

458 @readonly 

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

460 return self._phases 

461 

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

463 return self._phases[key] 

464 

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

466 line = yield from self._TaskStart(line) 

467 

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

469 

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 

492 

493 line = yield line 

494 

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) 

498 

499 isFinish = line.StartsWith("Ending") 

500 

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 

514 

515 

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 

522 

523 _task: TaskWithPhases 

524 _phaseIndex: int 

525 _duration: float 

526 

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

528 super().__init__() 

529 VivadoMessagesMixin.__init__(self) 

530 

531 self._task = task 

532 self._phaseIndex = None 

533 

534 @readonly 

535 def Task(self) -> TaskWithPhases: 

536 return self._task 

537 

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}.") 

541 

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

543 

544 line._kind = LineKind.PhaseStart 

545 nextLine = yield line 

546 return nextLine 

547 

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}.") 

552 

553 line._kind = LineKind.PhaseEnd 

554 line = yield line 

555 

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 

561 

562 line = yield line 

563 

564 line = yield line 

565 

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 

571 

572 line = yield line 

573 

574 line = yield line 

575 

576 return line 

577 

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

579 line = yield from self._PhaseStart(line) 

580 

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

582 

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 

591 

592 line = yield line 

593 

594 nextLine = yield from self._PhaseFinish(line) 

595 return nextLine 

596 

597 def __str__(self) -> str: 

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

599 

600 

601@export 

602class PhaseWithChildren(Phase): 

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

604 

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

606 

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

608 super().__init__(task) 

609 

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

611 

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

613 line = yield from self._PhaseStart(line) 

614 

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

616 

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

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

619 

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 

643 

644 line = yield line 

645 

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) 

649 

650 isFinish = False # line.StartsWith(SUBPHASE_PREFIX) 

651 

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 

665 

666 

667@export 

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

669 # _START: ClassVar[str] 

670 # _FINISH: ClassVar[str] 

671 

672 _phase: Phase 

673 _phaseIndex: int 

674 _subPhaseIndex: int 

675 _duration: float 

676 

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

678 super().__init__() 

679 VivadoMessagesMixin.__init__(self) 

680 

681 self._phase = phase 

682 self._phaseIndex = None 

683 self._subPhaseIndex = None 

684 

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}.") 

688 

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

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

691 

692 line._kind = LineKind.SubPhaseStart 

693 nextLine = yield line 

694 return nextLine 

695 

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

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

698 

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}.") 

701 

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 

706 

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 

712 

713 line = yield line 

714 

715 nextLine = yield line 

716 return nextLine 

717 

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

719 line = yield from self._SubPhaseStart(line) 

720 

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

722 

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) 

731 

732 line = yield line 

733 

734 nextLine = yield from self._SubPhaseFinish(line) 

735 return nextLine 

736 

737 def __str__(self) -> str: 

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

739 

740 

741@export 

742class SubPhaseWithChildren(SubPhase): 

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

744 

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

746 super().__init__(phase) 

747 

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

749 

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

751 line = yield from self._SubPhaseStart(line) 

752 

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

754 

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

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

757 

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 

780 

781 line = yield line 

782 

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 

787 

788 if isinstance(line, VivadoMessage): 

789 self._AddMessage(line) 

790 

791 try: 

792 line = yield phase.send(line) 

793 except StopIteration as ex: 

794 activeParsers.remove(parser) 

795 line = ex.value 

796 break 

797 

798 

799@export 

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

801 # _START: ClassVar[str] 

802 # _FINISH: ClassVar[str] 

803 

804 _subphase: SubPhase 

805 _phaseIndex: int 

806 _subPhaseIndex: int 

807 _subSubPhaseIndex: int 

808 _duration: float 

809 

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

811 super().__init__() 

812 VivadoMessagesMixin.__init__(self) 

813 

814 self._subphase = subphase 

815 self._phaseIndex = None 

816 self._subPhaseIndex = None 

817 self._subSubPhaseIndex = None 

818 

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() 

822 

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

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

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

826 

827 line._kind = LineKind.SubSubPhaseStart 

828 nextLine = yield line 

829 return nextLine 

830 

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

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

833 

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() 

836 

837 line._kind = LineKind.SubSubPhaseEnd 

838 line = yield line 

839 

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 

844 

845 line = yield line 

846 

847 nextLine = yield line 

848 return nextLine 

849 

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

851 line = yield from self._SubSubPhaseStart(line) 

852 

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

854 

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) 

863 

864 line = yield line 

865 

866 nextLine = yield from self._SubSubPhaseFinish(line) 

867 return nextLine 

868 

869 def __str__(self) -> str: 

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

871 

872 

873@export 

874class SubSubPhaseWithChildren(SubSubPhase): 

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

876 

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

878 super().__init__(subphase) 

879 

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

881 

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

883 line = yield from self._SubSubPhaseStart(line) 

884 

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

886 

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

888 

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 

908 

909 line = yield line 

910 

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 

915 

916 if isinstance(line, VivadoMessage): 

917 self._AddMessage(line) 

918 

919 try: 

920 line = yield phase.send(line) 

921 except StopIteration as ex: 

922 activeParsers.remove(parser) 

923 line = ex.value 

924 break 

925 

926 

927@export 

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

929 # _START: ClassVar[str] 

930 # _FINISH: ClassVar[str] 

931 

932 _subsubphase: SubSubPhase 

933 _phaseIndex: int 

934 _subPhaseIndex: int 

935 _subSubPhaseIndex: int 

936 _subSubSubPhaseIndex: int 

937 _duration: float 

938 

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

940 super().__init__() 

941 VivadoMessagesMixin.__init__(self) 

942 

943 self._subsubphase = subsubphase 

944 self._phaseIndex = None 

945 self._subPhaseIndex = None 

946 self._subSubPhaseIndex = None 

947 self._subSubSubPhaseIndex = None 

948 

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() 

952 

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

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

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

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

957 

958 line._kind = LineKind.SubSubSubPhaseStart 

959 nextLine = yield line 

960 return nextLine 

961 

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) 

964 

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() 

967 

968 line._kind = LineKind.SubSubSubPhaseEnd 

969 line = yield line 

970 

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 

975 

976 line = yield line 

977 

978 nextLine = yield line 

979 return nextLine 

980 

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

982 line = yield from self._SubSubSubPhaseStart(line) 

983 

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

985 

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) 

994 

995 line = yield line 

996 

997 nextLine = yield from self._SubSubSubPhaseFinish(line) 

998 return nextLine 

999 

1000 def __str__(self) -> str: 

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