feat: add OOT (Out-of-Tree) block path loading support
Add load_oot_blocks MCP tool to dynamically load GNU Radio OOT modules. Since GRC's Platform.build_library() clears all blocks on every call, the implementation combines default block paths with OOT paths before rebuilding. This allows loading custom blocks from: - /usr/local/share/gnuradio/grc/blocks (locally-built) - Custom user-specified directories Implementation: - PlatformMiddleware.load_oot_paths(): validates paths, combines with defaults, rebuilds library - PlatformProvider.load_oot_blocks(): exposes method to MCP layer - McpPlatformProvider: registers load_oot_blocks tool Returns useful diagnostics: added_paths, invalid_paths, and block counts before/after reload.
This commit is contained in:
parent
fdcbffba1a
commit
e9ac115728
@ -1,5 +1,8 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from gnuradio.grc.core.platform import Platform
|
from gnuradio.grc.core.platform import Platform
|
||||||
|
|
||||||
from gnuradio_mcp.middlewares.base import ElementMiddleware
|
from gnuradio_mcp.middlewares.base import ElementMiddleware
|
||||||
@ -11,6 +14,7 @@ class PlatformMiddleware(ElementMiddleware):
|
|||||||
def __init__(self, platform: Platform):
|
def __init__(self, platform: Platform):
|
||||||
super().__init__(platform)
|
super().__init__(platform)
|
||||||
self._platform = self._element
|
self._platform = self._element
|
||||||
|
self._oot_paths: list[str] = []
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def blocks(self) -> list[BlockTypeModel]:
|
def blocks(self) -> list[BlockTypeModel]:
|
||||||
@ -19,6 +23,70 @@ class PlatformMiddleware(ElementMiddleware):
|
|||||||
for block in self._platform.blocks.values()
|
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:
|
def make_flowgraph(self, filepath: str = "") -> FlowGraphMiddleware:
|
||||||
return FlowGraphMiddleware.from_file(self, filepath)
|
return FlowGraphMiddleware.from_file(self, filepath)
|
||||||
|
|
||||||
|
|||||||
@ -101,3 +101,27 @@ class PlatformProvider:
|
|||||||
def save_flowgraph(self, filepath: str) -> bool:
|
def save_flowgraph(self, filepath: str) -> bool:
|
||||||
self._platform_mw.save_flowgraph(filepath, self._flowgraph_mw)
|
self._platform_mw.save_flowgraph(filepath, self._flowgraph_mw)
|
||||||
return True
|
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)
|
||||||
|
|||||||
@ -26,6 +26,7 @@ class McpPlatformProvider:
|
|||||||
self._mcp_instance.tool(self._platform_provider.get_all_errors)
|
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.save_flowgraph)
|
||||||
self._mcp_instance.tool(self._platform_provider.get_all_available_blocks)
|
self._mcp_instance.tool(self._platform_provider.get_all_available_blocks)
|
||||||
|
self._mcp_instance.tool(self._platform_provider.load_oot_blocks)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def app(self) -> FastMCP:
|
def app(self) -> FastMCP:
|
||||||
|
|||||||
129
tests/unit/test_oot_blocks.py
Normal file
129
tests/unit/test_oot_blocks.py
Normal file
@ -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"] == []
|
||||||
Loading…
x
Reference in New Issue
Block a user