diff --git a/litescope/core.py b/litescope/core.py index db7e010..4038d25 100644 --- a/litescope/core.py +++ b/litescope/core.py @@ -67,7 +67,11 @@ def __init__(self, data_width, depth=16): self.specials += MultiReg(done, self.done.status) # Memory and configuration. - mem = stream.AsyncFIFO([("mask", data_width), ("value", data_width)], depth) + # Use a buffered AsyncFIFO so Mask/Value come from registers: the unbuffered FIFO's + # output is the block RAM read port, whose clock-to-output delay would otherwise feed + # the data_width-wide comparator and the consume handshake within a single scope cycle + # and break timing as probes get wide. + mem = stream.AsyncFIFO([("mask", data_width), ("value", data_width)], depth, buffered=True) mem = ClockDomainsRenamer({"write": "sys", "read": "scope"})(mem) self.submodules += mem self.comb += [ @@ -77,20 +81,40 @@ def __init__(self, data_width, depth=16): self.mem_full.status.eq(~mem.sink.ready) ] + # Register Sink Data/Valid so the data_width-wide comparator sits between registers. The + # probe stream is free-running (the Mux always asserts valid), so sampling it + # unconditionally delays Data and Hit together by one cycle and preserves their + # alignment (the trigger position within the capture is unchanged). + sink_d = stream.Endpoint(core_layout(data_width)) + self.sync.scope += [ + sink_d.valid.eq(sink.valid), + sink_d.data.eq(sink.data), + ] + self.comb += sink.ready.eq(1) + # Hit and memory read/flush. - hit = Signal() - flush = WaitTimer(2*depth) - flush = ClockDomainsRenamer("scope")(flush) - self.submodules += flush + # Flush pending terms on the falling edge of enable and stop as soon as the memory is + # empty: a fixed-duration flush window (previous implementation) also ran from reset and + # kept draining after the memory was already empty, silently consuming terms pushed + # shortly after reset or re-arming - the capture then started with an empty term memory, + # i.e. an immediate spurious trigger (easily hit with fast CSR transports). + hit = Signal() + flushing = Signal() + self.sync.scope += [ + If(~enable & enable_d, + flushing.eq(1) + ).Elif(~mem.source.valid, + flushing.eq(0) + ) + ] self.comb += [ - flush.wait.eq(~(~enable & enable_d)), # flush when disabling - hit.eq((sink.data & mem.source.mask) == (mem.source.value & mem.source.mask)), - mem.source.ready.eq((enable & hit) | ~flush.done), + hit.eq((sink_d.data & mem.source.mask) == (mem.source.value & mem.source.mask)), + mem.source.ready.eq((enable & sink_d.valid & hit) | flushing), ] # Output. self.comb += [ - sink.connect(source), + sink_d.connect(source, omit={"ready"}), # Done when all triggers have been consumed. done.eq(~mem.source.valid), source.hit.eq(done) @@ -307,6 +331,13 @@ def __init__(self, data_width, depth): mem_flush = ClockDomainsRenamer("scope")(mem_flush) self.submodules += mem_flush + # Read-path flush: samples from a previous capture/drain can linger in the CDC FIFO and + # read converter; drain them during FLUSH (no new data enters the read path in that + # state) so the upload starts with the new capture's first sample. + flush_read = Signal() + flush_read_sys = Signal() + self.specials += MultiReg(flush_read, flush_read_sys) + # FSM. fsm = FSM(reset_state="IDLE") fsm = ClockDomainsRenamer("scope")(fsm) @@ -323,6 +354,7 @@ def __init__(self, data_width, depth): sink.ready.eq(1), mem_flush.wait.eq(1), mem.source.ready.eq(1), + flush_read.eq(1), If(mem_flush.done, NextState("WAIT") ) @@ -357,7 +389,7 @@ def __init__(self, data_width, depth): self.comb += cdc.source.connect(read_source) self.comb += [ - read_source.ready.eq(self.mem_data.rd_stb | ~self.enable.storage), + read_source.ready.eq(self.mem_data.rd_stb | ~self.enable.storage | flush_read_sys), self.mem_data.status.eq(read_source.data) ] diff --git a/litescope/software/driver/analyzer.py b/litescope/software/driver/analyzer.py index c0d5b98..78a001c 100644 --- a/litescope/software/driver/analyzer.py +++ b/litescope/software/driver/analyzer.py @@ -71,6 +71,10 @@ def __init__(self, regs, name, config_csv=None, debug=False): self.offset = 0 self.length = None + # Configured trigger terms (mask, value); loaded into the gateware on each run() since + # the trigger memory consumes its terms on every capture. + self.trigger_terms = [] + # Disable trigger and storage self.trigger_enable.write(0) self.storage_enable.write(0) @@ -111,24 +115,45 @@ def build(self): if self.name == key[:len(self.name)]: key = key.replace(self.name + "_", "") setattr(self, key, value) - for signals in self.layouts.values(): + # Group-aware signal offsets/masks: triggers must resolve a signal's position in the + # currently selected group (a signal present in several groups can sit at different + # positions in each). + self.signal_offsets = {} + self.signal_masks = {} + for group, signals in self.layouts.items(): value = 1 for name, length in signals: - setattr(self, name + "_o", value) + self.signal_offsets[(group, name)] = value value = value*(2**length) - for signals in self.layouts.values(): value = 0 for name, length in signals: - setattr(self, name + "_m", (2**length-1) << value) + self.signal_masks[(group, name)] = (2**length-1) << value value += length + # Keep the flat _o/_m attributes for compatibility; warn when a signal is + # ambiguous across groups (the flat attribute then reflects the last group only). + ambiguous = set() + for (group, name), offset in self.signal_offsets.items(): + mask = self.signal_masks[(group, name)] + if hasattr(self, name + "_o") and \ + ((getattr(self, name + "_o") != offset) or (getattr(self, name + "_m") != mask)): + ambiguous.add(name) + setattr(self, name + "_o", offset) + setattr(self, name + "_m", mask) + for name in sorted(ambiguous): + print(f"{self.name}: warning: signal '{name}' is present in several groups at " + "different positions; triggers resolve it in the selected group.") + + def _signal_offset(self, name): + return self.signal_offsets.get((self.group, name), getattr(self, name + "_o")) + + def _signal_mask(self, name): + return self.signal_masks.get((self.group, name), getattr(self, name + "_m")) def configure_group(self, value): self.group = value self.mux_value.write(value) def add_trigger(self, value=0, mask=0, cond=None): - if self.trigger_mem_full.read(): - raise ValueError("Trigger memory full, too much conditions") if cond is not None: for k, v in cond.items(): # Check for binary/hexa expressions @@ -145,23 +170,41 @@ def add_trigger(self, value=0, mask=0, cond=None): if c != "x": v |= int(c, 16 if mx is not None else 2 ) m |= 0xf if mx is not None else 0b1 - value |= getattr(self, k + "_o")*v - mask |= getattr(self, k + "_m") & (getattr(self, k + "_o")*m) + value |= self._signal_offset(k)*v + mask |= self._signal_mask(k) & (self._signal_offset(k)*m) # Else convert to int else: - value |= getattr(self, k + "_o")*int(v, 0) - mask |= getattr(self, k + "_m") - self.trigger_mem_mask.write(mask) - self.trigger_mem_value.write(value) - self.trigger_mem_write.write(1) + value |= self._signal_offset(k)*int(v, 0) + mask |= self._signal_mask(k) + self.trigger_terms.append((mask, value)) def add_rising_edge_trigger(self, name): - self.add_trigger(getattr(self, name + "_o")*0, getattr(self, name + "_m")) - self.add_trigger(getattr(self, name + "_o")*1, getattr(self, name + "_m")) + self.add_trigger(self._signal_offset(name)*0, self._signal_mask(name)) + self.add_trigger(self._signal_offset(name)*1, self._signal_mask(name)) def add_falling_edge_trigger(self, name): - self.add_trigger(getattr(self, name + "_o")*1, getattr(self, name + "_m")) - self.add_trigger(getattr(self, name + "_o")*0, getattr(self, name + "_m")) + self.add_trigger(self._signal_offset(name)*1, self._signal_mask(name)) + self.add_trigger(self._signal_offset(name)*0, self._signal_mask(name)) + + def _load_trigger_terms(self, timeout=1.0): + # Disarm the trigger; the gateware flushes any leftover terms (previous capture, + # aborted sequence) on the falling edge of enable. + self.trigger_enable.write(0) + # The flush window is 2*trigger_depth scope cycles; any CSR access takes far longer, + # but poll done (trigger memory empty) so the reload below cannot race the flush. + deadline = time.time() + timeout + for _ in range(2): + while not self.trigger_done.read(): + if time.time() > deadline: + raise TimeoutError("Trigger memory flush timeout") + # (Re-)load the configured terms so a capture can be re-run without reconfiguring: + # the trigger memory consumes its terms on every capture. + for mask, value in self.trigger_terms: + if self.trigger_mem_full.read(): + raise ValueError("Trigger memory full, too much conditions") + self.trigger_mem_mask.write(mask) + self.trigger_mem_value.write(value) + self.trigger_mem_write.write(1) def configure_trigger(self, value=0, mask=0, cond=None): self.add_trigger(value, mask, cond) @@ -193,8 +236,16 @@ def run(self, offset=0, length=None): self.length = length if self.debug: self._log(f"run (offset={offset}, length={length})") + # Disarm the trigger and reload its terms first: a previous capture leaves the trigger + # armed with an empty (fully consumed) term memory, whose hit output is a constant + # level that would otherwise fire the re-armed storage immediately. With the terms + # reloaded, hit stays low until a real match. + self._load_trigger_terms() self.storage_offset.write(offset) self.storage_length.write(length) + # Storage arms on the rising edge of enable: clear it first so run() also re-arms + # after a previous capture on the same driver instance. + self.storage_enable.write(0) self.storage_enable.write(1) self.trigger_enable.write(1) @@ -203,6 +254,7 @@ def clear(self): self.offset = 0 self.length = None self.rle_enabled = False + self.trigger_terms = [] self.trigger_enable.write(0) self.storage_enable.write(0) if hasattr(self, "rle_enable"): @@ -321,11 +373,12 @@ def get_instant_value(self, group, name): self.data = DumpData(self.data_width) self.debug = False self.configure_group(group) + self.trigger_terms = [] self.configure_trigger() self.configure_subsampler(1) self.run(0, 1) self.wait_done() self.upload() - min_idx = log2_int(getattr(self, name + "_o")) - max_idx = min_idx + log2_int((getattr(self, name + "_m") >> min_idx) + 1) + min_idx = log2_int(self._signal_offset(name)) + max_idx = min_idx + log2_int((self._signal_mask(name) >> min_idx) + 1) return self.data[min_idx:max_idx][0] diff --git a/test/test_analyzer.py b/test/test_analyzer.py index 2c1067b..5077961 100644 --- a/test/test_analyzer.py +++ b/test/test_analyzer.py @@ -4,11 +4,14 @@ # Copyright (c) 2017-2026 Florent Kermarrec # SPDX-License-Identifier: BSD-2-Clause +import re import unittest import tempfile from migen import * +from litex.soc.interconnect import csr_bus + from litescope import LiteScopeAnalyzer from litescope.software.dump.common import DumpData @@ -33,8 +36,10 @@ class TestAnalyzer(unittest.TestCase): def test_analyzer(self): def generator(dut): dut.data = [] - # Configure Trigger - yield from dut.analyzer.trigger.mem_value.write(0x0010) + # Configure Trigger (on a counter value comfortably after arming completes; the + # term memory no longer accepts being armed with pending terms silently flushed, + # so the trigger must actually be enabled and reachable). + yield from dut.analyzer.trigger.mem_value.write(0x0400) yield from dut.analyzer.trigger.mem_mask.write(0xffff) yield from dut.analyzer.trigger.mem_write.write(1) @@ -45,6 +50,7 @@ def generator(dut): yield from dut.analyzer.storage.length.write(256) yield from dut.analyzer.storage.offset.write(8) yield from dut.analyzer.storage.enable.write(1) + yield from dut.analyzer.trigger.enable.write(1) yield for i in range(16): yield @@ -64,7 +70,11 @@ def __init__(self): generators = {"sys" : [generator(dut)]} clocks = {"sys": 10, "scope": 10} run_simulation(dut, generators, clocks, vcd_name="sim.vcd") - self.assertEqual(dut.data, [524 + 3*i for i in range(len(dut.data))]) + # Trigger value 0x400 (comfortably after the storage FLUSH window at depth=512) with + # offset=8 and subsampling=3: the capture holds the pre-trigger window followed by the + # match (0x400 at index 6) and the post-trigger samples. + self.assertEqual(dut.data, [1006 + 3*i for i in range(len(dut.data))]) + self.assertEqual(dut.data.index(0x400), 6) def test_analyzer_group_mux(self): def generator(dut): @@ -105,7 +115,10 @@ def __init__(self): generators = {"sys" : [generator(dut)]} clocks = {"sys": 10, "scope": 10} run_simulation(dut, generators, clocks) - self.assertEqual(dut.data, [132 + 3*i for i in range(len(dut.data))]) + # With offset=0 the trigger match is the first captured sample: 0xb0 at index 0 + # (the second group's signal, proving the mux selection), incrementing by 3. + self.assertEqual(dut.data[0], 0xb0) + self.assertEqual(dut.data, [(0xb0 + 3*i) & 0xff for i in range(len(dut.data))]) def test_analyzer_raw_msb_data_without_rle(self): def generator(dut): @@ -358,3 +371,148 @@ def get_name(self, signal): self.assertIn("enum,0,fsm_state,0,IDLE", lines) self.assertIn("enum,0,fsm_state,1,RUN", lines) self.assertIn("enum,0,fsm_state,2,DONE", lines) + + +# Wide trigger through the real multi-word CSR bus path --------------------------------------------- +# +# At data_width > 32 the trigger Mask/Value CSRs are compound (multiple bus words). The tests +# above poke CSRStorage.write() which sets the whole storage atomically, so the word-by-word bus +# path (what csr_builder/RemoteClient perform on hardware) was never exercised, and neither was +# a comparator wider than 32 bits. These tests drive the analyzer exclusively through a +# csr_bus.CSRBank, mirroring the host access pattern (MSB word first, ascending addresses, +# separate arm strobe). + +# The trigger sample's position within the capture (offset=8, subsampling=1) is locked below: +# it must stay stable across trigger-pipeline changes (only absolute capture time may shift). + + +class _WideDUT(Module): + def __init__(self, data_width=128, depth=64): + self.counter = counter = Signal(32) + self.sync += counter.eq(counter + 1) + probe = Signal(data_width) + self.comb += probe.eq(Cat(counter, ~counter, (counter + 0x12345678)[:32], (counter ^ 0x55AA55AA)[:32])) + self.submodules.analyzer = LiteScopeAnalyzer(probe, depth, csr_csv=None) + self.bus = csr_bus.Interface(data_width=32, address_width=14) + self.submodules.bank = csr_bus.CSRBank(self.analyzer.get_csrs(), address=0, bus=self.bus) + + +def _wide_pattern(c): + mask32 = 0xffffffff + return (( c & mask32) | + ((~c & mask32) << 32) | + (((c + 0x12345678) & mask32) << 64) | + (((c ^ 0x55AA55AA) & mask32) << 96)) + + +def _csr_addrs(dut, name): + # Bus word addresses of a (possibly compound) CSR; ascending = MSB word first with the + # default "big" ordering, matching litex's csr_builder host-side decomposition. + return [i for i, c in enumerate(dut.bank.simple_csrs) + if re.fullmatch(re.escape(name) + r"\d*", c.name)] + + +def _bus_csr_write(dut, name, value): + addrs = _csr_addrs(dut, name) + n = len(addrs) + for j, adr in enumerate(addrs): + yield from dut.bus.write(adr, (value >> (32*(n - 1 - j))) & 0xffffffff) + + +def _bus_word_read(dut, adr): + # csr_bus.Interface.read samples dat_r one cycle too early for the bank's registered + # read path; add the settle cycle here. + yield dut.bus.adr.eq(adr) + yield dut.bus.re.eq(1) + yield + yield dut.bus.re.eq(0) + yield + return (yield dut.bus.dat_r) + + +def _bus_csr_read(dut, name): + addrs = _csr_addrs(dut, name) + value = 0 + for adr in addrs: + value = (value << 32) | (yield from _bus_word_read(dut, adr)) + return value + + +def _bus_read_samples(dut, count, subwords): + samples = [] + for _ in range(count): + value = 0 + for j in range(subwords): + sub = (yield from _bus_csr_read(dut, "storage_mem_data")) + value |= sub << (32*j) + samples.append(value) + return samples + + +class TestAnalyzerWideTrigger(unittest.TestCase): + # Locked trigger sample position within the capture for offset=8: the matching sample + # lands at index 6. + EXPECTED_TRIGGER_INDEX = 6 + + def _capture(self, dut, trigger_value, offset=8, length=32, timeout=4096): + full_mask = 2**128 - 1 + yield from _bus_csr_write(dut, "subsampler_value", 0) + yield from _bus_csr_write(dut, "storage_offset", offset) + yield from _bus_csr_write(dut, "storage_length", length) + yield from _bus_csr_write(dut, "storage_enable", 0) + yield from _bus_csr_write(dut, "storage_enable", 1) + yield from _bus_csr_write(dut, "trigger_enable", 0) + yield from _bus_csr_write(dut, "trigger_mem_mask", full_mask) + yield from _bus_csr_write(dut, "trigger_mem_value", trigger_value) + yield from _bus_csr_write(dut, "trigger_mem_write", 1) + yield from _bus_csr_write(dut, "trigger_enable", 1) + # done stays asserted (IDLE) until the enable edge crosses into the scope domain: + # wait for the capture to start before waiting for it to complete. + for _ in range(timeout): + if not (yield from _bus_csr_read(dut, "storage_done")): + break + yield + else: + self.fail("capture did not start") + for _ in range(timeout): + if (yield from _bus_csr_read(dut, "storage_done")): + break + yield + else: + self.fail("capture did not complete") + # After completion the storage FIFO starts migrating samples into the CDC/read + # pipeline, so mem_level only reflects what still sits in the FIFO; the read path + # delivers the full capture. + level = yield from _bus_csr_read(dut, "storage_mem_level") + self.assertGreater(level, 0) + return (yield from _bus_read_samples(dut, length, 4)) + + def test_wide_trigger_via_csr_bus(self): + dut = _WideDUT() + results = {} + + def generator(): + yield + targets = [] + for run in range(2): + # Re-arm on each iteration: terms are consumed per capture and must be + # re-loaded (what the driver's run() does). + target = (yield dut.counter) + 500 + targets.append(target) + results[run] = (yield from self._capture(dut, _wide_pattern(target))) + results["targets"] = targets + + run_simulation(dut, {"sys": [generator()]}, {"sys": 10, "scope": 10}, vcd_name=None) + + for run in range(2): + samples = results[run] + base = samples[0] & 0xffffffff + self.assertEqual(samples, [_wide_pattern(base + i) for i in range(len(samples))], + msg=f"capture {run} is not a consecutive wide pattern") + # The trigger sample sits at the locked position within the capture. + self.assertEqual(samples[self.EXPECTED_TRIGGER_INDEX], + _wide_pattern(results["targets"][run])) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_driver.py b/test/test_driver.py index e435f63..9e7d0cc 100644 --- a/test/test_driver.py +++ b/test/test_driver.py @@ -70,6 +70,7 @@ def make_regs(name="analyzer", mem_level=0, mem_data=None, with_rle=False): "trigger_mem_value": FakeReg(), "trigger_mem_write": FakeReg(), "trigger_enable": FakeReg(), + "trigger_done": FakeReg(1), "subsampler_value": FakeReg(), "storage_offset": FakeReg(), "storage_length": FakeReg(), @@ -203,37 +204,36 @@ def test_conditional_trigger_parsing(self): self.clear_writes(regs) driver.add_trigger(cond={"flag": "1", "state": "0b1x0"}) + self.assertEqual(driver.trigger_terms, [(0xb, 0x9)]) - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x9]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xb]) - self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1]) - - self.clear_writes(regs) driver.add_trigger(cond={"wide": "0xax"}) + self.assertEqual(driver.trigger_terms[1], (0xf0, 0xa0)) - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0xa0]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xf0]) - self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1]) + # Terms are written to the gateware when (re)loaded, not when added. + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, []) + driver._load_trigger_terms() + self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x9, 0xa0]) + self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xb, 0xf0]) + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1, 1]) def test_edge_trigger_helpers(self): driver, regs = self.make_driver() self.clear_writes(regs) driver.add_rising_edge_trigger("flag") - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0, 1]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [1, 1]) + self.assertEqual(driver.trigger_terms, [(1, 0), (1, 1)]) - self.clear_writes(regs) + driver.trigger_terms = [] driver.add_falling_edge_trigger("flag") - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [1, 0]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [1, 1]) + self.assertEqual(driver.trigger_terms, [(1, 1), (1, 0)]) - def test_add_trigger_checks_memory_full(self): + def test_load_checks_memory_full(self): driver, regs = self.make_driver() regs.d["analyzer_trigger_mem_full"].value = 1 + driver.add_trigger(value=1, mask=1) with self.assertRaises(ValueError): - driver.add_trigger(value=1, mask=1) + driver._load_trigger_terms() def test_configure_subsampler_and_run(self): driver, regs = self.make_driver(depth=8, subsampler_width=4) @@ -250,14 +250,56 @@ def test_configure_subsampler_and_run(self): self.assertEqual(regs.d["analyzer_subsampler_value"].writes, [3, 15]) self.assertEqual(regs.d["analyzer_storage_offset"].writes, [2]) self.assertEqual(regs.d["analyzer_storage_length"].writes, [5]) - self.assertEqual(regs.d["analyzer_storage_enable"].writes, [1]) - self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [1]) + # run() re-arms storage (rising edge) and disarms/reloads/rearms the trigger. + self.assertEqual(regs.d["analyzer_storage_enable"].writes, [0, 1]) + self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [0, 1]) with self.assertRaises(AssertionError): driver.run(offset=8) with self.assertRaises(AssertionError): driver.run(length=9) + def test_run_reloads_trigger_terms(self): + driver, regs = self.make_driver() + self.clear_writes(regs) + + driver.add_trigger(value=0x3, mask=0x7) + driver.add_trigger(value=0x1, mask=0x1) + + driver.run(offset=0, length=8) + driver.run(offset=0, length=8) + + # Both terms are loaded on each run so captures can be re-run without reconfiguring. + self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x3, 0x1, 0x3, 0x1]) + self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0x7, 0x1, 0x7, 0x1]) + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1, 1, 1, 1]) + self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [0, 1, 0, 1]) + self.assertEqual(regs.d["analyzer_storage_enable"].writes, [0, 1, 0, 1]) + + def test_multi_group_signal_offsets(self): + # A signal present in several groups can sit at different positions in each; triggers + # must resolve it in the currently selected group. + self.tmpdir = tempfile.TemporaryDirectory() + config_csv = os.path.join(self.tmpdir.name, "analyzer.csv") + with open(config_csv, "w") as f: + f.write("config,None,data_width,8\n") + f.write("config,None,depth,16\n") + f.write("config,None,samplerate,100000000\n") + f.write("signal,0,flag,1\n") + f.write("signal,0,shared,3\n") + f.write("signal,1,shared,3\n") + regs = make_regs() + driver = LiteScopeAnalyzerDriver(regs, "analyzer", config_csv=config_csv) + + # Group 0: shared sits above flag; group 1: shared sits at bit 0. + driver.configure_group(0) + driver.add_trigger(cond={"shared": "0b101"}) + self.assertEqual(driver.trigger_terms[-1], (0xe, 0xa)) + + driver.configure_group(1) + driver.add_trigger(cond={"shared": "0b101"}) + self.assertEqual(driver.trigger_terms[-1], (0x7, 0x5)) + def test_configure_subsampler_rejects_invalid_values(self): driver, regs = self.make_driver(subsampler_width=4)