diff --git a/src/gnuradio_mcp/middlewares/platform.py b/src/gnuradio_mcp/middlewares/platform.py index 4e43158..df0f2e8 100644 --- a/src/gnuradio_mcp/middlewares/platform.py +++ b/src/gnuradio_mcp/middlewares/platform.py @@ -1,5 +1,8 @@ from __future__ import annotations +import os +from pathlib import Path + from gnuradio.grc.core.platform import Platform from gnuradio_mcp.middlewares.base import ElementMiddleware @@ -11,6 +14,7 @@ class PlatformMiddleware(ElementMiddleware): def __init__(self, platform: Platform): super().__init__(platform) self._platform = self._element + self._oot_paths: list[str] = [] @property def blocks(self) -> list[BlockTypeModel]: @@ -19,6 +23,70 @@ class PlatformMiddleware(ElementMiddleware): for block in self._platform.blocks.values() ] + @property + def default_block_paths(self) -> list[str]: + """Get the default block paths from Platform.Config.""" + return list(self._platform.config.block_paths) + + @property + def oot_paths(self) -> list[str]: + """Get the currently loaded OOT paths.""" + return self._oot_paths.copy() + + def load_oot_paths(self, paths: list[str]) -> dict: + """Load OOT (Out-of-Tree) block paths into the platform. + + Since Platform.build_library() does a full reset (clears all blocks), + we must rebuild with default_paths + oot_paths combined. + + Args: + paths: List of directory paths containing .block.yml files + + Returns: + dict with: + - added_paths: List of valid paths that were added + - invalid_paths: List of paths that don't exist + - blocks_before: Block count before reload + - blocks_after: Block count after reload + """ + blocks_before = len(self._platform.blocks) + + # Validate paths exist + valid_paths = [] + invalid_paths = [] + for path in paths: + expanded = os.path.expanduser(path) + if Path(expanded).is_dir(): + valid_paths.append(expanded) + else: + invalid_paths.append(path) + + if not valid_paths: + return { + "added_paths": [], + "invalid_paths": invalid_paths, + "blocks_before": blocks_before, + "blocks_after": blocks_before, + } + + # Combine default paths with OOT paths + combined_paths = self.default_block_paths + valid_paths + + # Rebuild the library with all paths + self._platform.build_library(path=combined_paths) + + # Track the OOT paths we've loaded + self._oot_paths = valid_paths + + blocks_after = len(self._platform.blocks) + + return { + "added_paths": valid_paths, + "invalid_paths": invalid_paths, + "blocks_before": blocks_before, + "blocks_after": blocks_after, + } + def make_flowgraph(self, filepath: str = "") -> FlowGraphMiddleware: return FlowGraphMiddleware.from_file(self, filepath) diff --git a/src/gnuradio_mcp/providers/base.py b/src/gnuradio_mcp/providers/base.py index c91c617..f677668 100644 --- a/src/gnuradio_mcp/providers/base.py +++ b/src/gnuradio_mcp/providers/base.py @@ -101,3 +101,27 @@ class PlatformProvider: def save_flowgraph(self, filepath: str) -> bool: self._platform_mw.save_flowgraph(filepath, self._flowgraph_mw) return True + + def load_oot_blocks(self, paths: List[str]) -> Dict[str, Any]: + """Load OOT (Out-of-Tree) block paths into the platform. + + OOT modules are third-party GNU Radio blocks installed separately. + They may be installed to: + - /usr/share/gnuradio/grc/blocks (system-wide via package manager) + - /usr/local/share/gnuradio/grc/blocks (locally-built) + - Custom paths specified by the user + + Since Platform.build_library() does a full reset, this method + combines the default block paths with the OOT paths and rebuilds. + + Args: + paths: List of directory paths containing .block.yml files + + Returns: + dict with: + - added_paths: List of valid paths that were added + - invalid_paths: List of paths that don't exist + - blocks_before: Block count before reload + - blocks_after: Block count after reload + """ + return self._platform_mw.load_oot_paths(paths) diff --git a/src/gnuradio_mcp/providers/mcp.py b/src/gnuradio_mcp/providers/mcp.py index a0e67da..c75e6b6 100644 --- a/src/gnuradio_mcp/providers/mcp.py +++ b/src/gnuradio_mcp/providers/mcp.py @@ -26,6 +26,7 @@ class McpPlatformProvider: self._mcp_instance.tool(self._platform_provider.get_all_errors) self._mcp_instance.tool(self._platform_provider.save_flowgraph) self._mcp_instance.tool(self._platform_provider.get_all_available_blocks) + self._mcp_instance.tool(self._platform_provider.load_oot_blocks) @property def app(self) -> FastMCP: diff --git a/tests/unit/test_oot_blocks.py b/tests/unit/test_oot_blocks.py new file mode 100644 index 0000000..7791602 --- /dev/null +++ b/tests/unit/test_oot_blocks.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import tempfile + +from gnuradio.grc.core.platform import Platform + +from gnuradio_mcp.middlewares.platform import PlatformMiddleware +from gnuradio_mcp.providers.base import PlatformProvider + + +class TestPlatformMiddlewareOOT: + """Tests for OOT (Out-of-Tree) block path loading.""" + + def test_default_block_paths_property(self, platform: Platform): + """Verify we can access default block paths from Platform.Config.""" + middleware = PlatformMiddleware(platform) + default_paths = middleware.default_block_paths + + assert isinstance(default_paths, list) + assert len(default_paths) > 0 + # Should include the system blocks path + assert any("gnuradio" in path for path in default_paths) + + def test_oot_paths_initially_empty(self, platform: Platform): + """Verify OOT paths are empty on fresh middleware.""" + middleware = PlatformMiddleware(platform) + assert middleware.oot_paths == [] + + def test_load_oot_paths_with_invalid_path(self, platform: Platform): + """Verify invalid paths are reported correctly.""" + middleware = PlatformMiddleware(platform) + blocks_before = len(middleware.blocks) + + result = middleware.load_oot_paths(["/nonexistent/path/to/blocks"]) + + assert result["added_paths"] == [] + assert result["invalid_paths"] == ["/nonexistent/path/to/blocks"] + assert result["blocks_before"] == blocks_before + assert result["blocks_after"] == blocks_before + assert middleware.oot_paths == [] + + def test_load_oot_paths_with_empty_directory(self, platform: Platform): + """Verify loading an empty directory doesn't break anything.""" + middleware = PlatformMiddleware(platform) + blocks_before = len(middleware.blocks) + + with tempfile.TemporaryDirectory() as tmpdir: + result = middleware.load_oot_paths([tmpdir]) + + assert result["added_paths"] == [tmpdir] + assert result["invalid_paths"] == [] + assert result["blocks_before"] == blocks_before + # Should have same or fewer blocks (empty dir adds nothing) + assert result["blocks_after"] <= blocks_before + assert middleware.oot_paths == [tmpdir] + + def test_load_oot_paths_with_mixed_valid_invalid(self, platform: Platform): + """Verify mixed valid/invalid paths are handled correctly.""" + middleware = PlatformMiddleware(platform) + + with tempfile.TemporaryDirectory() as tmpdir: + result = middleware.load_oot_paths([tmpdir, "/nonexistent/path"]) + + assert result["added_paths"] == [tmpdir] + assert result["invalid_paths"] == ["/nonexistent/path"] + assert middleware.oot_paths == [tmpdir] + + def test_load_oot_paths_expands_tilde(self, platform: Platform): + """Verify ~ is expanded to home directory.""" + middleware = PlatformMiddleware(platform) + + # This should either fail validation (if path doesn't exist) + # or succeed (if it does) - either way, it shouldn't error + result = middleware.load_oot_paths(["~/nonexistent_oot_test_dir"]) + + # The path should be in invalid_paths since it doesn't exist + assert "~/nonexistent_oot_test_dir" in result["invalid_paths"] + + def test_load_oot_paths_with_system_blocks_path(self, platform: Platform): + """Verify we can reload with the system blocks path (idempotent test).""" + middleware = PlatformMiddleware(platform) + + # Get a default path and use it as "OOT" (should be no-op essentially) + default_paths = middleware.default_block_paths + if default_paths: + result = middleware.load_oot_paths([default_paths[0]]) + + # Should have roughly the same number of blocks + # (might be slightly different due to how GRC handles duplicates) + assert result["blocks_after"] > 0 + assert result["added_paths"] == [default_paths[0]] + + +class TestPlatformProviderOOT: + """Tests for OOT block loading via PlatformProvider.""" + + def test_load_oot_blocks_method_exists( + self, platform_middleware: PlatformMiddleware + ): + """Verify the load_oot_blocks method is available on provider.""" + provider = PlatformProvider(platform_middleware) + assert hasattr(provider, "load_oot_blocks") + assert callable(provider.load_oot_blocks) + + def test_load_oot_blocks_returns_dict( + self, platform_middleware: PlatformMiddleware + ): + """Verify load_oot_blocks returns expected structure.""" + provider = PlatformProvider(platform_middleware) + + result = provider.load_oot_blocks(["/nonexistent/path"]) + + assert isinstance(result, dict) + assert "added_paths" in result + assert "invalid_paths" in result + assert "blocks_before" in result + assert "blocks_after" in result + + def test_load_oot_blocks_with_valid_path( + self, platform_middleware: PlatformMiddleware + ): + """Verify load_oot_blocks works with a valid directory.""" + provider = PlatformProvider(platform_middleware) + + with tempfile.TemporaryDirectory() as tmpdir: + result = provider.load_oot_blocks([tmpdir]) + + assert tmpdir in result["added_paths"] + assert result["invalid_paths"] == []