diff --git a/src/gnuradio_mcp/middlewares/flowgraph.py b/src/gnuradio_mcp/middlewares/flowgraph.py new file mode 100644 index 0000000..42ab8f6 --- /dev/null +++ b/src/gnuradio_mcp/middlewares/flowgraph.py @@ -0,0 +1,20 @@ + +from typing import List, Optional +from gnuradio.grc.core.FlowGraph import FlowGraph +from gnuradio_mcp.models import BlockModel + + +class FlowGraphMiddleware: + def __init__(self, flowgraph: FlowGraph): + self._flowgraph = flowgraph + + @property + def blocks(self) -> List[BlockModel]: + return [ + BlockModel(key=block.key, label=block.label) + for block in self._flowgraph.blocks + ] + + def add_block(self, key: str, name) -> None: + block = self._flowgraph.new_block(key) + block.params["id"].set_value(name) diff --git a/src/gnuradio_mcp/middlewares/platform.py b/src/gnuradio_mcp/middlewares/platform.py index 600d20e..523e866 100644 --- a/src/gnuradio_mcp/middlewares/platform.py +++ b/src/gnuradio_mcp/middlewares/platform.py @@ -1,12 +1,19 @@ -from typing import List +from typing import List, Optional from gnuradio.grc.core.platform import Platform - +from gnuradio.grc.core.FlowGraph import FlowGraph +from gnuradio_mcp.middlewares.flowgraph import FlowGraphMiddleware from gnuradio_mcp.models import BlockModel class PlatformMiddleware: def __init__(self, platform: Platform): self._platform = platform + flowgraph = self._platform.make_flow_graph("") + self._flowgraph_mw = FlowGraphMiddleware(flowgraph) @property def blocks(self) -> List[BlockModel]: return [BlockModel.from_block(block) for block in self._platform.blocks.values()] + + @property + def flowgraph(self) -> FlowGraphMiddleware: + return self._flowgraph_mw diff --git a/tests/test_flowgraph.py b/tests/test_flowgraph.py new file mode 100644 index 0000000..d16ad62 --- /dev/null +++ b/tests/test_flowgraph.py @@ -0,0 +1,42 @@ +import pytest +from gnuradio.grc.core.platform import Platform +from gnuradio_mcp.middlewares.flowgraph import FlowGraphMiddleware +from gnuradio import gr + +from gnuradio_mcp.models import BlockModel + +@pytest.fixture(scope="module") +def platform() -> Platform: + platform = Platform( + version=gr.version(), + version_parts=(gr.major_version(), gr.api_version(), gr.minor_version()), + prefs=gr.prefs(), + ) + platform.build_library() + return platform + +@pytest.fixture +def flowgraph_middleware(platform): + flowgraph = platform.make_flow_graph("") + return FlowGraphMiddleware(flowgraph) + +@pytest.fixture +def initial_blocks(flowgraph_middleware): + return [ + BlockModel(key=block.key, label=block.label) + for block in flowgraph_middleware._flowgraph.blocks + ] + +def test_blocks_match_initial_state(flowgraph_middleware, initial_blocks): + assert flowgraph_middleware.blocks == initial_blocks + +@pytest.mark.parametrize("block_index", [1, 2, 3]) +def test_add_block_preserves_and_adds(flowgraph_middleware, platform, initial_blocks, block_index): + block_keys = list(platform.blocks.keys()) + assert block_keys, "No blocks available in platform library." + block_key = block_keys[block_index] + block_name = "test_block" + flowgraph_middleware.add_block(block_key, block_name) + blocks = flowgraph_middleware.blocks + assert all(b in blocks for b in initial_blocks) + assert any(b.key == block_key for b in blocks)