English | δΈζ
Wavekit is a fundamental Python library for digital waveform analysis. By seamlessly converting VCD, FST, and FSDB data into Numpy arrays, it empowers engineers to perform high-performance signal processing, protocol analysis, and automated verification with ease.
π€ AI Integration: wavekit-mcp β MCP server for AI-assisted waveform analysis. Let AI load signals and run pattern matching β no manual coding required.
- Flexible Signal Extraction: Flexible batch signal extraction via brace expansion, integer ranges, and regular expressions β load groups of related signals in one call.
- Rich Analysis Tools: Numpy-like API for arithmetic, masking, bit-field manipulation, edge detection, and time/cycle slicing β compose complex signal queries in just a few lines.
- Pattern Matching: Unified temporal pattern runtime that scans waveforms in a single pass to extract protocol transactions, measure latencies, and detect timing violations.
- High-Performance Parsing & Storage: VCD, FST, and FSDB readers with Numpy-backed storage for fast loading and memory efficiency, handling large simulation files with ease.
pip install wavekitNote: To read FSDB files, the Verdi runtime library (libNPI.so) must be available at runtime. Configure via:
WAVEKIT_NPI_LIBβ direct path tolibNPI.soVERDI_HOMEβ Verdi installation directory (searches$VERDI_HOME/share/NPI/lib/...)LD_LIBRARY_PATHβ system library search path
The examples below use placeholder filenames such as
sim.vcd. Replace them with the path to your own VCD, FST, or FSDB file, and adjust signal paths to match your design hierarchy.
Use brace expansion or regular expressions to load multiple related signals in one call.
from wavekit import VcdReader
with VcdReader("jtag.vcd") as f:
# Keys contain BraceCapture objects whose groups are ("state",) / ("next",).
waves = f.load_matched_waveforms(
"tb.u0.J_{state,next}[3:0]",
clock_path="tb.tck",
)
# /regex/ is the canonical regex syntax; groups are stored by RegexCapture.
waves = f.load_matched_waveforms(
r"tb.u0./J_([a-z]+)/",
clock_path="tb.tck",
)Waveforms support Numpy-style arithmetic, masking, and edge detection out of the box.
import numpy as np
from wavekit import VcdReader
with VcdReader("fifo_tb.vcd") as f:
clock = "fifo_tb.clk"
depth = 8
w_ptr = f.load_waveform("fifo_tb.s_fifo.w_ptr[2:0]", clock=clock)
r_ptr = f.load_waveform("fifo_tb.s_fifo.r_ptr[2:0]", clock=clock)
wr_en = f.load_waveform("fifo_tb.s_fifo.wr_en", clock=clock)
occupancy = (w_ptr + depth - r_ptr) % depth
print(f"Average occupancy: {np.mean(occupancy.value):.2f}")
# Filter to cycles where a write is active
write_occ = occupancy.mask(wr_en == 1)
# Detect write bursts
burst_cycles = wr_en.rising_edge()To inspect unknown/high-impedance source bits without changing the ordinary
two-state value model, load an unsigned unknown mask alongside the value
waveform. Each mask bit is 1 where the source sample contained X or Z.
Experimental: The
load_unknown_mask/load_matched_unknown_masksAPIs are experimental and may change in a future release.
from wavekit import VcdReader
with VcdReader("fifo_tb.vcd") as f:
clock = "fifo_tb.clk"
data = f.load_waveform("fifo_tb.s_fifo.data[7:0]", clock=clock, xz_value=0)
unknown = f.load_unknown_mask("fifo_tb.s_fifo.data[7:0]", clock=clock)
# Keep only samples whose source bits were fully known.
known_data = data.mask(unknown == 0)Compute waveform expressions directly from signal path strings without loading each signal manually.
from wavekit import VcdReader
with VcdReader("fifo_tb.vcd") as f:
# Single mode: paths must each match exactly one signal
occupancy = f.eval(
"fifo_tb.s_fifo.w_ptr[2:0] - fifo_tb.s_fifo.r_ptr[2:0]",
clock="fifo_tb.clk",
)
# Zip mode: brace patterns expand per typed CaptureKey and evaluate per match.
occupancies = f.eval(
"tb.fifo_{0..3}.w_ptr[2:0] - tb.fifo_{0..3}.r_ptr[2:0]",
clock="tb.clk",
mode="zip",
)Pattern scans a waveform and extracts all matching transactions β a request/response pair, a burst, a stall interval, or any other repeating timing pattern.
There are two ways to describe a pattern:
- Declarative β chain steps like
.wait(),.consume(),.capture(),.loop(). Best for fixed transaction flows. - Programmable β pass a handler function to
Pattern(...). Best for dynamic branches, per-ID routing, and other complex flows.
AXI-Lite read latency
from wavekit import VcdReader, Pattern
with VcdReader("axi_tb.vcd") as f:
clk = "tb.clk"
arvalid = f.load_waveform("tb.dut.arvalid", clock=clk)
arready = f.load_waveform("tb.dut.arready", clock=clk)
rvalid = f.load_waveform("tb.dut.rvalid", clock=clk)
rready = f.load_waveform("tb.dut.rready", clock=clk)
rdata = f.load_waveform("tb.dut.rdata[31:0]", clock=clk)
result = (
Pattern(timeout=256)
.wait(arvalid & arready) # AR handshake β transaction starts
.wait(rvalid & rready) # R handshake β transaction ends
.capture("rdata", rdata)
.match()
)
ok = result.filter_ok()
print(f"Read latencies (cycles): {ok.duration.value}")
print(f"Read data: {ok.captures['rdata'].value}")AXI write burst (multi-beat)
beat = Pattern().consume(wvalid & wready, channel="w").capture("beats", wdata, mode="list")
result = (
Pattern()
.wait(awvalid & awready) # AW handshake β burst starts
.loop(beat, until=wlast) # collect each beat until wlast
.timeout(512)
.match()
)
for i, inst in enumerate(result.filter_ok()):
print(f"Burst {i}: {len(inst.captures['beats'])} beats")Stall detection
stall = valid & (ready == 0)
result = (
Pattern()
.wait(stall.rising_edge()) # stall begins
.loop(Pattern().delay(1), when=stall) # keep waiting until stall ends
.match()
)
stalls = result.filter_ok()
print(f"Stall durations: {stalls.duration.value} cycles")Out-of-order AXI reads by ID
When R beats from different IDs interleave on the bus, match each AR to its response beats by arid and collect results as Python dicts.
arfire = arvalid & arready # precompute outside the handler
rfire = rvalid & rready
async def read_burst(ctx):
if ctx.value(arfire):
my_id = ctx.value(arid)
beats = []
while True:
await ctx.consume(
lambda: ctx.value(rfire) and ctx.value(rid) == my_id,
channel=("r", my_id),
)
beats.append(int(ctx.value(rdata)))
if ctx.value(rlast):
break
return {"arid": my_id, "beats": beats}
return None
records = Pattern(read_burst, timeout=64).collect()Some tips for programmable patterns:
- Precompute fixed waveform expressions (like
fire = valid & ready) outside the handler function so they aren't rebuilt every cycle. - Start the handler with
if ctx.value(fire): ...andreturn Noneotherwise β this tells the runtime which cycles begin a transaction.
| Method | Description |
|---|---|
VcdReader(file) / FstReader(file) / FsdbReader(file) |
Open a waveform file. Use as a context manager. FsdbReader requires Verdi runtime (WAVEKIT_NPI_LIB, VERDI_HOME, or LD_LIBRARY_PATH). |
reader.load_waveform(signal, clock, ...) |
Load one signal sampled on every clock edge. Returns Waveform. |
reader.load_unknown_mask(signal, clock, ...) |
Experimental. Load X/Z bit presence as an unsigned mask Waveform. |
reader.load_matched_waveforms(signal_path, clock_path, ...) |
Batch-load matching signals. Returns dict[CaptureKey, Waveform]. |
reader.load_matched_unknown_masks(signal_path, clock_path, ...) |
Experimental. Batch-load X/Z masks for matched signals. Returns dict[CaptureKey, Waveform]. |
reader.eval(expr, clock, mode='single'|'zip', ...) |
Evaluate an arithmetic expression with embedded signal paths. |
reader.get_matched_signals(path) |
Resolve a query to Signal objects without loading data. |
reader.get_matched_scopes(path) |
Resolve a query to Scope objects. |
reader.top_scopes |
Immutable tuple of root Scope nodes. |
has_fsdb_support() |
Report whether the Verdi FSDB runtime is currently available. |
Matched-reader and hierarchy query APIs use CaptureKey = tuple[Capture, ...].
Ordinary exact-name components are omitted, so an exact query uses key ().
Bindings remain typed: BraceCapture and RegexCapture expose .groups,
WildcardCapture records the matched .path, and FSDB module queries retain an
ExactCapture with .definition.
Node is the immutable base for Scope and Signal. Nodes expose .parent,
immutable .children, local .base_name / .name, and qualified .full_name.
A Signal additionally exposes .range, .native_range, .width,
.native_width, .is_leaf, and .composite_type; Range(start, end) preserves
HDL declaration direction.
Pattern syntax used in signal paths:
| Syntax | Example | Effect |
|---|---|---|
{a,b,c} |
sig_{read,write} |
Enumerate named variants |
{N..M} |
fifo_{0..3}.ptr |
Integer range |
{N..M..step} |
lane_{0..6..2} |
Stepped range |
/<regex>/ |
/([a-z]+)_valid/ |
Canonical regex syntax with capture groups |
@<regex> |
@([a-z]+)_valid |
Legacy-compatible regex syntax |
* / ** |
tb.*.valid / tb.**.valid |
Single-level / recursive wildcard |
$ModName |
tb.$fifo_unit.ptr |
Match a direct-child scope by module/definition name (FSDB only) |
$$ModName |
tb.$$fifo_unit.ptr |
Match any-depth descendant scope by module/definition name (FSDB only) |
A Waveform wraps three parallel numpy arrays (.value, .clock, .time). All operations return a new Waveform.
Arithmetic & comparison: +, -, *, //, %, **, /, &, |, ^, ~, ==, !=, <<, >>
Filtering & slicing
| Method | Description |
|---|---|
wave.mask(mask) |
Keep samples where a boolean Waveform or array is True |
wave.filter(fn) |
Keep samples where fn(value) is True |
wave.cycle_slice(begin, end) |
Trim to clock cycle range [begin, end) |
wave.time_slice(begin, end) |
Trim to simulation time range |
wave.slice(begin_idx, end_idx) |
Trim by array index |
wave.take(indices) |
Select samples at given indices |
Transformation
| Method | Description |
|---|---|
wave.map(fn, width, signed) |
Element-wise transform |
wave.unique_consecutive() |
Remove consecutive duplicates |
wave.downsample(chunk, fn) |
Aggregate into chunks |
wave.as_signed() / wave.as_unsigned() |
Reinterpret signedness |
Bit manipulation
| Method / Syntax | Description |
|---|---|
wave[high:low] |
Extract bit field (Verilog convention, returns unsigned) |
wave[n] |
Extract single bit |
wave.split_bits(n) |
Split into n-bit groups (LSB first) |
Waveform.concatenate([w0, w1, ...]) |
Concatenate (w0 = LSB) |
wave.bit_count() |
Population count |
Edge detection (1-bit only)
| Method | Description |
|---|---|
wave.rising_edge() |
True at 0β1 transitions |
wave.falling_edge() |
True at 1β0 transitions |
Relative time access
| Method | Description |
|---|---|
wave.relative(offset, pad, pad_value) |
Shift by offset cycles (positive = future, negative = past) |
wave.ahead(n, pad, pad_value) |
Look n cycles into the future (shorthand for relative(n)) |
wave.back(n, pad, pad_value) |
Look n cycles into the past (shorthand for relative(-n)) |
pad controls boundary handling: 'repeat' (default) pads with the first/last value, 'value' pads with a given pad_value.
# Rising edge detection
rising = (wave == 0) & wave.ahead()
# Compare current vs 3 cycles ago
changed = wave != wave.back(3)Construction
| API | Description |
|---|---|
Pattern(timeout=..., max_active=...) |
Create a Declarative Pattern. Add steps with builder methods, then call .match(). |
Pattern(async_fn, timeout=..., max_active=...) |
Create a Programmable Pattern. The async function receives ctx. |
.match(start_cycle=None, end_cycle=None) |
Run the pattern and return MatchResult. In a Programmable Pattern, return ctx.OK for success and None to skip. |
.collect(start_cycle=None, end_cycle=None) |
Programmable Pattern only. Collect each non-None Python return value. |
Declarative Steps
| Method | Description |
|---|---|
.wait(cond, *, require=None) |
Block until cond is True without consuming the event. Resumes in the same cycle when already true; use .delay(1) for next-cycle behavior. require is checked each waiting cycle (failure β REQUIRE_VIOLATED). |
.consume(cond, channel, *, require=None) |
Block until cond is True and this instance can exclusively consume from channel. Resumes in the same cycle on success. Use this for FIFO request/response pairing and per-key routing. |
.delay(n, *, require=None) |
Advance n cycles. delay(0) is a no-op. require must hold every cycle. |
.capture(name, signal, *, mode='last') |
Record signal value at current cycle. mode='last' (default) overwrites; 'first' keeps the first write; 'list' appends to a list. |
.require(cond) |
Assert condition; fail with REQUIRE_VIOLATED if False. |
.loop(body, *, until=None, when=None) |
until: do-while (exit when True after body). when: while (exit when False before body). |
.repeat(body, n) |
Execute body exactly n times. n may be a callable. |
.branch(cond, true_body, false_body) |
Conditional branch. |
.timeout(max_cycles) |
Terminate unfinished instances with TIMEOUT. |
The same time and ownership operations are available inside Programmable
Patterns as await ctx.wait(...), await ctx.consume(...), and
await ctx.delay(...).
Programmable Context
| API | Description |
|---|---|
ctx.value(waveform, offset=0) |
Read a scalar value at the current sample plus optional offset. |
ctx.cycle(waveform, offset=0) |
Read the cycle number at the current sample plus optional offset. |
ctx.time(waveform, offset=0) |
Read the timestamp at the current sample plus optional offset. |
await ctx.wait(cond, require=None) |
Observe cycles until cond is true; does not consume the event. |
await ctx.consume(cond, channel, require=None) |
Wait for cond and exclusively consume from channel. |
await ctx.delay(n, require=None) |
Advance n cycles. |
ctx.capture(name, value, mode='last') |
Record a capture for programmable .match(). |
ctx.OK |
Return from programmable .match() to record a successful match. |
Dynamic callbacks
Declarative callbacks use callable(index, captures). index is the absolute
waveform sample index; it is not rebased when match(start_cycle=...) is used.
Channels and consume vs. wait
When multiple in-flight instances are waiting for the same kind of event, plain
wait() won't pair each request with its own response β every instance sees
every event. consume() solves this: it hands the event to exactly one instance
per cycle, in FIFO order.
A Channel is the FIFO queue that consume() uses. Pass a Channel object, a
hashable key, or a dynamic callback to consume(..., channel=...). All
instances sharing the same channel key consume from the same queue.
from collections import defaultdict
from wavekit import Channel, Pattern
# Multi-bank cache: each bank has its own response port, so two banks
# can return data in the *same* cycle. A per-bank Channel lets each in-flight
# read consume from its own bank while preserving FIFO order within that bank.
banks = defaultdict(Channel)
result = (
Pattern()
.wait(req_valid)
.capture('bank', req_addr & 1)
.consume(
lambda i, cap: bank_valid[cap['bank']].value[i],
channel=lambda i, cap: banks[cap['bank']],
)
.capture('rdata',
lambda i, cap: bank_data[cap['bank']].value[i])
.match()
)MatchResult
| Field | Description |
|---|---|
.start / .end |
Start and end cycle of each match (both inclusive). |
.duration |
end - start + 1 cycles. |
.status |
MatchStatus.OK, TIMEOUT, or REQUIRE_VIOLATED. |
.captures |
dict[str, Waveform] of captured values. |
.ok |
Boolean Waveform where status == MatchStatus.OK. |
.failed |
Boolean Waveform where status != MatchStatus.OK. |
.filter_ok() |
Return only OK matches. |
.filter_status(status) |
Return only matches with the given MatchStatus or integer status. |
.filter_failed() |
Return only non-OK matches. |
This project uses Poetry for dependency management and packaging.
git clone https://github.com/cxzzzz/wavekit.git
cd wavekit
poetry installTests are located in the tests/ directory and run with pytest.
# Run all tests
poetry run pytest
# Run a specific test file
poetry run pytest tests/test_pattern.py
# Run with verbose output
poetry run pytest -vThis project uses Ruff for linting and formatting.
# Check for lint errors
poetry run ruff check .
# Check formatting (no changes)
poetry run ruff format --check .
# Auto-fix formatting
poetry run ruff format .poetry run mypy .Contributions are welcome! Please open an issue to discuss a bug or feature request before submitting a pull request. When contributing code, make sure all tests pass and the linter reports no errors:
poetry run pytest
poetry run ruff check .
poetry run ruff format --check .This project is licensed under the MIT License. See the LICENSE file for details.