Wait, poll, predicate ===================== This page is the test-author's polling toolkit: how to block until hardware reaches a state, how to express that state as a value comparison or a predicate, how to sample a register many times for statistical checks, and what happens when the wait times out. Every register and field exposes the same small surface; the goal is that the most common test idiom — "wait until this bit is set" — is one method call, not a hand-rolled loop. Why this matters ---------------- From the design sketch: ``wait_for`` is the single most common test idiom; it deserves dedicated ergonomics. A naive polling loop in user code is easy to get subtly wrong: missing a timeout, never sleeping, never logging the last value seen, never back-pressuring a busy bus. The methods on this page handle all of that uniformly, so test code reads as a statement of intent ("wait for tx_ready") and not as bus plumbing. The contract: - **Default-bounded.** Every wait takes a ``timeout=`` and raises a descriptive error if it expires. - **Returns the value seen.** A successful wait returns the matching read so callers don't immediately read again. - **Sampling is built in.** ``read(n=...)`` and ``histogram(n=...)`` cover debouncing and glitch-detection without a hand-rolled loop. - **Sync first, async parallel.** Every wait has an ``await_for``/``aio*`` dual usable from ``asyncio`` code. ``wait_for`` on a single field ------------------------------ The shortest path: wait until a field equals a value. .. code-block:: python # Block until tx_ready becomes True, or raise after 1.0 s. soc.uart.status.tx_ready.wait_for(True, timeout=1.0) # Tune the polling cadence and add jitter to avoid lock-step # interference with hardware events. soc.uart.status.tx_ready.wait_for( True, timeout=1.0, period=0.001, jitter=True, ) Arguments: - ``value`` (positional): the target. For a 1-bit field, ``True`` / ``False``. For an enum field, the enum member (``BaudRate.BAUD_115200``) or its int value. Type-checked against the field's encoding. - ``timeout`` (seconds): hard upper bound on the wait. Required in practice; the implementation raises if you call ``wait_for`` with no timeout to keep test runs from hanging. - ``period`` (seconds, default master-dependent): nominal time between polls. Smaller is more responsive, larger is gentler on the bus. - ``jitter`` (bool, default ``False``): perturb each sleep by a small random factor. Useful when the polled signal is itself periodic and a fixed period could alias. On success, ``wait_for`` returns the matching ``FieldValue`` (the value the field had on the read that satisfied the comparison). On timeout, it raises ``WaitTimeoutError`` (see below). .. note:: ``wait_for`` is robust against exotic equality comparators. If the wrapped value's ``==`` raises (for example, NumPy ``ndarray``'s vector ``==`` when shapes don't broadcast, or a third-party value wrapper that throws on incompatible types), the polling loop treats that comparison as a no-match and keeps polling instead of crashing the test. This matters when callers wrap fields with custom equality behaviour. ``wait_until`` on a register predicate -------------------------------------- When the wait condition involves more than one field of the same register — "tx_ready set **and** error clear" — use ``wait_until`` on the register, not ``wait_for`` on a field. The predicate receives a fresh ``RegisterValue`` per poll, so it sees a coherent snapshot for that one bus read: .. code-block:: python # Each poll is one bus read; the predicate sees a coherent register value. soc.uart.status.wait_until( lambda v: v.tx_ready and not v.error, timeout=1.0, ) The predicate's argument is a ``RegisterValue``; field accessors (``v.tx_ready``, ``v.error``) return the same typed values you'd get from ``soc.uart.status.read()``. Returning truthy ends the wait; returning falsy re-polls. ``wait_until`` accepts the same ``timeout``, ``period``, and ``jitter`` arguments as ``wait_for``. On success it returns the ``RegisterValue`` that satisfied the predicate. Use ``wait_for`` when the condition is "field equals value"; use ``wait_until`` when the condition is anything more complex, even if it's still over a single field. ``wait_until`` is also the right tool for inequality checks (``v.count > 16``) and for waits that depend on external state (``lambda v: v.tx_ready and feature_enabled``). IRQ shortcut ------------ For interrupt sources, the polling pattern is so common that the :doc:`interrupts` group exposes its own ``wait``: .. code-block:: python # Block until tx_done is pending, or raise after 1.0 s. soc.uart.interrupts.tx_done.wait(timeout=1.0) This is the same as ``soc.uart.status.tx_done_int.wait_for(True, timeout=1.0)`` for chips where the interrupt state register is the natural place to wait, but it reads better and respects the per-RDL clear/acknowledge semantics. See :doc:`interrupts` for the full interrupt surface (``enable``, ``clear``, ``fire``, group operations). Sampling -------- Two patterns come up so often that they live next to ``wait_for``: capturing N reads of the same register, and bucketing them into a histogram. .. code-block:: python # 100 fresh bus reads of soc.adc.sample, returned as a NumPy array. samples = soc.adc.sample.read(n=100) samples.shape # (100,) samples.mean() # 1000 reads, bucketed by value. from collections import Counter hist = soc.adc.sample.histogram(n=1000) hist.most_common(5) A note on shape polymorphism: ``soc.adc.sample.read()`` (no ``n``) returns a single decoded value, the same shape as any other field read. Passing ``n=...`` switches to a NumPy array of length ``n``. The cutover is the keyword argument, not the operation. ``read(n=...)`` is also valid on register and memory views; the array's dtype matches the register width or the memory's word width. Sampling is intentionally separate from ``wait_for``. Use ``wait_for`` to synchronize ("I want a known state before I read"); use ``read(n=...)`` to characterize ("I want N samples of whatever the hardware is doing"). Return types ~~~~~~~~~~~~ The two sampling helpers return distinct, freshly allocated containers: - ``sample(field, n=N)`` (and the equivalent ``read(n=N)``) returns a :class:`numpy.ndarray`. The dtype is whatever NumPy infers from the readings — for register/field reads that decode to Python ``int`` this lands on a platform-appropriate integer width (typically ``int64``/``uint64`` on 64-bit hosts). Callers that need a specific dtype should cast explicitly. The array is freshly allocated per call; it is not aliased to any internal buffer, so mutation of the returned array is safe. - ``histogram(field, n=N)`` returns a :class:`collections.Counter` keyed by value. ``histogram`` deliberately avoids materialising an intermediate ndarray so it works with read payloads that are unhashable as a NumPy dtype but hashable as Python objects (notably enum members and other rich field values). .. code-block:: python import numpy as np from collections import Counter samples = soc.adc.sample.read(n=100) assert isinstance(samples, np.ndarray) assert samples.shape == (100,) hist = soc.adc.sample.histogram(n=1000) assert isinstance(hist, Counter) hist.most_common(5) Async equivalents ----------------- Every blocking wait has a coroutine dual. The methods are renamed with an ``await_`` or ``aio`` prefix so the call site reads correctly under ``await``: .. code-block:: python # asyncio test await soc.uart.status.tx_ready.await_for(True, timeout=1.0) await soc.uart.status.await_until( lambda v: v.tx_ready and not v.error, timeout=1.0, ) await soc.uart.interrupts.tx_done.aiowait(timeout=1.0) The argument shapes match their sync counterparts. Returns and exceptions are the same. Use the async forms whenever you're already inside an ``async def`` test or a notebook running an event loop, so a slow hardware response doesn't block the loop. The complementary surface (``aread`` / ``awrite`` / ``amodify`` on every node) is described in the bus & masters documentation; everything on this page about timeouts, predicates, and sampling applies identically there. .. note:: **The async dual is not natively non-blocking.** There is no native async master transport at this layer. Inside ``await_for``, ``await_until``, and ``aiowait`` the polling loop calls the sync ``node.read()`` directly; only the inter-poll sleep is an ``await asyncio.sleep`` yield point. The bus read therefore blocks the running event loop for its full duration. This is fine for in-process mocks and short transactions, but if your master has long ``read()`` latency you should not assume the loop stays responsive while a wait is in flight — do not expect true non-blocking I/O from these helpers. Want true offload? Wrap the underlying transport in :class:`peakrdl_pybind11.runtime.async_session.AsyncSession`, which uses :func:`asyncio.run_in_executor` against a :class:`concurrent.futures.ThreadPoolExecutor` to push every bus call to a worker thread. ``await_for``/``await_until``/``aiowait`` themselves do not go through that path — they call the sync node's ``read()`` directly. .. note:: **The loop= kwarg is accepted but not forwarded.** ``await_for``, ``await_until``, and ``aiowait`` accept a ``loop=`` keyword for API symmetry with older asyncio code, but the value is *ignored*: it is not passed to :func:`asyncio.sleep`, because the explicit-loop parameter was deprecated in Python 3.8 and removed in 3.10. Pass it if you like for backwards source compatibility; do not rely on it to bind the wait to a non-default event loop. .. code-block:: python # Both forms are valid; the loop= argument is accepted but ignored. await soc.uart.status.tx_ready.await_for(True, timeout=1.0) await soc.uart.status.tx_ready.await_for( True, timeout=1.0, loop=asyncio.get_running_loop() ) Timeout error ------------- On timeout, every wait raises ``WaitTimeoutError`` with a descriptive message that names the wait target, the expected condition, the last value seen, and how long the wait ran: .. code-block:: text WaitTimeoutError: soc.uart.status.tx_ready did not reach True within 1.000s last value seen : False expected : True polled : 1024 reads, period=0.001s For predicate waits, the message includes a short repr of the failing ``RegisterValue`` so post-mortem doesn't require a re-run: .. code-block:: text WaitTimeoutError: soc.uart.status predicate not satisfied within 1.000s last value seen : UartStatus(0x00000004) tx_ready=0 error=1 polled : 1024 reads, period=0.001s For deeper post-mortem, every wait accepts a ``capture=True`` flag that attaches the full list of sampled ``RegisterValue`` (or ``FieldValue``) objects to the exception under ``.samples``: .. code-block:: python try: soc.uart.status.wait_until( lambda v: v.tx_ready and not v.error, timeout=1.0, capture=True, ) except WaitTimeoutError as e: # Replay the trace for the bug report. for v in e.samples: print(v.hex(), v.error, v.tx_ready) ``capture`` is opt-in because the sample list can grow long for tight polling loops; the default error message is descriptive enough for most failures. Which ``WaitTimeoutError``? ~~~~~~~~~~~~~~~~~~~~~~~~~~~ Two classes share the name: - :class:`peakrdl_pybind11.errors.WaitTimeoutError` — the one this module raises. It is the user-facing class re-exported at the top-level ``peakrdl_pybind11.errors`` path. It inherits from both :class:`PeakRDLError ` and :class:`TimeoutError`. - :class:`peakrdl_pybind11.runtime.errors.WaitTimeoutError` — the entry in the lower-level error taxonomy, with a different positional argument signature. Some lower-level code paths still raise this one; it inherits only from :class:`TimeoutError`. Both classes inherit from :class:`TimeoutError`, so a single ``except TimeoutError`` catches either. If you want to catch every PeakRDL-pybind11 error in one clause, catch :class:`PeakRDLError `; that catches the wait-poll variant but **not** the runtime-only one (the runtime one does not inherit from ``PeakRDLError``). When you really need to distinguish which class was raised, import the user-facing one explicitly: .. code-block:: python from peakrdl_pybind11.errors import WaitTimeoutError try: soc.uart.status.tx_ready.wait_for(True, timeout=1.0) except WaitTimeoutError as e: # e.path, e.expected, e.last_seen, e.timeout, e.polls are all set. ... See :doc:`errors` for the full typed-error taxonomy and the rules for catching errors broadly versus narrowly. See also -------- - :doc:`interrupts` — interrupt sources, group operations, and the ``wait``/``aiowait`` shortcuts. - :doc:`observers` — read/write hooks, including how to capture sampled values from a wait without ``capture=True`` (an observer sees every read the wait performs, with no opt-in needed).