cocotb_benches.dau_range_partitioner_tb¶

Attributes¶

`partition_routes_rows_by_range`(dut)
`boundary_key_lands_in_upper_partition`(dut)
`all_rows_to_one_partition`(dut)
`negative_keys_partition_signed`(dut)
`last_lands_on_each_partitions_final_row`(dut)
`int32_extreme_keys_partition_signed`(dut)
`backpressure_on_every_stream`(dut)
`torn_row_status_survives_backpressure`(dut)
`status_backpressure_stalls_the_front`(dut)
`constrained_random_streams_match_golden`(dut)
`_num_parts`(dut) → int
`_default_splitters`(dut) → list[int]
`_spread_splitters`(lo, hi, count) → list[int]
`_signed32`(value) → int
`_random_monotonic_splitters`(rng, count) → list[int]
`_random_rows`(rng) → list[tuple[int, int, int, int]]
`_set_splitters`(dut, splitters) → None
`_start`(dut) → None
`_run`(dut, rows, splitters, *[, ready_pattern])
`_send_rows`(dut, rows) → None
`_drive_torn_first_row`(dut) → None
`_drive_and_collect`(dut, golden, ready_pattern)	Drive all partition output/status streams and reassemble per-partition
`_assert_quiescent`(dut, num_parts) → None	After a batch completes, nothing more may appear on any output or status
`_drain_partition_output`(dut, partition) → None
`_collect_all_status`(dut, *[, only])	Wait for a status beat from every (or `only`) partition; return
`_unpack_rows`(words) → list[tuple[int, int, int, int]]

async cocotb_benches.dau_range_partitioner_tb.partition_routes_rows_by_range(dut)¶

async cocotb_benches.dau_range_partitioner_tb.boundary_key_lands_in_upper_partition(dut)¶

async cocotb_benches.dau_range_partitioner_tb.all_rows_to_one_partition(dut)¶

async cocotb_benches.dau_range_partitioner_tb.negative_keys_partition_signed(dut)¶

async cocotb_benches.dau_range_partitioner_tb.last_lands_on_each_partitions_final_row(dut)¶

async cocotb_benches.dau_range_partitioner_tb.int32_extreme_keys_partition_signed(dut)¶

async cocotb_benches.dau_range_partitioner_tb.backpressure_on_every_stream(dut)¶

async cocotb_benches.dau_range_partitioner_tb.torn_row_status_survives_backpressure(dut)¶

async cocotb_benches.dau_range_partitioner_tb.status_backpressure_stalls_the_front(dut)¶

async cocotb_benches.dau_range_partitioner_tb.constrained_random_streams_match_golden(dut)¶

cocotb_benches.dau_range_partitioner_tb._default_splitters(dut) → list[int]¶

cocotb_benches.dau_range_partitioner_tb._spread_splitters(lo: int, hi: int, count: int) → list[int]¶

cocotb_benches.dau_range_partitioner_tb._random_monotonic_splitters(rng: random.Random, count: int) → list[int]¶

cocotb_benches.dau_range_partitioner_tb._random_rows(rng: random.Random) → list[tuple[int, int, int, int]]¶

cocotb_benches.dau_range_partitioner_tb._set_splitters(dut, splitters: list[int]) → None¶

async cocotb_benches.dau_range_partitioner_tb._run(dut, rows, splitters, *, ready_pattern=(1,))¶

async cocotb_benches.dau_range_partitioner_tb._send_rows(dut, rows) → None¶

async cocotb_benches.dau_range_partitioner_tb._drive_torn_first_row(dut) → None¶

async cocotb_benches.dau_range_partitioner_tb._drive_and_collect(dut, golden, ready_pattern)¶: Drive all partition output/status streams and reassemble per-partition rows. Each active partition gets an independent (phase-shifted) ready pattern so lanes stall out of step. A partition that expects rows completes on its last word; a partition that expects zero rows completes on its success status. After completion the streams must fall quiescent.

async cocotb_benches.dau_range_partitioner_tb._assert_quiescent(dut, num_parts: int) → None¶: After a batch completes, nothing more may appear on any output or status stream even with every ready held high (no trailing/duplicated beats).

async cocotb_benches.dau_range_partitioner_tb._drain_partition_output(dut, partition: int) → None¶

async cocotb_benches.dau_range_partitioner_tb._collect_all_status(dut, *, only: set[int] | None = None)¶: Wait for a status beat from every (or only) partition; return (error, code) per partition (None for partitions not collected).

cocotb_benches.dau_range_partitioner_tb._unpack_rows(words: list[int]) → list[tuple[int, int, int, int]]¶