Spin-poll ``iface`` for any of ``patterns`` until ``timeout`` elapses. Args: iface: SerialInterface to read lines from. patterns: Substrings to look for in incoming serial lines. Match is by simple ``in`` containment (not regex) so callers can pass the li
(
iface: "SerialInterface",
patterns: Iterable[str],
timeout: float,
spin_interval: float = 0.05,
on_line: "Callable[[str], None] | None" = None,
)
| 52 | |
| 53 | |
| 54 | async def wait_for_signal( |
| 55 | iface: "SerialInterface", |
| 56 | patterns: Iterable[str], |
| 57 | timeout: float, |
| 58 | spin_interval: float = 0.05, |
| 59 | on_line: "Callable[[str], None] | None" = None, |
| 60 | ) -> tuple[str | None, list[str]]: |
| 61 | """Spin-poll ``iface`` for any of ``patterns`` until ``timeout`` elapses. |
| 62 | |
| 63 | Args: |
| 64 | iface: SerialInterface to read lines from. |
| 65 | patterns: Substrings to look for in incoming serial lines. Match is by |
| 66 | simple ``in`` containment (not regex) so callers can pass the |
| 67 | literal boot banner text. |
| 68 | timeout: Maximum wall-clock seconds to wait. Acts as the fallback when |
| 69 | the expected signal never arrives: the helper returns ``None`` |
| 70 | without raising so slow devices still work. |
| 71 | spin_interval: Per-iteration read timeout / polling cadence (seconds). |
| 72 | The helper calls ``iface.read_lines(timeout=spin_interval)`` each |
| 73 | tick so it can exit promptly once a matching line appears. |
| 74 | on_line: Optional callback invoked for every non-matching line that |
| 75 | was drained. Useful for mirroring verbose boot output. |
| 76 | |
| 77 | Returns: |
| 78 | Tuple of (matched_pattern, drained_lines). ``matched_pattern`` is the |
| 79 | first ``patterns`` element found, or ``None`` on timeout. ``drained_lines`` |
| 80 | contains every line observed during the wait, including the matching |
| 81 | one, so callers that previously relied on ``drain_boot_output()`` still |
| 82 | see the boot banner. |
| 83 | """ |
| 84 | patterns_list = list(patterns) |
| 85 | drained: list[str] = [] |
| 86 | # Short-circuit: nothing to match means caller expects us to just drain |
| 87 | # for the full duration. We still poll so we can yield to the event loop. |
| 88 | if not patterns_list: |
| 89 | deadline = time.monotonic() + timeout |
| 90 | while time.monotonic() < deadline: |
| 91 | try: |
| 92 | async for line in iface.read_lines(timeout=spin_interval): |
| 93 | drained.append(line) |
| 94 | if on_line is not None: |
| 95 | on_line(line) |
| 96 | if time.monotonic() >= deadline: |
| 97 | break |
| 98 | except KeyboardInterrupt as ki: |
| 99 | from ci.util.global_interrupt_handler import handle_keyboard_interrupt |
| 100 | |
| 101 | handle_keyboard_interrupt(ki) |
| 102 | raise |
| 103 | except Exception: |
| 104 | # Read timeouts are normal. |
| 105 | pass |
| 106 | await asyncio.sleep(0) |
| 107 | return (None, drained) |
| 108 | |
| 109 | deadline = time.monotonic() + timeout |
| 110 | while time.monotonic() < deadline: |
| 111 | try: |