"""OOT module export middleware. Exports embedded Python blocks to full OOT module structure with CMakeLists.txt, .block.yml, and Python package layout. """ from __future__ import annotations import ast import logging import os import re from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Any from gnuradio_mcp.models import ( GeneratedBlockCode, OOTExportResult, OOTSkeletonResult, ) if TYPE_CHECKING: from gnuradio_mcp.middlewares.flowgraph import FlowGraphMiddleware logger = logging.getLogger(__name__) # ────────────────────────────────────────────── # File Templates # ────────────────────────────────────────────── ROOT_CMAKELISTS_TEMPLATE = '''cmake_minimum_required(VERSION 3.8) project({module_name} CXX) # Select the release build type by default set(CMAKE_BUILD_TYPE "Release") # Make sure our local CMake Modules path comes first list(INSERT CMAKE_MODULE_PATH 0 ${{CMAKE_SOURCE_DIR}}/cmake/Modules) # Find GNU Radio find_package(Gnuradio "3.10" REQUIRED) # Setup GNU Radio install directories include(GrVersion) # Set component set(GR_{module_upper}_INCLUDE_DIRS ${{CMAKE_CURRENT_SOURCE_DIR}}/include) set(GR_PKG_DOC_DIR ${{GR_DOC_DIR}}/${{CMAKE_PROJECT_NAME}}-${{VERSION_INFO_MAJOR}}.${{VERSION_INFO_API}}.${{VERSION_INFO_MINOR}}) ######################################################################## # Setup the include and linker paths ######################################################################## include_directories( ${{CMAKE_SOURCE_DIR}}/lib ${{CMAKE_SOURCE_DIR}}/include ${{CMAKE_BINARY_DIR}}/lib ${{CMAKE_BINARY_DIR}}/include ${{Boost_INCLUDE_DIRS}} ${{GNURADIO_ALL_INCLUDE_DIRS}} ) link_directories( ${{Boost_LIBRARY_DIRS}} ${{GNURADIO_RUNTIME_LIBRARY_DIRS}} ) ######################################################################## # Create uninstall target ######################################################################## configure_file( ${{CMAKE_SOURCE_DIR}}/cmake/cmake_uninstall.cmake.in ${{CMAKE_CURRENT_BINARY_DIR}}/cmake_uninstall.cmake @ONLY) add_custom_target(uninstall ${{CMAKE_COMMAND}} -P ${{CMAKE_CURRENT_BINARY_DIR}}/cmake_uninstall.cmake ) ######################################################################## # Add subdirectories ######################################################################## add_subdirectory(python/{module_name}) add_subdirectory(grc) ######################################################################## # Install cmake search helper ######################################################################## install(FILES cmake/Modules/{module_name}Config.cmake DESTINATION ${{GR_LIBRARY_DIR}}/cmake/{module_name} ) ''' PYTHON_CMAKELISTS_TEMPLATE = '''# Copyright {year} {author} # SPDX-License-Identifier: GPL-3.0-or-later ######################################################################## # Include python install macros ######################################################################## include(GrPython) ######################################################################## # Install python sources ######################################################################## GR_PYTHON_INSTALL( FILES __init__.py {block_files} DESTINATION ${{GR_PYTHON_DIR}}/{module_name} ) ######################################################################## # Handle the unit tests ######################################################################## include(GrTest) if(ENABLE_TESTING) set(GR_TEST_TARGET_DEPS gnuradio-{module_name}) GR_ADD_TEST(qa_{module_name} ${{PYTHON_EXECUTABLE}} -B ${{CMAKE_CURRENT_SOURCE_DIR}}/qa_{module_name}.py) endif() ''' GRC_CMAKELISTS_TEMPLATE = '''# Copyright {year} {author} # SPDX-License-Identifier: GPL-3.0-or-later install(FILES {yml_files} DESTINATION ${{GR_DATA_DIR}}/grc/blocks ) ''' PYTHON_INIT_TEMPLATE = '''# # Copyright {year} {author} # SPDX-License-Identifier: GPL-3.0-or-later # """ GNU Radio {module_name} module Generated by gr-mcp """ from importlib import import_module # Import OOT blocks {imports} ''' BLOCK_YML_TEMPLATE = '''id: {module_name}_{block_name} label: {block_label} category: [{module_name}] templates: imports: from {module_name} import {block_name} make: {module_name}.{block_name}({make_args}) {callbacks} parameters: {parameters} inputs: {inputs} outputs: {outputs} documentation: |- {documentation} file_format: 1 ''' CMAKE_UNINSTALL_TEMPLATE = '''if(NOT EXISTS "@CMAKE_CURRENT_BINARY_DIR@/install_manifest.txt") message(FATAL_ERROR "Cannot find install manifest: @CMAKE_CURRENT_BINARY_DIR@/install_manifest.txt") endif(NOT EXISTS "@CMAKE_CURRENT_BINARY_DIR@/install_manifest.txt") file(READ "@CMAKE_CURRENT_BINARY_DIR@/install_manifest.txt" files) string(REGEX REPLACE "\\n" ";" files "${files}") foreach(file ${files}) message(STATUS "Uninstalling $ENV{DESTDIR}${file}") if(IS_SYMLINK "$ENV{DESTDIR}${file}" OR EXISTS "$ENV{DESTDIR}${file}") exec_program( "@CMAKE_COMMAND@" ARGS "-E remove \\"$ENV{DESTDIR}${file}\\"" OUTPUT_VARIABLE rm_out RETURN_VALUE rm_retval ) if(NOT "${rm_retval}" STREQUAL 0) message(FATAL_ERROR "Problem when removing $ENV{DESTDIR}${file}") endif(NOT "${rm_retval}" STREQUAL 0) else(IS_SYMLINK "$ENV{DESTDIR}${file}" OR EXISTS "$ENV{DESTDIR}${file}") message(STATUS "File $ENV{DESTDIR}${file} does not exist.") endif(IS_SYMLINK "$ENV{DESTDIR}${file}" OR EXISTS "$ENV{DESTDIR}${file}") endforeach(file) ''' CMAKE_CONFIG_TEMPLATE = '''find_package(PkgConfig) PKG_CHECK_MODULES(PC_{module_upper} QUIET {module_name}) if(PC_{module_upper}_FOUND) set({module_upper}_FOUND TRUE) set({module_upper}_INCLUDE_DIRS ${{PC_{module_upper}_INCLUDE_DIRS}}) set({module_upper}_LIBRARIES ${{PC_{module_upper}_LIBRARIES}}) else() set({module_upper}_FOUND FALSE) endif() include(FindPackageHandleStandardArgs) find_package_handle_standard_args({module_name} DEFAULT_MSG {module_upper}_FOUND) mark_as_advanced({module_upper}_INCLUDE_DIRS {module_upper}_LIBRARIES) ''' class OOTExporterMiddleware: """Exports embedded blocks to full OOT module structure. Creates a gr_modtool-compatible directory structure that can be built and installed as a standard GNU Radio OOT module. """ def __init__(self, flowgraph_mw: FlowGraphMiddleware | None = None): """Initialize the OOT exporter. Args: flowgraph_mw: Optional flowgraph middleware for accessing blocks. """ self._flowgraph_mw = flowgraph_mw # ────────────────────────────────────────── # Module Skeleton Generation # ────────────────────────────────────────── def generate_oot_skeleton( self, module_name: str, output_dir: str, author: str = "gr-mcp", description: str = "", ) -> OOTSkeletonResult: """Generate an empty OOT module structure. Creates the directory structure and CMake files for a new GNU Radio OOT module. Blocks can be added later. Args: module_name: Module name (e.g., "custom" for gr-custom) output_dir: Base directory for the module author: Author name for copyright headers description: Module description Returns: OOTSkeletonResult with paths and structure info. """ module_name = self._sanitize_module_name(module_name) module_upper = module_name.upper() year = datetime.now().year # Create directory structure base_path = Path(output_dir) dirs = { "root": base_path, "cmake": base_path / "cmake" / "Modules", "grc": base_path / "grc", "python": base_path / "python" / module_name, "include": base_path / "include" / module_name, "lib": base_path / "lib", } try: for d in dirs.values(): d.mkdir(parents=True, exist_ok=True) files_created: dict[str, list[str]] = {k: [] for k in dirs.keys()} # Root CMakeLists.txt root_cmake = dirs["root"] / "CMakeLists.txt" root_cmake.write_text(ROOT_CMAKELISTS_TEMPLATE.format( module_name=module_name, module_upper=module_upper, )) files_created["root"].append("CMakeLists.txt") # cmake/Modules/{module_name}Config.cmake config_cmake = dirs["cmake"] / f"{module_name}Config.cmake" config_cmake.write_text(CMAKE_CONFIG_TEMPLATE.format( module_name=module_name, module_upper=module_upper, )) files_created["cmake"].append(f"{module_name}Config.cmake") # cmake/cmake_uninstall.cmake.in uninstall_cmake = dirs["root"] / "cmake" / "cmake_uninstall.cmake.in" uninstall_cmake.write_text(CMAKE_UNINSTALL_TEMPLATE) files_created["cmake"].append("cmake_uninstall.cmake.in") # python/{module_name}/__init__.py init_py = dirs["python"] / "__init__.py" init_py.write_text(PYTHON_INIT_TEMPLATE.format( year=year, author=author, module_name=module_name, imports="", )) files_created["python"].append("__init__.py") # python/{module_name}/CMakeLists.txt python_cmake = dirs["python"] / "CMakeLists.txt" python_cmake.write_text(PYTHON_CMAKELISTS_TEMPLATE.format( year=year, author=author, module_name=module_name, block_files="", )) files_created["python"].append("CMakeLists.txt") # grc/CMakeLists.txt grc_cmake = dirs["grc"] / "CMakeLists.txt" grc_cmake.write_text(GRC_CMAKELISTS_TEMPLATE.format( year=year, author=author, yml_files="", )) files_created["grc"].append("CMakeLists.txt") # README.md readme = dirs["root"] / "README.md" readme.write_text(f"# gr-{module_name}\n\n{description}\n\nGenerated by gr-mcp.\n") files_created["root"].append("README.md") return OOTSkeletonResult( success=True, module_name=module_name, output_dir=str(base_path), structure=files_created, next_steps=[ "Add blocks using export_block_to_oot()", "Build with: mkdir build && cd build && cmake .. && make", "Install with: sudo make install", ], ) except Exception as e: logger.exception("Failed to generate OOT skeleton") return OOTSkeletonResult( success=False, module_name=module_name, output_dir=str(base_path), next_steps=[f"Error: {e}"], ) # ────────────────────────────────────────── # Block Export # ────────────────────────────────────────── def export_block_to_oot( self, generated: GeneratedBlockCode, module_name: str, output_dir: str, author: str = "gr-mcp", ) -> OOTExportResult: """Export a generated block to an OOT module. Creates or updates an OOT module with the given block. If the module doesn't exist, creates the skeleton first. Args: generated: GeneratedBlockCode from block generator module_name: Module name (e.g., "custom") output_dir: Base directory for the module author: Author name for copyright headers Returns: OOTExportResult with file paths and status. """ module_name = self._sanitize_module_name(module_name) block_name = self._sanitize_block_name(generated.block_name) year = datetime.now().year base_path = Path(output_dir) files_created: list[str] = [] try: # Create skeleton if needed if not (base_path / "CMakeLists.txt").exists(): skeleton = self.generate_oot_skeleton(module_name, output_dir, author) if not skeleton.success: return OOTExportResult( success=False, module_name=module_name, block_name=block_name, output_dir=str(base_path), error="Failed to create module skeleton", ) files_created.extend( [f"{k}/{f}" for k, files in skeleton.structure.items() for f in files] ) # Write block Python file python_dir = base_path / "python" / module_name block_file = python_dir / f"{block_name}.py" block_file.write_text(generated.source_code) files_created.append(f"python/{module_name}/{block_name}.py") # Update __init__.py self._update_init_file(python_dir, block_name) # Update python/CMakeLists.txt self._update_python_cmake(python_dir, block_name) # Generate and write .block.yml grc_dir = base_path / "grc" yml_content = self._generate_block_yml( generated, module_name, block_name ) yml_file = grc_dir / f"{module_name}_{block_name}.block.yml" yml_file.write_text(yml_content) files_created.append(f"grc/{module_name}_{block_name}.block.yml") # Update grc/CMakeLists.txt self._update_grc_cmake(grc_dir, module_name, block_name) return OOTExportResult( success=True, module_name=module_name, block_name=block_name, output_dir=str(base_path), files_created=files_created, build_ready=True, ) except Exception as e: logger.exception("Failed to export block") return OOTExportResult( success=False, module_name=module_name, block_name=block_name, output_dir=str(base_path), error=str(e), ) def export_from_flowgraph( self, block_name: str, module_name: str, output_dir: str, author: str = "gr-mcp", ) -> OOTExportResult: """Export an embedded block from the current flowgraph. Extracts the source code from an epy_block in the flowgraph and exports it to a full OOT module. Args: block_name: Name of the epy_block in the flowgraph module_name: Target module name output_dir: Base directory for the module author: Author name Returns: OOTExportResult with file paths and status. """ if self._flowgraph_mw is None: return OOTExportResult( success=False, module_name=module_name, block_name=block_name, output_dir=output_dir, error="No flowgraph middleware configured", ) try: # Get block from flowgraph block_mw = self._flowgraph_mw.get_block(block_name) block = block_mw._block # Check it's an epy_block if block.key != "epy_block": return OOTExportResult( success=False, module_name=module_name, block_name=block_name, output_dir=output_dir, error=f"Block {block_name} is not an epy_block (type: {block.key})", ) # Extract source code source_code = block.params["_source_code"].get_value() # Parse to get block info from gnuradio_mcp.models import SignatureItem generated = GeneratedBlockCode( source_code=source_code, block_name=block_name, block_class="sync_block", inputs=[SignatureItem(dtype="float", vlen=1)], outputs=[SignatureItem(dtype="float", vlen=1)], is_valid=True, ) return self.export_block_to_oot(generated, module_name, output_dir, author) except Exception as e: logger.exception("Failed to export from flowgraph") return OOTExportResult( success=False, module_name=module_name, block_name=block_name, output_dir=output_dir, error=str(e), ) # ────────────────────────────────────────── # Block YAML Generation # ────────────────────────────────────────── def _generate_block_yml( self, generated: GeneratedBlockCode, module_name: str, block_name: str, ) -> str: """Generate .block.yml content for a block.""" # Parse source to extract info block_info = self._parse_block_source(generated.source_code) # Build label block_label = block_name.replace("_", " ").title() # Build make args make_args = ", ".join([ f"{p.name}=${{{p.name}}}" for p in generated.parameters ]) if generated.parameters else "" # Build callbacks section callbacks = "" if block_info.get("callbacks"): callbacks = "callbacks:\n" + "\n".join([ f" - set_{name}(${{{name}}})" for name in block_info["callbacks"] ]) # Build parameters section params_yml = "" for p in generated.parameters: params_yml += f"""- id: {p.name} label: {p.name.replace('_', ' ').title()} dtype: {self._python_to_grc_dtype(p.dtype)} default: {repr(p.default)} """ # Build inputs section inputs_yml = "" for i, inp in enumerate(generated.inputs): inputs_yml += f"""- label: in{i} domain: stream dtype: {inp.dtype} vlen: {inp.vlen} """ # Build outputs section outputs_yml = "" for i, out in enumerate(generated.outputs): outputs_yml += f"""- label: out{i} domain: stream dtype: {out.dtype} vlen: {out.vlen} """ # Documentation doc = block_info.get("doc", generated.generation_prompt or "Custom block") return BLOCK_YML_TEMPLATE.format( module_name=module_name, block_name=block_name, block_label=block_label, make_args=make_args, callbacks=callbacks, parameters=params_yml or "[]", inputs=inputs_yml or "[]", outputs=outputs_yml or "[]", documentation=doc.replace("\n", "\n "), ) def _parse_block_source(self, source_code: str) -> dict[str, Any]: """Parse block source to extract metadata.""" result: dict[str, Any] = { "class_name": "blk", "base_class": "gr.sync_block", "doc": "", "callbacks": [], } try: tree = ast.parse(source_code) for node in ast.walk(tree): if isinstance(node, ast.ClassDef): result["class_name"] = node.name # Get docstring if (node.body and isinstance(node.body[0], ast.Expr) and isinstance(node.body[0].value, ast.Constant)): result["doc"] = node.body[0].value.value # Get base class if node.bases: result["base_class"] = ast.unparse(node.bases[0]) # Find setter methods (callbacks) for item in node.body: if isinstance(item, ast.FunctionDef): if item.name.startswith("set_"): param = item.name[4:] # Remove "set_" prefix result["callbacks"].append(param) break except Exception as e: logger.warning(f"Failed to parse block source: {e}") return result # ────────────────────────────────────────── # File Update Helpers # ────────────────────────────────────────── def _update_init_file(self, python_dir: Path, block_name: str): """Update __init__.py to import the new block.""" init_file = python_dir / "__init__.py" content = init_file.read_text() import_line = f"from .{block_name} import blk as {block_name}" if import_line not in content: # Add import if "# Import OOT blocks" in content: content = content.replace( "# Import OOT blocks", f"# Import OOT blocks\n{import_line}" ) else: content += f"\n{import_line}\n" init_file.write_text(content) def _update_python_cmake(self, python_dir: Path, block_name: str): """Update python/CMakeLists.txt to include the new block.""" cmake_file = python_dir / "CMakeLists.txt" content = cmake_file.read_text() block_entry = f" {block_name}.py" if block_entry not in content: # Find FILES section and add content = re.sub( r"(FILES\s+\n\s+__init__\.py)", f"\\1\n{block_entry}", content, ) cmake_file.write_text(content) def _update_grc_cmake(self, grc_dir: Path, module_name: str, block_name: str): """Update grc/CMakeLists.txt to include the new .block.yml.""" cmake_file = grc_dir / "CMakeLists.txt" content = cmake_file.read_text() yml_entry = f" {module_name}_{block_name}.block.yml" if yml_entry not in content: # Find install(FILES section if "install(FILES" in content: content = re.sub( r"(install\(FILES\s*\n)", f"\\1{yml_entry}\n", content, ) else: content = f"""install(FILES {yml_entry} DESTINATION ${{GR_DATA_DIR}}/grc/blocks ) """ cmake_file.write_text(content) # ────────────────────────────────────────── # Utility Methods # ────────────────────────────────────────── def _sanitize_module_name(self, name: str) -> str: """Sanitize module name for valid Python/CMake identifiers.""" # Remove gr- prefix if present if name.lower().startswith("gr-"): name = name[3:] # Replace invalid characters name = re.sub(r"[^a-zA-Z0-9_]", "_", name) # Ensure starts with letter if name and name[0].isdigit(): name = "m" + name return name.lower() def _sanitize_block_name(self, name: str) -> str: """Sanitize block name for valid Python identifier.""" # Remove any numeric suffix (e.g., _0) name = re.sub(r"_\d+$", "", name) # Replace invalid characters name = re.sub(r"[^a-zA-Z0-9_]", "_", name) # Ensure starts with letter if name and name[0].isdigit(): name = "b" + name return name.lower() def _python_to_grc_dtype(self, dtype: str) -> str: """Convert Python type to GRC dtype string.""" mapping = { "float": "real", "int": "int", "str": "string", "bool": "bool", "complex": "complex", } return mapping.get(dtype, "raw")