Ryan Malloy
5db7d71d2b
Implements complete workflow for generating GNU Radio blocks from descriptions: Block Generation: - generate_sync_block, generate_basic_block, generate_interp_block, generate_decim_block tools for creating different block types - Template-based code generation with customizable work logic - Automatic validation via AST parsing and signature checking Protocol Analysis: - Parse protocol specifications into structured models - Generate decoder pipelines matching modulation to demodulator blocks - Templates for BLE, Zigbee, LoRa, POCSAG, ADS-B protocols OOT Export: - Export generated blocks to full OOT module structure - Generate CMakeLists.txt, block YAML, Python modules - gr_modtool-compatible output Dynamic Tool Registration: - enable_block_dev_mode/disable_block_dev_mode for context management - Tools only registered when needed (reduces LLM context usage) Includes comprehensive test coverage and end-to-end demo.
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import pytest
|
|
from fastmcp import Client
|
|
|
|
from main import app as mcp_app
|
|
|
|
|
|
@pytest.fixture
|
|
async def main_mcp_client():
|
|
async with Client(mcp_app) as mcp_client:
|
|
yield mcp_client
|
|
|
|
|
|
async def test_get_blocks(main_mcp_client: Client):
|
|
"""Test retrieving available blocks."""
|
|
blocks = await main_mcp_client.call_tool(name="get_blocks")
|
|
assert blocks.data is not None
|
|
# We don't know exactly what blocks are there, but should get a list
|
|
assert isinstance(blocks.data, list)
|
|
|
|
|
|
async def test_make_and_remove_block(main_mcp_client: Client):
|
|
"""Test making a block and then removing it."""
|
|
block_type = "analog_sig_source_x"
|
|
|
|
# helper to check if block exists in the flowgraph
|
|
async def get_block_names():
|
|
current_blocks = await main_mcp_client.call_tool(name="get_blocks")
|
|
# FastMCP may return dicts or Pydantic models depending on serialization
|
|
return [b["name"] if isinstance(b, dict) else b.name for b in current_blocks.data]
|
|
|
|
# 1. Create a block
|
|
result = await main_mcp_client.call_tool(
|
|
name="make_block", arguments={"block_name": block_type}
|
|
)
|
|
assert result.data is not None
|
|
# The output is the new block name, likely something like "analog_sig_source_x_0"
|
|
new_block_name = str(result.data)
|
|
assert block_type in new_block_name
|
|
|
|
# Verify it exists
|
|
block_names = await get_block_names()
|
|
assert new_block_name in block_names
|
|
|
|
# 2. Remove the block
|
|
remove_result = await main_mcp_client.call_tool(
|
|
name="remove_block", arguments={"block_name": new_block_name}
|
|
)
|
|
assert remove_result.data is True
|
|
|
|
# Verify it's gone
|
|
block_names = await get_block_names()
|
|
assert new_block_name not in block_names
|