wardviaene · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
@@ -301,7 +301,7 @@ the pod's second UART, so `uart_open` keeps the console. Needs gateware v48 (`gp
 | Tool | Purpose | Parameters |
 | --- | --- | --- |
 | `capture_adc` | Capture the ADC: calibrated statistics, dominant frequency and a min/max envelope; `clipped_samples` counts samples past the range's full scale. `oversample` converts at 400 kS/s and averages down to `sample_rate_hz` for lower noise, about 5 s at most. | `samples=4096`, `sample_rate_hz`, `source`, `points=200`, `trigger_la`, `trigger_edge=rising`, `trigger_timeout=10`, `range`, `oversample` |
-| `capture_la` | Capture all 14 LA channels: levels, edge count, first edge and frequency per channel. | `samples=4096`, `sample_rate_hz`, `stop_dac_after`, `trigger_la`, `trigger_edge=rising`, `trigger_timeout=10` |
+| `capture_la` | Capture all 14 LA channels: levels, edge count, first edge and frequency per channel. Up to 48 MS/s on gateware v50+. | `samples=4096`, `sample_rate_hz`, `stop_dac_after`, `trigger_la`, `trigger_edge=rising`, `trigger_timeout=10`, `trigger_pre_samples=0` |
 | `capture_correlated` | ADC and LA from one hardware trigger, on one timebase. | `adc_samples=4096`, `adc_sample_rate_hz`, `la_samples=4096`, `la_sample_rate_hz`, `stop_dac_after`, `points=200`, `trigger_la`, `trigger_edge=rising`, `trigger_timeout=10`, `source` |
 | `decode_la` | Decode I2C, UART or SPI from the last LA capture (or a new one). | **`protocol`**, `sda`, `scl`, `rx`, `baud`, `sclk`, `mosi`, `miso`, `cs`, `mode=0`, `capture=last`, `samples=16384`, `sample_rate_hz=1000000`, `max_items=200` |
 | `la_timing` | Edge timestamps, pulse widths, frequency, duty cycle and channel-to-channel delay from the last capture. | **`la`**, `edge=rising`, `to_la`, `to_edge=rising`, `after=0`, `max_edges=100` |

@@ -526,6 +526,11 @@ class LaCaptureResult(BaseModel):
     trigger: Optional[str] = Field(None, description=(
         "The trigger that started the capture, e.g. 'LA9 rising'; null for a free-running capture. "
         "t = 0 is the trigger moment."))
+    pre_samples: int = Field(0, description=(
+        "Samples before the trigger (trigger_pre_samples): the trigger is at this sample index."))
+    truncated: Optional[str] = Field(None, description=(
+        "Above 12 MS/s only: why the capture stopped early ('busy': the signals changed too often, "
+        "'memory'), so it holds fewer samples than asked; null when complete."))
 
 
 class PulseStats(BaseModel):

@@ -286,9 +286,17 @@ def _channel(pod: Any, value: Any, key: str) -> int:
     return int(la)
 
 
-def _trigger(la: Any, edge: str) -> Optional[Trigger]:
+def _trigger(la: Any, edge: str, pre_samples: int = 0) -> Optional[Trigger]:
     """A :class:`Trigger` for the capture tools (the SDK resolves wiring names)."""
-    return None if la is None else Trigger(la, edge)  # type: ignore[arg-type]
+    if la is None:
+        return None
+    if not pre_samples:                       # an SDK older than pre_samples takes the two arguments
+        return Trigger(la, edge)  # type: ignore[arg-type]
+    try:
+        return Trigger(la, edge, pre_samples)  # type: ignore[arg-type,call-arg]
+    except TypeError:
+        raise ValueError("trigger_pre_samples needs embeddedci with Trigger(pre_samples=...); "
+                         "upgrade the embeddedci package") from None
 
 
 # -- connection ---------------------------------------------------------------------
@@ -1514,17 +1522,23 @@ async def capture_la(
     trigger_la: TriggerLa = None,
     trigger_edge: Annotated[TriggerEdge, Field(description="What starts the capture on trigger_la.")] = "rising",
     trigger_timeout: TriggerTimeoutS = 10.0,
+    trigger_pre_samples: Annotated[int, Field(ge=0, le=16_777_215, description=(
+        "With trigger_la: also return up to this many samples from before the trigger "
+        "(gateware v50+)."))] = 0,
     ctx: Context = None,  # type: ignore[assignment]
 ) -> m.LaCaptureResult:
     """Capture all 14 LA channels and summarise each: levels, edge count, first edge, estimated frequency.
 
     With trigger_la the capture starts on that edge or level instead of immediately, so a short
     event can be caught at a high sample rate. The capture is kept for decode_la and la_timing.
+    sample_rate_hz goes up to 12 MS/s, or 48 MS/s on gateware v50+ (48, 24 or 16 MS/s); above
+    12 MS/s a signal changing over ~6 million times per second for long stops early (truncated);
+    retry at 12 MS/s.
     """
     def op() -> m.LaCaptureResult:
         la = SESSION.require().capture_la(samples, sample_rate_hz=sample_rate_hz,
                                           stop_dac_after=stop_dac_after,
-                                          trigger=_trigger(trigger_la, trigger_edge),
+                                          trigger=_trigger(trigger_la, trigger_edge, trigger_pre_samples),
                                           trigger_timeout=trigger_timeout)
         SESSION.last_la = la
         return la_summary(la)

@@ -97,7 +97,10 @@ def la_summary(la: LaCapture) -> m.LaCaptureResult:
             est_frequency_hz=round(edges / 2 / la.duration, 3) if (edges >= 2 and la.duration > 0) else None,
         ))
     return m.LaCaptureResult(samples=len(la), sample_rate_hz=rate, duration=la.duration,
-                             channels=channels, trigger=trigger_text(la.trigger))
+                             channels=channels, trigger=trigger_text(la.trigger),
+                             # getattr: an SDK older than these fields (the MCP's minimum pin)
+                             pre_samples=getattr(la, "pre_samples", 0),
+                             truncated=getattr(la, "truncated", "") or None)
 
 
 # -- wiring, pins and power ------------------------------------------------------------

@@ -846,6 +846,14 @@
           "description": "Wait for an edge or level on this LA channel (a number, or a wiring-profile name) before sampling, so t = 0 is that moment. Omit for a free-running capture. Needs the capture_trigger capability.",
           "title": "Trigger La"
         },
+        "trigger_pre_samples": {
+          "default": 0,
+          "description": "With trigger_la: also return up to this many samples from before the trigger (gateware v50+).",
+          "maximum": 16777215,
+          "minimum": 0,
+          "title": "Trigger Pre Samples",
+          "type": "integer"
+        },
         "trigger_timeout": {
           "default": 10.0,
           "description": "Seconds to wait for the trigger before the capture is abandoned with a TriggerTimeout.",

@@ -1263,6 +1263,21 @@ to 600). When the condition never happens the pod aborts the capture and raises
 (`.la`, `.edge`). The trigger channel can have any function — triggers observe. Triggers need gateware
 v35+ (`Capabilities.capture_trigger`).
 
+`capture_la` can also return what happened before the trigger: `Trigger(..., pre_samples=N)` (gateware
+v50+, any rate up to 48 MS/s). The capture's `pre_samples` says how many it holds (fewer when the
+capture was armed only shortly before the trigger, or the signals changed very often before it), and
+the trigger is at that sample index:
+
+```python
+la = bp.capture_la(20_000, sample_rate_hz=24_000_000,
+                   trigger=Trigger("CS", "falling", pre_samples=2_000))
+t0 = la.pre_samples / la.sample_rate_hz                          # the trigger, in seconds
+```
+
+Above 12 MS/s (gateware v50+) the pod records only the changes, at up to about 6 million per second
+for all channels together. A signal that changes faster for long ends such a capture early: it then
+holds the samples up to that point and `truncated` says why (`"busy"` or `"memory"`).
+
 ## DAC: generator, replay and faults
 
 Every DAC output keeps running until stopped. The returned `DacHandle` / `ReplayHandle` is a

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import base64
+import logging
 import math
 import sys
 from array import array
@@ -22,6 +23,8 @@
 from .errors import BenchPodError
 from .results import Capture, CorrelatedCapture, LaCapture, Trigger
 
+_log = logging.getLogger(__name__)
+
 #: The firmware's single-shot ``capture`` reads into a fixed buffer of this many samples; a larger
 #: ADC capture streams from PSRAM through ``capture_dual`` instead.
 ADC_SHALLOW_MAX_SAMPLES = 32768
@@ -45,6 +48,8 @@ def _apply_trigger(req: dict, trigger: Optional[Trigger], timeout: float) -> Non
     if not 0 < timeout <= MAX_TRIGGER_TIMEOUT:
         raise ValueError(f"trigger_timeout must be > 0 and <= {MAX_TRIGGER_TIMEOUT:g} seconds, got {timeout!r}")
     req["trigger"] = {"la": trigger.la, "edge": trigger.edge}
+    if trigger.pre_samples:
+        req["trigger"]["pre_samples"] = int(trigger.pre_samples)
     req["trigger_timeout_ms"] = max(1, int(round(timeout * 1000)))
 
 
@@ -232,9 +237,15 @@ def capture_la(transport: Any, *, samples: int = 4096, sample_rate_hz: Optional[
     edges: List[List[int]] = []
     upto = 0
     rate = 0.0
+    pre = 0
+    truncated = ""
     for chunk in _stream_or_samples(transport, req):
         if chunk.get("la_rate_hz"):
             rate = float(chunk["la_rate_hz"])
+        if chunk.get("pre_samples"):
+            pre = int(chunk["pre_samples"])
+        if chunk.get("truncated"):
+            truncated = str(chunk["truncated"])
         # The firmware answers la_capture RUN-LENGTH ENCODED — {"la":true,"la_edges":[[i,word],…],
         # "la_upto":N} — exactly as it does for capture_dual, and sends no "data" array at all. RLE
         # wins when present; dense "data" remains the fallback for the server-relayed shape.
@@ -247,7 +258,10 @@ def capture_la(transport: Any, *, samples: int = 4096, sample_rate_hz: Optional[
         if isinstance(data, list):
             dense.extend(int(x) for x in data)
     words = _expand_la_edges(edges, upto or int(samples)) if (edges or upto) else dense
-    return LaCapture(words=words, sample_rate_hz=_rate_hz(rate, sample_rate_hz), trigger=trigger)
+    if truncated:
+        _log.warning("capture_la stopped early (%s): %d of %d samples", truncated, len(words), samples)
+    return LaCapture(words=words, sample_rate_hz=_rate_hz(rate, sample_rate_hz), trigger=trigger,
+                     pre_samples=pre, truncated=truncated)
 
 
 def _expand_la_edges(edges: List[List[int]], upto: int) -> List[int]:

@@ -1772,6 +1772,14 @@ def capture_la(self, samples: int = 4096, *, sample_rate_hz: Optional[float] = N
                    trigger_timeout: float = 10.0) -> LaCapture:
         """Capture raw 14-channel logic-analyzer words and return a :class:`LaCapture`.
 
+        ``sample_rate_hz`` goes up to 12 MS/s, or 48 MS/s on gateware v50 and later
+        (:attr:`Capabilities` ``la_max_rate_hz``): above 12 MS/s the pod runs 48, 24 or 16 MS/s and
+        records only the changes, so a signal changing more than about 6 million times per second
+        (all channels together) for longer than its buffer lasts stops early: the capture then holds
+        the samples up to there and :attr:`LaCapture.truncated` says why. ``LaCapture.sample_rate_hz``
+        is always the rate the pod ran. ``Trigger(..., pre_samples=N)`` also returns samples from
+        before the trigger (gateware v50+, any rate up to 48 MS/s).
+
         ``stop_dac_after`` (seconds) cuts a concurrently-running DAC that far into the capture,
         sample-precise from the capture's hardware t0 (gateware >= v21). ``trigger`` (a
         :class:`Trigger`, gateware >= v35) makes t0 an edge or level on an LA channel —
@@ -1824,7 +1832,7 @@ def _resolve_trigger(self, trigger: Optional[Trigger]) -> Optional[Trigger]:
             raise ValueError(f"trigger must be a Trigger, e.g. Trigger(9, 'rising'), got {trigger!r}")
         self._require_capability("capture_trigger", "capture triggers")
         if isinstance(trigger.la, str):
-            return Trigger(self._resolve_la(trigger.la, "trigger")[0], trigger.edge)
+            return Trigger(self._resolve_la(trigger.la, "trigger")[0], trigger.edge, trigger.pre_samples)
         return trigger
 
     def decode(self, source: Union[LaCapture, Sequence[int]], protocol: DecodeProtocol = "i2c", *,

@@ -31,17 +31,24 @@ class Trigger:
     is the trigger moment, and a DAC co-trigger or ``stop_dac_after`` counts from it::
 
         la = bp.capture_la(100_000, sample_rate_hz=1_000_000, trigger=Trigger(9, "rising"))
+
+    ``pre_samples`` (``capture_la`` only, gateware v50+) also returns up to that many samples from
+    before the trigger: the capture's :attr:`LaCapture.pre_samples` says how many it holds, and the
+    trigger is at that index.
     """
 
     la: Union[int, str]
     edge: TriggerEdge = "rising"
+    pre_samples: int = 0
 
     def __post_init__(self) -> None:
         if self.edge not in TRIGGER_EDGES:
             raise ValueError(f"trigger edge must be one of {', '.join(TRIGGER_EDGES)}, got {self.edge!r}")
         if isinstance(self.la, bool) or not (
                 isinstance(self.la, str) or (isinstance(self.la, int) and 1 <= self.la <= 14)):
             raise ValueError(f"trigger la must be an LA channel 1-14 or a wiring name, got {self.la!r}")
+        if isinstance(self.pre_samples, bool) or not isinstance(self.pre_samples, int) or self.pre_samples < 0:
+            raise ValueError(f"trigger pre_samples must be an integer >= 0, got {self.pre_samples!r}")
 
 
 def _check_edge(edge: str, name: str = "edge") -> None:
@@ -241,6 +248,11 @@ class LaCapture:
     channels: int = 14
     #: The trigger that started the capture (t = 0 is its moment), or ``None``.
     trigger: Optional[Trigger] = None
+    #: Samples before the trigger (``Trigger(pre_samples=...)``): the trigger is at this index.
+    pre_samples: int = 0
+    #: ``""``, or why a capture above 12 MS/s stopped early and holds fewer samples than asked:
+    #: ``"busy"`` (the signals changed faster than the pod can store) or ``"memory"``.
+    truncated: str = ""
 
     def __len__(self) -> int:
         return len(self.words)

@@ -566,7 +566,9 @@
       "words",
       "sample_rate_hz",
       "channels",
-      "trigger"
+      "trigger",
+      "pre_samples",
+      "truncated"
     ],
     "channel": "(self, la: 'int') -> 'List[int]'",
     "decode": "(self, protocol: 'str' = 'i2c', **channels)",
@@ -916,7 +918,8 @@
   "Trigger": {
     "__fields__": [
       "la",
-      "edge"
+      "edge",
+      "pre_samples"
     ]
   },
   "TriggerEdge": "Literal['rising', 'falling', 'high', 'low']",

@@ -14,6 +14,8 @@
 
 import pytest
 
+from embeddedci.benchpod import BenchPod, Trigger
+
 from embeddedci.benchpod import (
     BenchPodError,
     CanReadResult,
@@ -686,6 +688,70 @@ def test_la_reports_the_rate_it_achieved(pod, bench, samples, rate_hz):
         time.sleep(max(0.0, t0 + steps * 2 * delay + 0.1 - time.monotonic()))
 
 
+# Above 12 MS/s (gateware v50+) the pod records the changes at 48 MS/s and replies in the same form:
+# FPGA-timed 4 us pulses must measure exactly 4 us in samples at 48, 24 and 16 MS/s.
+def _needs_v50(pod):
+    if int(pod.status().get("gateware") or 0) < 50:
+        pytest.skip("needs gateware v50 (LA above 12 MS/s)")
+
+
+@pytest.mark.parametrize("rate_hz,width", [(48_000_000, 192), (24_000_000, 96), (16_000_000, 64)])
+def test_la_above_12_msps_reads_exact_pulse_widths(pod, bench, rate_hz, width):
+    _needs_v50(pod)
+    ch = bench.free_la[0]
+    delay = 4e-6
+    pod.la_step(ch, steps=20_000, delay=delay)                # 160 ms of 4 us / 4 us
+    try:
+        la = pod.capture_la(100_000, sample_rate_hz=rate_hz)
+        assert la.sample_rate_hz == rate_hz and len(la) == 100_000 and not la.truncated
+        bits = la.channel(ch)
+        runs, n = [], 1
+        for a, b in zip(bits, bits[1:]):
+            if a == b:
+                n += 1
+            else:
+                runs.append(n)
+                n = 1
+        runs = runs[1:]
+        assert len(runs) > 100 and set(runs) == {width}, f"pulse widths {sorted(set(runs))}, want {width}"
+    finally:
+        time.sleep(0.2)
+
+
+def test_la_pre_trigger_returns_the_samples_before_it(pod, bench, benchpod_connection):
+    # The line is held low (an undriven LA pin may read high) and the pulse train starts from a
+    # second connection 100 ms after the arm, so 2000 samples of history exist when it fires.
+    import threading
+
+    _needs_v50(pod)
+    ch = bench.free_la[0]
+    pin = pod.gpio(ch, "output", level=0)
+    errors = []
+
+    def train():
+        try:
+            time.sleep(0.1)
+            with BenchPod(benchpod_connection, la_voltage=None) as other:
+                other.la_step(ch, steps=2_000, delay=4e-6)
+        except Exception as e:                                # pragma: no cover - reported below
+            errors.append(e)
+
+    th = threading.Thread(target=train)
+    th.start()
+    try:
+        la = pod.capture_la(10_000, sample_rate_hz=48_000_000,
+                            trigger=Trigger(ch, "rising", pre_samples=2_000), trigger_timeout=5.0)
+    finally:
+        th.join()
+        time.sleep(0.1)
+        pin.release()
+    assert not errors, errors
+    assert la.pre_samples == 2_000 and len(la) == 12_000
+    bits = la.channel(ch)
+    assert set(bits[:la.pre_samples]) == {0}, "the line was not low before the trigger"
+    assert bits[la.pre_samples] == 1, "the trigger index is not on the rising edge"
+
+
 # The host clock is the independent reference for the capture clock: events sent at known host
 # times land at sample indices, and the fitted slope is the real sample rate. The drift a wrong
 # rate causes grows with the capture length (samples / 24 MHz regardless of rate: ~170 ms for

@@ -9,7 +9,7 @@
 from embeddedci.benchpod import capture as cap_mod
 from embeddedci.benchpod.capabilities import Capabilities
 from embeddedci.benchpod.errors import BenchPodError
-from embeddedci.benchpod.results import Capture, CorrelatedCapture, LaCapture
+from embeddedci.benchpod.results import Capture, CorrelatedCapture, LaCapture, Trigger
 
 
 class FakeCaptureTransport:
@@ -156,6 +156,29 @@ def test_capture_la_expands_rle_frames():
     assert la.words == [1, 1, 1, 0, 0]
 
 
+def test_capture_la_pre_trigger_and_truncated():
+    # gateware v50: the pre-trigger count and an early stop ride the first frame
+    t = FakeCaptureTransport({"la_capture": [
+        {"status": "ok", "bits": 16, "la_rate_hz": 48000000, "pre_samples": 2, "la": True,
+         "la_edges": [[0, 1], [2, 3]], "la_upto": 5, "more": False},
+    ]})
+    la = cap_mod.capture_la(t, samples=3, sample_rate_hz=48e6, trigger=Trigger(3, "falling", pre_samples=2))
+    assert t.sent[0]["trigger"] == {"la": 3, "edge": "falling", "pre_samples": 2}
+    assert la.words == [1, 1, 3, 3, 3] and la.pre_samples == 2 and la.truncated == ""
+    t = FakeCaptureTransport({"la_capture": [
+        {"status": "ok", "bits": 16, "la_rate_hz": 48000000, "truncated": "busy", "la": True,
+         "la_edges": [[0, 1]], "la_upto": 2, "more": False},
+    ]})
+    la = cap_mod.capture_la(t, samples=1000, sample_rate_hz=48e6)
+    assert len(la.words) == 2 and la.truncated == "busy" and la.pre_samples == 0
+    assert "pre_samples" not in t.sent[0].get("trigger", {})
+
+
+def test_trigger_pre_samples_must_be_a_count():
+    with pytest.raises(ValueError):
+        Trigger(3, "rising", pre_samples=-1)
+
+
 def test_capture_correlated_expands_rle_la_edges():
     # ADC dense (2 samples) + LA as RLE transitions covering 5 samples.
     t = FakeCaptureTransport({"capture_dual": [