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
14 KiB
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 |
Related Documentation
- GRC Runtime Communication - XML-RPC and ControlPort
- OOT Catalog - Curated OOT module directory