gr-mcp/docs/block-dev-workflow.md
Ryan Malloy 212832e7e4 feat: expose protocol analysis, OOT export tools; harden for release
Wire up protocol analysis (parse_protocol_spec, generate_decoder_chain,
get_missing_oot_modules), signal analysis (analyze_iq_file), and OOT
export (generate_oot_skeleton, export_block_to_oot, export_from_flowgraph)
as MCP tools with integration tests.

Security fixes from Hamilton review:
- Remove `from __future__ import annotations` from tool registration
  files (breaks FastMCP schema generation)
- Add blocklist guard to evaluate_expression (was unsandboxed eval)
- Replace string interpolation with base64 encoding in Docker test
  harness (prevents code injection)
- Add try/finally cleanup for temp files and Docker containers
- Replace assert with proper ValueError in flowgraph block creation
- Log OOT auto-discovery failures instead of swallowing silently

Packaging:
- Move entry point to src/gnuradio_mcp/server.py with script entry
  point (uv run gnuradio-mcp)
- Add PyPI metadata (authors, license, classifiers, urls)
- Add MIT LICENSE file
- Rewrite README for current feature set (80+ tools)
- Document single-session limitation
2026-02-20 13:17:11 -07:00

14 KiB
Raw Permalink Blame History

AI-Assisted Block Development Workflow

This document describes the complete workflow for developing custom GNU Radio blocks using GR-MCP's block development tools. The workflow enables rapid iteration from concept to distributable OOT module.

Overview

GR-MCP provides an AI-assisted development workflow that transforms signal processing requirements into working GNU Radio blocks:

Protocol Spec / IQ Recording
         │
         ▼
┌─────────────────────────┐
│   Protocol Analysis     │  parse_protocol_spec()
│   Signal Detection      │  analyze_iq_file()
└─────────────────────────┘
         │
         ▼
┌─────────────────────────┐
│   Block Generation      │  generate_sync_block()
│   Decoder Chains        │  generate_decoder_chain()
└─────────────────────────┘
         │
         ▼
┌─────────────────────────┐
│   Validation & Test     │  validate_block_code()
│   Docker Testing        │  test_block_in_docker()
└─────────────────────────┘
         │
         ▼
┌─────────────────────────┐
│   OOT Export            │  export_block_to_oot()
│   Distribution          │  generate_oot_skeleton()
└─────────────────────────┘

Enabling Block Dev Mode

Block development tools are dynamically registered to minimize context usage. Enable them when needed:

# Check if enabled
get_block_dev_mode()

# Enable block development tools
enable_block_dev_mode()

# Disable when done
disable_block_dev_mode()

Phase 1: Protocol Analysis

Parse natural language protocol specifications into structured models.

parse_protocol_spec

Extracts modulation, framing, and encoding parameters from protocol descriptions.

result = parse_protocol_spec(
    spec_text="""
    GFSK signal at 250k baud, deviation: 25khz
    Preamble: 0xAA (8 bytes)
    Sync word: 0x2D, 0xD4
    CRC-16 at end of frame
    """
)

# Returns ProtocolModel with:
# - modulation.scheme = "GFSK"
# - modulation.symbol_rate = 250000.0
# - modulation.deviation = 25000.0
# - framing.preamble = "0xAA"
# - framing.sync_word = "0x2D, 0xD4"
# - encoding.crc = "CRC-16"

Supported Parameters:

Category Parameters
Modulation scheme (FSK, GFSK, BPSK, QPSK, OFDM, CSS), symbol_rate, deviation, order
Framing preamble, sync_word, header_format, frame_length
Encoding fec_type, interleaving, whitening, crc

generate_decoder_chain

Creates a complete decoder pipeline from a parsed protocol specification.

# Parse protocol first
protocol = parse_protocol_spec("GFSK at 50k baud, deviation: 25khz")

# Generate decoder chain
chain = generate_decoder_chain(
    protocol=protocol,
    sample_rate=2000000.0
)

# Returns DecoderPipelineModel with:
# - blocks: list of DecoderBlock with block_type, parameters
# - connections: list of (src_block, src_port, dst_block, dst_port)
# - variables: dict of flowgraph variables
# - missing_oot_modules: list of required OOT modules

Generated Blocks by Modulation:

Modulation Blocks Generated
FSK/GFSK low_pass_filter → analog_quadrature_demod_cf → clock_recovery
BPSK costas_loop_cc → constellation_decoder_cb
LoRa/CSS freq_xlating_fir_filter → lora_demod (requires gr-lora_sdr)

get_missing_oot_modules

Check which OOT modules are required for a decoder chain.

# Parse a LoRa protocol
protocol = parse_protocol_spec("CSS/LoRa at SF7, 125kHz bandwidth")

# Check missing modules
missing = get_missing_oot_modules(protocol)
# Returns: ["gr-lora_sdr"]

Phase 2: Signal Analysis

Analyze IQ recordings to identify signal characteristics.

analyze_iq_file

Performs FFT-based spectral analysis and signal detection.

result = analyze_iq_file(
    file_path="/path/to/recording.cf32",
    sample_rate=2000000.0,
    dtype="complex64"  # or "complex128", "int16"
)

# Returns IQAnalysisResult with:
# - sample_count: int
# - duration_seconds: float
# - power_stats: {min_db, max_db, mean_db, std_db}
# - spectral_features: {peak_frequency, bandwidth_3db, ...}
# - signals_detected: list of detected signal regions

Supported Data Types:

Format dtype Parameter
Complex float32 (GNU Radio default) complex64
Complex float64 complex128
Interleaved int16 (RTL-SDR) int16

Phase 3: Block Generation

Generate GNU Radio block code from specifications.

generate_sync_block

Creates a gr.sync_block with 1:1 input/output sample relationship.

result = generate_sync_block(
    name="pm_demod",
    description="Phase modulation demodulator",
    inputs=[{"dtype": "complex", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    parameters=[
        {"name": "sensitivity", "dtype": "float", "default": 1.0}
    ],
    work_logic="Extract instantaneous phase from complex samples"
)

# Returns GeneratedBlockCode with:
# - source_code: complete Python block implementation
# - block_name: "pm_demod"
# - block_class: "sync_block"
# - is_valid: bool
# - validation_errors: list[str]

Work Templates:

Pre-built templates for common operations:

Template Description
gain Multiply samples by gain factor
add Add constant to samples
threshold Binary threshold comparison
# Using a work template
result = generate_sync_block(
    name="my_gain",
    description="Variable gain",
    inputs=[{"dtype": "float", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    parameters=[{"name": "gain", "dtype": "float", "default": 1.0}],
    work_template="gain"
)

generate_basic_block

Creates a gr.basic_block with custom input/output ratios.

result = generate_basic_block(
    name="frame_sync",
    description="Frame synchronizer with variable output",
    inputs=[{"dtype": "byte", "vlen": 1}],
    outputs=[{"dtype": "byte", "vlen": 1}],
    parameters=[
        {"name": "sync_word", "dtype": "int", "default": 0x2DD4}
    ],
    work_logic="Search for sync word and output aligned frames",
    forecast_logic="noutput_items + len(self.buffer)"
)

generate_interp_block / generate_decim_block

Create blocks with fixed interpolation or decimation ratios.

# Interpolating block (2x output samples per input)
interp = generate_interp_block(
    name="upsample_2x",
    description="2x upsampler with zero-stuffing",
    inputs=[{"dtype": "float", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    interpolation=2,
    work_logic="Zero-stuff between samples"
)

# Decimating block (4x fewer output samples)
decim = generate_decim_block(
    name="downsample_4x",
    description="4x downsampler",
    inputs=[{"dtype": "float", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    decimation=4,
    work_logic="Output every 4th sample"
)

validate_block_code

Static analysis without execution.

result = validate_block_code(source_code=my_block_code)

# Returns ValidationResult with:
# - is_valid: bool
# - errors: list[str] (syntax errors, missing imports)
# - warnings: list[str] (style issues, potential bugs)

test_block_in_docker

Test generated blocks in an isolated container.

result = test_block_in_docker(
    source_code=my_block_code,
    test_input=[1.0, 2.0, 3.0, 4.0],
    expected_output=[2.0, 4.0, 6.0, 8.0],  # optional
    timeout_seconds=30
)

# Returns BlockTestResult with:
# - passed: bool
# - actual_output: list[float]
# - error_message: str (if failed)
# - execution_time_ms: float

Phase 4: OOT Export

Export generated blocks to distributable OOT modules.

generate_oot_skeleton

Create an empty gr_modtool-compatible module structure.

result = generate_oot_skeleton(
    module_name="mymodule",
    output_dir="/path/to/gr-mymodule",
    author="Your Name",
    description="My custom GNU Radio blocks"
)

# Creates:
# gr-mymodule/
# ├── CMakeLists.txt
# ├── python/
# │   └── mymodule/
# │       └── __init__.py
# └── grc/
#     └── (empty, for .block.yml files)

export_block_to_oot

Export a generated block to an existing or new OOT module.

# First generate a block
block = generate_sync_block(
    name="pm_demod",
    description="Phase modulation demodulator",
    inputs=[{"dtype": "complex", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    parameters=[{"name": "sensitivity", "dtype": "float", "default": 1.0}]
)

# Export to OOT module
result = export_block_to_oot(
    generated=block,
    module_name="apollo",
    output_dir="/path/to/gr-apollo",
    author="Ryan Malloy"
)

# Creates:
# gr-apollo/
# ├── CMakeLists.txt
# ├── python/apollo/
# │   ├── __init__.py
# │   └── pm_demod.py      ← Block implementation
# └── grc/
#     └── apollo_pm_demod.block.yml  ← GRC block definition

export_from_flowgraph

Export an embedded Python block from the current flowgraph.

# After creating an embedded block with create_embedded_python_block()
result = export_from_flowgraph(
    block_name="epy_block_0",
    module_name="custom",
    output_dir="/path/to/gr-custom",
    author="Your Name"
)

Complete Workflow Example

Example: Apollo USB PCM Decoder

This example demonstrates the full workflow for creating a decoder for Apollo mission telemetry.

# 1. Enable block dev mode
enable_block_dev_mode()

# 2. Parse the protocol specification
protocol = parse_protocol_spec("""
    Apollo Unified S-Band PCM telemetry:
    - BPSK subcarrier at 1.024 MHz
    - 51.2 kbps bit rate
    - Manchester encoding
    - Frame: 128 words × 8 bits @ 50 fps
    - Frame sync pattern: 0xEB9000
""")

# 3. Generate decoder chain
chain = generate_decoder_chain(
    protocol=protocol,
    sample_rate=2048000.0  # 2x subcarrier for Nyquist
)

# 4. Generate custom phase demodulator
pm_demod = generate_sync_block(
    name="pm_demod",
    description="Apollo PM demodulator for 0.133 rad deviation",
    inputs=[{"dtype": "complex", "vlen": 1}],
    outputs=[{"dtype": "float", "vlen": 1}],
    parameters=[
        {"name": "deviation", "dtype": "float", "default": 0.133}
    ],
    work_logic="""
        # Extract instantaneous phase
        phase = numpy.angle(input_items[0])
        # Differentiate to get PM signal
        output_items[0][:] = numpy.diff(phase, prepend=phase[0]) * self.deviation
    """
)

# 5. Validate the generated block
validation = validate_block_code(pm_demod.source_code)
if not validation.is_valid:
    print(f"Errors: {validation.errors}")

# 6. Test in Docker
test = test_block_in_docker(
    source_code=pm_demod.source_code,
    test_input=[1+0j, 0+1j, -1+0j, 0-1j],  # 90° phase steps
    timeout_seconds=30
)

# 7. Export to OOT module
export_block_to_oot(
    generated=pm_demod,
    module_name="apollo",
    output_dir="/home/user/gr-apollo",
    author="Ryan Malloy"
)

# 8. Build and install the module
install_oot_module(
    git_url="file:///home/user/gr-apollo",
    branch="main"
)

Three-Tier Development Model

GR-MCP supports three levels of block persistence:

Tier Mechanism Persistence Use Case
1 create_embedded_python_block() In .grc file Rapid iteration
2 validate_block_code() + flowgraph Memory only Session testing
3 export_block_to_oot() File system Distribution

Workflow Progression:

Tier 1: Rapid Iteration
    create_embedded_python_block() → modify → test → iterate
                     │
                     ▼ (satisfied with block)
Tier 2: Validation
    validate_block_code() → test_block_in_docker()
                     │
                     ▼ (ready for distribution)
Tier 3: Export
    export_block_to_oot() → install_oot_module()

Resources

Block dev mode provides prompt and template resources:

# Access via MCP resources
"prompts://block-generation/sync-block"      # gr.sync_block patterns
"prompts://block-generation/basic-block"     # gr.basic_block patterns
"prompts://protocol-analysis/decoder-chain"  # Decoder pipeline guidance
"templates://block/sync-block"               # Python code template
"templates://oot/cmake"                       # CMakeLists.txt template
"templates://oot/block-yaml"                  # .block.yml template

Tool Reference

Protocol Analysis Tools

Tool Description
parse_protocol_spec Parse natural language protocol spec → ProtocolModel
generate_decoder_chain ProtocolModel → complete decoder pipeline
get_missing_oot_modules Check which OOT modules are required

Signal Analysis Tools

Tool Description
analyze_iq_file FFT analysis of IQ recordings

Block Generation Tools

Tool Description
generate_sync_block Create 1:1 sample processing block
generate_basic_block Create variable-ratio block
generate_interp_block Create interpolating block
generate_decim_block Create decimating block
validate_block_code Static code analysis
test_block_in_docker Isolated container testing
parse_block_prompt Parse natural language → generation params

OOT Export Tools

Tool Description
generate_oot_skeleton Create empty module structure
export_block_to_oot Export generated block to OOT
export_from_flowgraph Export embedded block to OOT