wavekit

English | 中文

wavekit is a high-performance Python library for digital waveform analysis. It loads VCD, FST, and FSDB signals into NumPy-backed Waveform objects, and builds on them with efficient signal processing, protocol analysis, and automated verification APIs.

🤖 AI integration: wavekit-mcp provides an MCP server for AI-assisted waveform analysis: load signals and run pattern matching from AI tools without hand-written scripts.

Features

• Batch signal extraction: use brace expansion, integer ranges, regex, and wildcards to load related signals in one call.
• Rich analysis API: use NumPy-style arithmetic, masking, bit slicing, edge detection, and time/cycle slicing to build signal queries in a few lines.
• Temporal pattern matching: use declarative and programmable Pattern APIs to extract protocol transactions, measure handshake latency, and detect timeout or hang failures.
• High-performance waveform processing: read VCD, FST, and FSDB files into compact NumPy-backed arrays for fast loading and efficient memory use.

Installation

pip install wavekit

Note: To read FSDB files, the Verdi runtime library (libNPI.so) must be available at runtime. Configure via:

• WAVEKIT_NPI_LIB: direct path to libNPI.so
• VERDI_HOME: Verdi installation directory (searches $VERDI_HOME/share/NPI/lib/...)
• LD_LIBRARY_PATH: system library search path

Quick start

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.

1. Batch Signal Extraction

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",
    )

2. Signal Analysis

Waveforms support NumPy-style arithmetic, masking, and edge detection.

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_masks APIs 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)

3. Expression Evaluation

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",
    )

4. Pattern Matching

Pattern scans a waveform and extracts matching transactions, such as a request/response pair, a burst, a stall interval, or another repeating timing pattern.

Choose the form that matches the transaction shape:

• Declarative: use .wait(), .consume(), .capture(), and .loop() steps to describe fixed flows.
• Programmable: use a normal Python function when the flow depends on waveform values, such as dynamic branches or per-ID routing.

In declarative patterns, the first blocking step selects transaction start cycles; later blocking steps describe timing within each matched transaction.

match(pattern) returns MatchRecords, while collect(body) returns a Python list of extracted values.

Declarative examples

AXI-Lite read latency

from wavekit import VcdReader
from wavekit.pattern import Pattern, match, collect

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)

    pattern = (
        Pattern()
        .wait(arvalid & arready)   # AR handshake → transaction starts
        .wait(rvalid  & rready)    # R  handshake → transaction ends
        .capture("rdata", rdata)
    )
    result = match(pattern)

    ok = result.filter_ok()
    latency = ok.end.clock - ok.start.clock
    print(f"Read latencies (cycles): {latency}")
    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")

pattern = (
    Pattern()
    .wait(awvalid & awready)   # AW handshake → burst starts
    .loop(beat, until=wlast)   # collect each beat until wlast
)
result = match(pattern)

for i, inst in enumerate(result.filter_ok()):
    print(f"Burst {i}: {len(inst.captures['beats'])} beats")

Stall detection

stall = valid & (ready == 0)

pattern = (
    Pattern()
    .wait(stall.rising_edge())             # stall begins
    .loop(Pattern().delay(1), when=stall)  # keep waiting until stall ends
)
result = match(pattern)

stalls = result.filter_ok()
stall_cycles = stalls.duration.value - 1
print(f"Stall durations: {stall_cycles} cycles")

Programmable example

DMA-style command stream

Use programmable control flow when a command's opcode changes the following timing shape.

cmd_fire = cmd_valid & cmd_ready   # precompute outside the handler
w_fire = w_valid & w_ready
rsp_fire = rsp_valid & rsp_ready
r_fire = r_valid & r_ready

OP_READ = 0
OP_WRITE = 1

def read_dma_cmd(ctx):
    if not ctx.value(cmd_fire):
        return None

    op = int(ctx.value(cmd_op))
    addr = int(ctx.value(cmd_addr))
    length = int(ctx.value(cmd_len))

    if op == OP_WRITE:
        data = []
        for _ in range(length):
            ctx.consume(w_fire, channel="wdata")
            data.append(int(ctx.value(w_data)))

        ctx.consume(rsp_fire, channel="rsp")
        return {
            "op": "write",
            "addr": addr,
            "data": data,
            "status": int(ctx.value(rsp_status)),
        }

    if op == OP_READ:
        ctx.consume(rsp_fire, channel="rsp")
        data = []
        for _ in range(length):
            ctx.consume(r_fire, channel="rdata")
            data.append(int(ctx.value(r_data)))

        return {"op": "read", "addr": addr, "data": data}

    ctx.require(False, message=f"unknown DMA op {op}")
    return None

commands = collect(read_dma_cmd)
print(f"Captured {len(commands)} commands")

Some tips for programmable patterns:

• Precompute fixed waveform expressions, such as fire = valid & ready, outside the handler so they are not rebuilt every cycle.
• Start the handler with if ctx.value(fire): ... to test whether the current cycle starts a transaction, and return None otherwise.
• Use ctx.try_consume(...) for non-blocking polling or arbitration between candidate channels. For a linear burst, ctx.consume(...) is more direct.
• Add timeout=<cycles> only when a blocking step needs a wait bound. In match(), timeout becomes MatchStatus.Timeout(...); in collect(), it raises PatternError.

API reference

Reader

Signal path patterns

$ and $$ are path-step modifiers: they can combine with exact names, brace expansion, regex, and a trailing range selector in the same query. For example: tb.$/fifo_(in|out)/.data_{0..3}[7:0].

Matched-reader and hierarchy query APIs return dictionaries keyed by CaptureKey = tuple[Capture, ...]. Exact-name components are omitted, so a fully exact query uses key ().

Waveform

A Waveform wraps three parallel numpy arrays (.value, .clock, .time). All operations return a new Waveform.

Arithmetic & comparison: +, -, *, //, %, **, /, &, |, ^, ~, ==, !=, <<, >>

Filtering & slicing

Transformation

Bit manipulation

Edge detection (1-bit only)

Relative time access

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)

Pattern

Construction

Declarative Steps

The same time and ownership operations are available inside Programmable patterns as ctx.wait(...), ctx.consume(...), and ctx.delay(...).

Programmable Context

Dynamic callbacks

Callback arguments depend on whether the callback is used in the declarative API or the programmable API:

• In declarative APIs such as Pattern().wait(...) and Pattern().consume(...), callbacks receive (index, captures).
• In programmable functions, callbacks passed to methods such as ctx.wait(...) and ctx.consume(...) take no arguments. If they need the current index, captures, or signal values, close over ctx.

Channels and consume vs. wait

wait() is observational: every matching instance can see the same event. consume() adds ownership: only one instance can claim a given (channel, cycle). On the same channel, matches with earlier start cycles claim available events first.

A Channel is an identity token for consume ownership. Pass a Channel object, a hashable key, or a dynamic callback to consume(..., channel=...). All instances sharing the same channel key compete for the same logical channel.

from collections import defaultdict
from wavekit.pattern import Channel, Pattern, match

# Multi-bank cache: each bank has its own response port, so two banks can return
# data in the same cycle. Per-bank channels let independent requests consume
# independent response streams.
banks = defaultdict(Channel)

pattern = (
    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])
)
result = match(pattern)

MatchRecords

MatchRecords[i] returns a single MatchRecord, and slices return another MatchRecords batch.

Development

This project uses Poetry for dependency management and packaging.

Setup

git clone https://github.com/cxzzzz/wavekit.git
cd wavekit
poetry install

Testing

Tests 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 -v

Linting & Formatting

This 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 .

Type Checking

poetry run mypy .

Contributing

Issues and pull requests are welcome. Before sending a pull request, run the tests and format checks:

poetry run pytest
poetry run ruff check .
poetry run ruff format --check .

License

This project is licensed under the MIT License. See the LICENSE file for details.