diff --git a/docs/development.md b/docs/development.md index 2dbc9eb..b94629b 100644 --- a/docs/development.md +++ b/docs/development.md @@ -61,8 +61,7 @@ kicad-mcp/ │ ├── kicad_utils.py # KiCad-specific functions │ ├── python_path.py # Python path setup for KiCad modules │ ├── drc_history.py # DRC history tracking -│ ├── env.py # Environment variable handling -│ └── logger.py # Logging utilities +│ └── env.py # Environment variable handling ``` ## Adding New Features @@ -254,8 +253,7 @@ For debugging, use: 1. The Python debugger (pdb) 2. Print statements to the console (captured in Claude Desktop logs) -3. Structured logging via the Logger class -4. The MCP Inspector tool +3. The MCP Inspector tool ## Performance Considerations diff --git a/kicad_mcp/context.py b/kicad_mcp/context.py index 02c4c2b..6a236bd 100644 --- a/kicad_mcp/context.py +++ b/kicad_mcp/context.py @@ -3,16 +3,12 @@ Lifespan context management for KiCad MCP Server. """ from contextlib import asynccontextmanager from dataclasses import dataclass -from typing import AsyncIterator, Optional, Dict, Any +from typing import AsyncIterator, Dict, Any from mcp.server.fastmcp import FastMCP -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.python_path import setup_kicad_python_path -# Create logger for this module -logger = Logger() - @dataclass class KiCadAppContext: """Type-safe context for KiCad MCP server.""" @@ -36,12 +32,12 @@ async def kicad_lifespan(server: FastMCP) -> AsyncIterator[KiCadAppContext]: Yields: KiCadAppContext: A typed context object shared across all handlers """ - logger.info("Starting KiCad MCP server initialization") + print("Starting KiCad MCP server initialization") # Initialize resources on startup - logger.info("Setting up KiCad Python modules") + print("Setting up KiCad Python modules") kicad_modules_available = setup_kicad_python_path() - logger.info(f"KiCad Python modules available: {kicad_modules_available}") + print(f"KiCad Python modules available: {kicad_modules_available}") # Create in-memory cache for expensive operations cache: Dict[str, Any] = {} @@ -53,37 +49,37 @@ async def kicad_lifespan(server: FastMCP) -> AsyncIterator[KiCadAppContext]: # Import any KiCad modules that should be preloaded if kicad_modules_available: try: - logger.info("Preloading KiCad Python modules") + print("Preloading KiCad Python modules") # Core PCB module used in multiple tools import pcbnew - logger.info(f"Successfully preloaded pcbnew module: {getattr(pcbnew, 'GetBuildVersion', lambda: 'unknown')()}") + print(f"Successfully preloaded pcbnew module: {getattr(pcbnew, 'GetBuildVersion', lambda: 'unknown')()}") cache["pcbnew_version"] = getattr(pcbnew, "GetBuildVersion", lambda: "unknown")() except ImportError as e: - logger.warning(f"Failed to preload some KiCad modules: {str(e)}") + print(f"Failed to preload some KiCad modules: {str(e)}") # Yield the context to the server - server runs during this time - logger.info("KiCad MCP server initialization complete") + print("KiCad MCP server initialization complete") yield KiCadAppContext( kicad_modules_available=kicad_modules_available, cache=cache ) finally: # Clean up resources when server shuts down - logger.info("Shutting down KiCad MCP server") + print("Shutting down KiCad MCP server") # Clear the cache if cache: - logger.debug(f"Clearing cache with {len(cache)} entries") + print(f"Clearing cache with {len(cache)} entries") cache.clear() # Clean up any temporary directories import shutil for temp_dir in created_temp_dirs: try: - logger.debug(f"Removing temporary directory: {temp_dir}") + print(f"Removing temporary directory: {temp_dir}") shutil.rmtree(temp_dir, ignore_errors=True) except Exception as e: - logger.error(f"Error cleaning up temporary directory {temp_dir}: {str(e)}") + print(f"Error cleaning up temporary directory {temp_dir}: {str(e)}") - logger.info("KiCad MCP server shutdown complete") + print("KiCad MCP server shutdown complete") diff --git a/kicad_mcp/resources/drc_resources.py b/kicad_mcp/resources/drc_resources.py index 2752dba..0660e69 100644 --- a/kicad_mcp/resources/drc_resources.py +++ b/kicad_mcp/resources/drc_resources.py @@ -2,19 +2,13 @@ Design Rule Check (DRC) resources for KiCad PCB files. """ import os -import json -import tempfile -from typing import Dict, Any, List + from mcp.server.fastmcp import FastMCP from kicad_mcp.utils.file_utils import get_project_files -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.drc_history import get_drc_history from kicad_mcp.tools.drc_impl.cli_drc import run_drc_via_cli -# Create logger for this module -logger = Logger() - def register_drc_resources(mcp: FastMCP) -> None: """Register DRC resources with the MCP server. @@ -32,7 +26,7 @@ def register_drc_resources(mcp: FastMCP) -> None: Returns: Markdown-formatted DRC history report """ - logger.info(f"Generating DRC history report for project: {project_path}") + print(f"Generating DRC history report for project: {project_path}") if not os.path.exists(project_path): return f"Project not found: {project_path}" @@ -151,7 +145,7 @@ def register_drc_resources(mcp: FastMCP) -> None: Returns: Markdown-formatted DRC report """ - logger.info(f"Generating DRC report for project: {project_path}") + print(f"Generating DRC report for project: {project_path}") if not os.path.exists(project_path): return f"Project not found: {project_path}" @@ -162,7 +156,7 @@ def register_drc_resources(mcp: FastMCP) -> None: return "PCB file not found in project" pcb_file = files["pcb"] - logger.info(f"Found PCB file: {pcb_file}") + print(f"Found PCB file: {pcb_file}") # Try to run DRC via command line drc_results = run_drc_via_cli(pcb_file) diff --git a/kicad_mcp/server.py b/kicad_mcp/server.py index b730b01..d343c05 100644 --- a/kicad_mcp/server.py +++ b/kicad_mcp/server.py @@ -28,33 +28,29 @@ from kicad_mcp.prompts.bom_prompts import register_bom_prompts from kicad_mcp.prompts.pattern_prompts import register_pattern_prompts # Import utils -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.python_path import setup_kicad_python_path # Import context management from kicad_mcp.context import kicad_lifespan -# Create logger for this module -logger = Logger() - def create_server() -> FastMCP: """Create and configure the KiCad MCP server.""" - logger.info("Initializing KiCad MCP server") + print("Initializing KiCad MCP server") # Try to set up KiCad Python path kicad_modules_available = setup_kicad_python_path() if kicad_modules_available: - logger.info("KiCad Python modules successfully configured") + print("KiCad Python modules successfully configured") else: - logger.warning("KiCad Python modules not available - some features will be disabled") + print("KiCad Python modules not available - some features will be disabled") # Initialize FastMCP server mcp = FastMCP("KiCad", lifespan=kicad_lifespan) - logger.info("Created FastMCP server instance with lifespan management") + print("Created FastMCP server instance with lifespan management") # Register resources - logger.debug("Registering resources...") + print("Registering resources...") register_project_resources(mcp) register_file_resources(mcp) register_drc_resources(mcp) @@ -63,7 +59,7 @@ def create_server() -> FastMCP: register_pattern_resources(mcp) # Register tools - logger.debug("Registering tools...") + print("Registering tools...") register_project_tools(mcp) register_analysis_tools(mcp) register_export_tools(mcp) @@ -73,11 +69,11 @@ def create_server() -> FastMCP: register_pattern_tools(mcp) # Register prompts - logger.debug("Registering prompts...") + print("Registering prompts...") register_prompts(mcp) register_drc_prompts(mcp) register_bom_prompts(mcp) register_pattern_prompts(mcp) - logger.info("Server initialization complete") + print("Server initialization complete") return mcp diff --git a/kicad_mcp/tools/bom_tools.py b/kicad_mcp/tools/bom_tools.py index 203b599..5053c3c 100644 --- a/kicad_mcp/tools/bom_tools.py +++ b/kicad_mcp/tools/bom_tools.py @@ -31,10 +31,10 @@ def register_bom_tools(mcp: FastMCP) -> None: Returns: Dictionary with BOM analysis results """ - logger.info(f"Analyzing BOM for project: {project_path}") + print(f"Analyzing BOM for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} @@ -50,10 +50,10 @@ def register_bom_tools(mcp: FastMCP) -> None: for file_type, file_path in files.items(): if "bom" in file_type.lower() or file_path.lower().endswith(".csv"): bom_files[file_type] = file_path - logger.info(f"Found potential BOM file: {file_path}") + print(f"Found potential BOM file: {file_path}") if not bom_files: - logger.warning("No BOM files found for project") + print("No BOM files found for project") ctx.info("No BOM files found for project") return { "success": False, @@ -82,7 +82,7 @@ def register_bom_tools(mcp: FastMCP) -> None: bom_data, format_info = parse_bom_file(file_path) if not bom_data or len(bom_data) == 0: - logger.warning(f"Failed to parse BOM file: {file_path}") + print(f"Failed to parse BOM file: {file_path}") continue # Analyze the BOM data @@ -99,10 +99,10 @@ def register_bom_tools(mcp: FastMCP) -> None: total_unique_components += analysis["unique_component_count"] total_components += analysis["total_component_count"] - logger.info(f"Successfully analyzed BOM file: {file_path}") + print(f"Successfully analyzed BOM file: {file_path}") except Exception as e: - logger.error(f"Error analyzing BOM file {file_path}: {str(e)}", exc_info=True) + print(f"Error analyzing BOM file {file_path}: {str(e)}", exc_info=True) results["bom_files"][file_type] = { "path": file_path, "error": str(e) @@ -165,10 +165,10 @@ def register_bom_tools(mcp: FastMCP) -> None: Returns: Dictionary with export results """ - logger.info(f"Exporting BOM for project: {project_path}") + print(f"Exporting BOM for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} @@ -184,7 +184,7 @@ def register_bom_tools(mcp: FastMCP) -> None: # We need the schematic file to generate a BOM if "schematic" not in files: - logger.error("Schematic file not found in project") + print("Schematic file not found in project") ctx.info("Schematic file not found in project") return {"success": False, "error": "Schematic file not found"} @@ -205,7 +205,7 @@ def register_bom_tools(mcp: FastMCP) -> None: ctx.info("Attempting to export BOM using KiCad Python modules...") export_result = await export_bom_with_python(schematic_file, project_dir, project_name, ctx) except Exception as e: - logger.error(f"Error exporting BOM with Python modules: {str(e)}", exc_info=True) + print(f"Error exporting BOM with Python modules: {str(e)}", exc_info=True) ctx.info(f"Error using Python modules: {str(e)}") export_result = {"success": False, "error": str(e)} @@ -215,7 +215,7 @@ def register_bom_tools(mcp: FastMCP) -> None: ctx.info("Attempting to export BOM using command-line tools...") export_result = await export_bom_with_cli(schematic_file, project_dir, project_name, ctx) except Exception as e: - logger.error(f"Error exporting BOM with CLI: {str(e)}", exc_info=True) + print(f"Error exporting BOM with CLI: {str(e)}", exc_info=True) ctx.info(f"Error using command-line tools: {str(e)}") export_result = {"success": False, "error": str(e)} @@ -242,7 +242,7 @@ def parse_bom_file(file_path: str) -> Tuple[List[Dict[str, Any]], Dict[str, Any] - List of component dictionaries - Dictionary with format information """ - logger.info(f"Parsing BOM file: {file_path}") + print(f"Parsing BOM file: {file_path}") # Check file extension _, ext = os.path.splitext(file_path) @@ -342,18 +342,18 @@ def parse_bom_file(file_path: str) -> Tuple[List[Dict[str, Any]], Dict[str, Any] for row in reader: components.append(dict(row)) except: - logger.error(f"Failed to parse unknown file format: {file_path}") + print(f"Failed to parse unknown file format: {file_path}") return [], {"detected_format": "unsupported"} except Exception as e: - logger.error(f"Error parsing BOM file: {str(e)}", exc_info=True) + print(f"Error parsing BOM file: {str(e)}", exc_info=True) return [], {"error": str(e)} # Check if we actually got components if not components: - logger.warning(f"No components found in BOM file: {file_path}") + print(f"No components found in BOM file: {file_path}") else: - logger.info(f"Successfully parsed {len(components)} components from {file_path}") + print(f"Successfully parsed {len(components)} components from {file_path}") # Add a sample of the fields found if components: @@ -372,7 +372,7 @@ def analyze_bom_data(components: List[Dict[str, Any]], format_info: Dict[str, An Returns: Dictionary with analysis results """ - logger.info(f"Analyzing {len(components)} components") + print(f"Analyzing {len(components)} components") # Initialize results results = { @@ -547,7 +547,7 @@ def analyze_bom_data(components: List[Dict[str, Any]], format_info: Dict[str, An if "currency" not in results: results["currency"] = "USD" # Default except: - logger.warning("Failed to parse cost data") + print("Failed to parse cost data") # Add extra insights if ref_col and value_col: @@ -557,7 +557,7 @@ def analyze_bom_data(components: List[Dict[str, Any]], format_info: Dict[str, An results["most_common_values"] = {str(k): int(v) for k, v in most_common.items()} except Exception as e: - logger.error(f"Error analyzing BOM data: {str(e)}", exc_info=True) + print(f"Error analyzing BOM data: {str(e)}", exc_info=True) # Fallback to basic analysis results["unique_component_count"] = len(components) results["total_component_count"] = len(components) @@ -577,7 +577,7 @@ async def export_bom_with_python(schematic_file: str, output_dir: str, project_n Returns: Dictionary with export results """ - logger.info(f"Exporting BOM for schematic: {schematic_file}") + print(f"Exporting BOM for schematic: {schematic_file}") await ctx.report_progress(30, 100) try: @@ -587,7 +587,7 @@ async def export_bom_with_python(schematic_file: str, output_dir: str, project_n import pcbnew # For now, return a message indicating this method is not implemented yet - logger.warning("BOM export with Python modules not fully implemented") + print("BOM export with Python modules not fully implemented") ctx.info("BOM export with Python modules not fully implemented yet") return { @@ -597,7 +597,7 @@ async def export_bom_with_python(schematic_file: str, output_dir: str, project_n } except ImportError: - logger.error("Failed to import KiCad Python modules") + print("Failed to import KiCad Python modules") return { "success": False, "error": "Failed to import KiCad Python modules", @@ -621,7 +621,7 @@ async def export_bom_with_cli(schematic_file: str, output_dir: str, project_name import platform system = platform.system() - logger.info(f"Exporting BOM using CLI tools on {system}") + print(f"Exporting BOM using CLI tools on {system}") await ctx.report_progress(40, 100) # Output file path @@ -696,7 +696,7 @@ async def export_bom_with_cli(schematic_file: str, output_dir: str, project_name } try: - logger.info(f"Running command: {' '.join(cmd)}") + print(f"Running command: {' '.join(cmd)}") await ctx.report_progress(60, 100) # Run the command @@ -704,8 +704,8 @@ async def export_bom_with_cli(schematic_file: str, output_dir: str, project_name # Check if the command was successful if process.returncode != 0: - logger.error(f"BOM export command failed with code {process.returncode}") - logger.error(f"Error output: {process.stderr}") + print(f"BOM export command failed with code {process.returncode}") + print(f"Error output: {process.stderr}") return { "success": False, @@ -746,7 +746,7 @@ async def export_bom_with_cli(schematic_file: str, output_dir: str, project_name } except subprocess.TimeoutExpired: - logger.error("BOM export command timed out after 30 seconds") + print("BOM export command timed out after 30 seconds") return { "success": False, "error": "BOM export command timed out after 30 seconds", @@ -754,7 +754,7 @@ async def export_bom_with_cli(schematic_file: str, output_dir: str, project_name } except Exception as e: - logger.error(f"Error exporting BOM: {str(e)}", exc_info=True) + print(f"Error exporting BOM: {str(e)}", exc_info=True) return { "success": False, "error": f"Error exporting BOM: {str(e)}", diff --git a/kicad_mcp/tools/drc_impl/cli_drc.py b/kicad_mcp/tools/drc_impl/cli_drc.py index 6325ff4..2d3518e 100644 --- a/kicad_mcp/tools/drc_impl/cli_drc.py +++ b/kicad_mcp/tools/drc_impl/cli_drc.py @@ -8,12 +8,8 @@ import tempfile from typing import Dict, Any, Optional from mcp.server.fastmcp import Context -from kicad_mcp.utils.logger import Logger from kicad_mcp.config import system -# Create logger for this module -logger = Logger() - async def run_drc_via_cli(pcb_file: str, ctx: Context) -> Dict[str, Any]: """Run DRC using KiCad command line tools. @@ -39,7 +35,7 @@ async def run_drc_via_cli(pcb_file: str, ctx: Context) -> Dict[str, Any]: # Find kicad-cli executable kicad_cli = find_kicad_cli() if not kicad_cli: - logger.error("kicad-cli not found in PATH or common installation locations") + print("kicad-cli not found in PATH or common installation locations") results["error"] = "kicad-cli not found. Please ensure KiCad 9.0+ is installed and kicad-cli is available." return results @@ -57,19 +53,19 @@ async def run_drc_via_cli(pcb_file: str, ctx: Context) -> Dict[str, Any]: pcb_file ] - logger.info(f"Running command: {' '.join(cmd)}") + print(f"Running command: {' '.join(cmd)}") process = subprocess.run(cmd, capture_output=True, text=True) # Check if the command was successful if process.returncode != 0: - logger.error(f"DRC command failed with code {process.returncode}") - logger.error(f"Error output: {process.stderr}") + print(f"DRC command failed with code {process.returncode}") + print(f"Error output: {process.stderr}") results["error"] = f"DRC command failed: {process.stderr}" return results # Check if the output file was created if not os.path.exists(output_file): - logger.error("DRC report file not created") + print("DRC report file not created") results["error"] = "DRC report file not created" return results @@ -78,14 +74,14 @@ async def run_drc_via_cli(pcb_file: str, ctx: Context) -> Dict[str, Any]: try: drc_report = json.load(f) except json.JSONDecodeError: - logger.error("Failed to parse DRC report JSON") + print("Failed to parse DRC report JSON") results["error"] = "Failed to parse DRC report JSON" return results # Process the DRC report violations = drc_report.get("violations", []) violation_count = len(violations) - logger.info(f"DRC completed with {violation_count} violations") + print(f"DRC completed with {violation_count} violations") await ctx.report_progress(70, 100) ctx.info(f"DRC completed with {violation_count} violations") @@ -111,7 +107,7 @@ async def run_drc_via_cli(pcb_file: str, ctx: Context) -> Dict[str, Any]: return results except Exception as e: - logger.error(f"Error in CLI DRC: {str(e)}", exc_info=True) + print(f"Error in CLI DRC: {str(e)}", exc_info=True) results["error"] = f"Error in CLI DRC: {str(e)}" return results @@ -136,7 +132,7 @@ def find_kicad_cli() -> Optional[str]: return result.stdout.strip() except Exception as e: - logger.error(f"Error finding kicad-cli: {str(e)}") + print(f"Error finding kicad-cli: {str(e)}") # If we get here, kicad-cli is not in PATH # Try common installation locations diff --git a/kicad_mcp/tools/drc_impl/ipc_drc.py b/kicad_mcp/tools/drc_impl/ipc_drc.py index 3ebe1b8..08334be 100644 --- a/kicad_mcp/tools/drc_impl/ipc_drc.py +++ b/kicad_mcp/tools/drc_impl/ipc_drc.py @@ -2,15 +2,12 @@ Design Rule Check (DRC) implementation using the KiCad IPC API. """ import os -from typing import Dict, Any +from typing import Any, Dict + from mcp.server.fastmcp import Context -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.kicad_api_detection import check_ipc_api_environment -# Create logger for this module -logger = Logger() - async def run_drc_with_ipc_api(pcb_file: str, ctx: Context) -> Dict[str, Any]: """Run DRC using the KiCad IPC API (kicad-python). This requires a running instance of KiCad with the IPC API enabled. @@ -26,7 +23,7 @@ async def run_drc_with_ipc_api(pcb_file: str, ctx: Context) -> Dict[str, Any]: # Import the kicad-python modules import kipy from kipy.board_types import DrcExclusion, DrcSeverity - logger.info("Successfully imported kipy modules") + print("Successfully imported kipy modules") # Check if we're running in a KiCad IPC plugin environment is_plugin, socket_path = check_ipc_api_environment() @@ -72,7 +69,7 @@ async def run_drc_with_ipc_api(pcb_file: str, ctx: Context) -> Dict[str, Any]: doc = await kicad.open_document(pcb_file) board_doc = doc except Exception as e: - logger.error(f"Error opening board: {str(e)}") + print(f"Error opening board: {str(e)}") return { "success": False, "method": "ipc", @@ -104,7 +101,7 @@ async def run_drc_with_ipc_api(pcb_file: str, ctx: Context) -> Dict[str, Any]: violations = drc_report.violations violation_count = len(violations) - logger.info(f"DRC completed with {violation_count} violations") + print(f"DRC completed with {violation_count} violations") ctx.info(f"DRC completed with {violation_count} violations") # Process the violations @@ -148,14 +145,14 @@ async def run_drc_with_ipc_api(pcb_file: str, ctx: Context) -> Dict[str, Any]: return results except ImportError as e: - logger.error(f"Failed to import kipy modules: {str(e)}") + print(f"Failed to import kipy modules: {str(e)}") return { "success": False, "method": "ipc", "error": f"Failed to import kipy modules: {str(e)}" } except Exception as e: - logger.error(f"Error in IPC API DRC: {str(e)}", exc_info=True) + print(f"Error in IPC API DRC: {str(e)}", exc_info=True) return { "success": False, "method": "ipc", diff --git a/kicad_mcp/tools/drc_tools.py b/kicad_mcp/tools/drc_tools.py index c7a4b4a..f3ad0f0 100644 --- a/kicad_mcp/tools/drc_tools.py +++ b/kicad_mcp/tools/drc_tools.py @@ -6,7 +6,6 @@ from typing import Dict, Any from mcp.server.fastmcp import FastMCP, Context from kicad_mcp.utils.file_utils import get_project_files -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.drc_history import save_drc_result, get_drc_history, compare_with_previous from kicad_mcp.utils.kicad_api_detection import get_best_api_approach @@ -14,9 +13,6 @@ from kicad_mcp.utils.kicad_api_detection import get_best_api_approach from kicad_mcp.tools.drc_impl.cli_drc import run_drc_via_cli from kicad_mcp.tools.drc_impl.ipc_drc import run_drc_with_ipc_api -# Create logger for this module -logger = Logger() - def register_drc_tools(mcp: FastMCP) -> None: """Register DRC tools with the MCP server. @@ -34,10 +30,10 @@ def register_drc_tools(mcp: FastMCP) -> None: Returns: Dictionary with DRC history entries """ - logger.info(f"Getting DRC history for project: {project_path}") + print(f"Getting DRC history for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} # Get history entries @@ -78,20 +74,20 @@ def register_drc_tools(mcp: FastMCP) -> None: Returns: Dictionary with DRC results and statistics """ - logger.info(f"Running DRC check for project: {project_path}") + print(f"Running DRC check for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} # Get PCB file from project files = get_project_files(project_path) if "pcb" not in files: - logger.error("PCB file not found in project") + print("PCB file not found in project") return {"success": False, "error": "PCB file not found in project"} pcb_file = files["pcb"] - logger.info(f"Found PCB file: {pcb_file}") + print(f"Found PCB file: {pcb_file}") # Report progress to user await ctx.report_progress(10, 100) @@ -106,19 +102,19 @@ def register_drc_tools(mcp: FastMCP) -> None: if api_approach == "cli": # Use CLI approach (kicad-cli) - logger.info("Using kicad-cli for DRC") + print("Using kicad-cli for DRC") ctx.info("Using KiCad CLI for DRC check...") drc_results = await run_drc_via_cli(pcb_file, ctx) elif api_approach == "ipc": # Use IPC API approach (kicad-python) - logger.info("Using IPC API for DRC") + print("Using IPC API for DRC") ctx.info("Using KiCad IPC API for DRC check...") drc_results = await run_drc_with_ipc_api(pcb_file, ctx) else: # No API available - logger.error("No KiCad API available for DRC") + print("No KiCad API available for DRC") return { "success": False, "error": "No KiCad API available for DRC. Please install KiCad 9.0 or later." diff --git a/kicad_mcp/tools/export_tools.py b/kicad_mcp/tools/export_tools.py index 1426ba2..cd7c340 100644 --- a/kicad_mcp/tools/export_tools.py +++ b/kicad_mcp/tools/export_tools.py @@ -10,12 +10,8 @@ from typing import Dict, Any, Optional from mcp.server.fastmcp import FastMCP, Context, Image from kicad_mcp.utils.file_utils import get_project_files -from kicad_mcp.utils.logger import Logger from kicad_mcp.config import KICAD_APP_PATH, system -# Create logger for this module -logger = Logger() - def register_export_tools(mcp: FastMCP) -> None: """Register export tools with the MCP server. @@ -26,10 +22,10 @@ def register_export_tools(mcp: FastMCP) -> None: @mcp.tool() def validate_project(project_path: str) -> Dict[str, Any]: """Basic validation of a KiCad project.""" - logger.info(f"Validating project: {project_path}") + print(f"Validating project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") return {"valid": False, "error": f"Project not found: {project_path}"} issues = [] @@ -37,11 +33,11 @@ def register_export_tools(mcp: FastMCP) -> None: # Check for essential files if "pcb" not in files: - logger.warning("Missing PCB layout file") + print("Missing PCB layout file") issues.append("Missing PCB layout file") if "schematic" not in files: - logger.warning("Missing schematic file") + print("Missing schematic file") issues.append("Missing schematic file") # Validate project file @@ -49,12 +45,12 @@ def register_export_tools(mcp: FastMCP) -> None: with open(project_path, 'r') as f: import json json.load(f) - logger.debug("Project file validated successfully") + print("Project file validated successfully") except json.JSONDecodeError: - logger.error("Invalid project file format (JSON parsing error)") + print("Invalid project file format (JSON parsing error)") issues.append("Invalid project file format (JSON parsing error)") except Exception as e: - logger.error(f"Error reading project file: {str(e)}") + print(f"Error reading project file: {str(e)}") issues.append(f"Error reading project file: {str(e)}") result = { @@ -64,7 +60,7 @@ def register_export_tools(mcp: FastMCP) -> None: "files_found": list(files.keys()) } - logger.info(f"Validation result: {'valid' if result['valid'] else 'invalid'}") + print(f"Validation result: {'valid' if result['valid'] else 'invalid'}") return result @mcp.tool() @@ -83,27 +79,27 @@ def register_export_tools(mcp: FastMCP) -> None: app_context = ctx.request_context.lifespan_context kicad_modules_available = app_context.kicad_modules_available - logger.info(f"Generating thumbnail for project: {project_path}") + print(f"Generating thumbnail for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return None # Get PCB file from project files = get_project_files(project_path) if "pcb" not in files: - logger.error("PCB file not found in project") + print("PCB file not found in project") ctx.info("PCB file not found in project") return None pcb_file = files["pcb"] - logger.info(f"Found PCB file: {pcb_file}") + print(f"Found PCB file: {pcb_file}") # Check cache cache_key = f"thumbnail_{pcb_file}_{os.path.getmtime(pcb_file)}" if hasattr(app_context, 'cache') and cache_key in app_context.cache: - logger.info(f"Using cached thumbnail for {pcb_file}") + print(f"Using cached thumbnail for {pcb_file}") return app_context.cache[cache_key] await ctx.report_progress(10, 100) @@ -120,12 +116,12 @@ def register_export_tools(mcp: FastMCP) -> None: return thumbnail # If pcbnew method failed, log it but continue to try alternative method - logger.warning("Failed to generate thumbnail with pcbnew, trying CLI method") + print("Failed to generate thumbnail with pcbnew, trying CLI method") except Exception as e: - logger.error(f"Error using pcbnew for thumbnail: {str(e)}", exc_info=True) + print(f"Error using pcbnew for thumbnail: {str(e)}", exc_info=True) ctx.info(f"Error with pcbnew method, trying alternative approach") else: - logger.info("KiCad Python modules not available, trying CLI method") + print("KiCad Python modules not available, trying CLI method") # Method 2: Try to use command-line tools try: @@ -136,7 +132,7 @@ def register_export_tools(mcp: FastMCP) -> None: app_context.cache[cache_key] = thumbnail return thumbnail except Exception as e: - logger.error(f"Error using CLI for thumbnail: {str(e)}", exc_info=True) + print(f"Error using CLI for thumbnail: {str(e)}", exc_info=True) ctx.info(f"Error generating thumbnail with CLI method") # If all methods fail, inform the user @@ -144,10 +140,10 @@ def register_export_tools(mcp: FastMCP) -> None: return None except asyncio.CancelledError: - logger.info("Thumbnail generation cancelled") + print("Thumbnail generation cancelled") raise # Re-raise to let MCP know the task was cancelled except Exception as e: - logger.error(f"Unexpected error in thumbnail generation: {str(e)}") + print(f"Unexpected error in thumbnail generation: {str(e)}") ctx.info(f"Error: {str(e)}") return None @@ -159,44 +155,44 @@ def register_export_tools(mcp: FastMCP) -> None: app_context = ctx.request_context.lifespan_context kicad_modules_available = app_context.kicad_modules_available - logger.info(f"Generating thumbnail for project: {project_path}") + print(f"Generating thumbnail for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return None # Get PCB file files = get_project_files(project_path) if "pcb" not in files: - logger.error("PCB file not found in project") + print("PCB file not found in project") ctx.info("PCB file not found in project") return None pcb_file = files["pcb"] - logger.info(f"Found PCB file: {pcb_file}") + print(f"Found PCB file: {pcb_file}") if not kicad_modules_available: - logger.warning("KiCad Python modules are not available - cannot generate thumbnail") + print("KiCad Python modules are not available - cannot generate thumbnail") ctx.info("KiCad Python modules are not available") return None # Check cache cache_key = f"project_thumbnail_{pcb_file}_{os.path.getmtime(pcb_file)}" if hasattr(app_context, 'cache') and cache_key in app_context.cache: - logger.info(f"Using cached project thumbnail for {pcb_file}") + print(f"Using cached project thumbnail for {pcb_file}") return app_context.cache[cache_key] try: # Try to import pcbnew import pcbnew - logger.info("Successfully imported pcbnew module") + print("Successfully imported pcbnew module") # Load the PCB file - logger.debug(f"Loading PCB file: {pcb_file}") + print(f"Loading PCB file: {pcb_file}") board = pcbnew.LoadBoard(pcb_file) if not board: - logger.error("Failed to load PCB file") + print("Failed to load PCB file") ctx.info("Failed to load PCB file") return None @@ -205,12 +201,12 @@ def register_export_tools(mcp: FastMCP) -> None: width = board_box.GetWidth() / 1000000.0 # Convert to mm height = board_box.GetHeight() / 1000000.0 - logger.info(f"PCB dimensions: {width:.2f}mm x {height:.2f}mm") + print(f"PCB dimensions: {width:.2f}mm x {height:.2f}mm") ctx.info(f"PCB dimensions: {width:.2f}mm x {height:.2f}mm") # Create temporary directory for output with tempfile.TemporaryDirectory() as temp_dir: - logger.debug(f"Created temporary directory: {temp_dir}") + print(f"Created temporary directory: {temp_dir}") # Create PLOT_CONTROLLER for plotting pctl = pcbnew.PLOT_CONTROLLER(board) @@ -241,7 +237,7 @@ def register_export_tools(mcp: FastMCP) -> None: plot_basename = "thumbnail" output_filename = os.path.join(temp_dir, f"{plot_basename}.png") - logger.debug(f"Plotting PCB to: {output_filename}") + print(f"Plotting PCB to: {output_filename}") # Plot PNG pctl.OpenPlotfile(plot_basename, pcbnew.PLOT_FORMAT_PNG, "Thumbnail") @@ -252,7 +248,7 @@ def register_export_tools(mcp: FastMCP) -> None: plot_file = os.path.join(temp_dir, f"{plot_basename}.png") if not os.path.exists(plot_file): - logger.error(f"Expected plot file not found: {plot_file}") + print(f"Expected plot file not found: {plot_file}") ctx.info("Failed to generate PCB image") return None @@ -260,7 +256,7 @@ def register_export_tools(mcp: FastMCP) -> None: with open(plot_file, 'rb') as f: img_data = f.read() - logger.info(f"Successfully generated thumbnail, size: {len(img_data)} bytes") + print(f"Successfully generated thumbnail, size: {len(img_data)} bytes") # Create and cache the image thumbnail = Image(data=img_data, format="png") @@ -270,19 +266,19 @@ def register_export_tools(mcp: FastMCP) -> None: return thumbnail except ImportError as e: - logger.error(f"Failed to import pcbnew module: {str(e)}") + print(f"Failed to import pcbnew module: {str(e)}") ctx.info(f"Failed to import pcbnew module: {str(e)}") return None except Exception as e: - logger.error(f"Error generating thumbnail: {str(e)}", exc_info=True) + print(f"Error generating thumbnail: {str(e)}", exc_info=True) ctx.info(f"Error generating thumbnail: {str(e)}") return None except asyncio.CancelledError: - logger.info("Project thumbnail generation cancelled") + print("Project thumbnail generation cancelled") raise except Exception as e: - logger.error(f"Unexpected error in project thumbnail generation: {str(e)}", exc_info=True) + print(f"Unexpected error in project thumbnail generation: {str(e)}", exc_info=True) ctx.info(f"Error: {str(e)}") return None @@ -299,14 +295,14 @@ async def generate_thumbnail_with_pcbnew(pcb_file: str, ctx: Context) -> Optiona """ try: import pcbnew - logger.info("Successfully imported pcbnew module") + print("Successfully imported pcbnew module") await ctx.report_progress(20, 100) # Load the PCB file - logger.debug(f"Loading PCB file with pcbnew: {pcb_file}") + print(f"Loading PCB file with pcbnew: {pcb_file}") board = pcbnew.LoadBoard(pcb_file) if not board: - logger.error("Failed to load PCB file with pcbnew") + print("Failed to load PCB file with pcbnew") return None # Report progress @@ -318,11 +314,11 @@ async def generate_thumbnail_with_pcbnew(pcb_file: str, ctx: Context) -> Optiona width_mm = board_box.GetWidth() / 1000000.0 # Convert to mm height_mm = board_box.GetHeight() / 1000000.0 - logger.info(f"PCB dimensions: {width_mm:.2f}mm x {height_mm:.2f}mm") + print(f"PCB dimensions: {width_mm:.2f}mm x {height_mm:.2f}mm") # Create temporary directory for output with tempfile.TemporaryDirectory() as temp_dir: - logger.debug(f"Created temporary directory: {temp_dir}") + print(f"Created temporary directory: {temp_dir}") # Create PLOT_CONTROLLER for plotting pctl = pcbnew.PLOT_CONTROLLER(board) @@ -355,7 +351,7 @@ async def generate_thumbnail_with_pcbnew(pcb_file: str, ctx: Context) -> Optiona # Determine output filename plot_basename = "thumbnail" - logger.debug(f"Plotting PCB to PNG") + print(f"Plotting PCB to PNG") await ctx.report_progress(50, 100) # Plot PNG @@ -369,22 +365,22 @@ async def generate_thumbnail_with_pcbnew(pcb_file: str, ctx: Context) -> Optiona plot_file = os.path.join(temp_dir, f"{plot_basename}.png") if not os.path.exists(plot_file): - logger.error(f"Expected plot file not found: {plot_file}") + print(f"Expected plot file not found: {plot_file}") return None # Read the image file with open(plot_file, 'rb') as f: img_data = f.read() - logger.info(f"Successfully generated thumbnail, size: {len(img_data)} bytes") + print(f"Successfully generated thumbnail, size: {len(img_data)} bytes") await ctx.report_progress(90, 100) return Image(data=img_data, format="png") except ImportError as e: - logger.error(f"Failed to import pcbnew module: {str(e)}") + print(f"Failed to import pcbnew module: {str(e)}") return None except Exception as e: - logger.error(f"Error generating thumbnail with pcbnew: {str(e)}", exc_info=True) + print(f"Error generating thumbnail with pcbnew: {str(e)}", exc_info=True) return None async def generate_thumbnail_with_cli(pcb_file: str, ctx: Context) -> Optional[Image]: @@ -399,7 +395,7 @@ async def generate_thumbnail_with_cli(pcb_file: str, ctx: Context) -> Optional[I Image object containing the PCB thumbnail or None if generation failed """ try: - logger.info("Attempting to generate thumbnail using command line tools") + print("Attempting to generate thumbnail using command line tools") await ctx.report_progress(20, 100) # Check for required command-line tools based on OS @@ -408,22 +404,22 @@ async def generate_thumbnail_with_cli(pcb_file: str, ctx: Context) -> Optional[I if not os.path.exists(pcbnew_cli) and shutil.which("pcbnew_cli") is not None: pcbnew_cli = "pcbnew_cli" # Try to use from PATH elif not os.path.exists(pcbnew_cli): - logger.error(f"pcbnew_cli not found at {pcbnew_cli} or in PATH") + print(f"pcbnew_cli not found at {pcbnew_cli} or in PATH") return None elif system == "Windows": pcbnew_cli = os.path.join(KICAD_APP_PATH, "bin", "pcbnew_cli.exe") if not os.path.exists(pcbnew_cli) and shutil.which("pcbnew_cli") is not None: pcbnew_cli = "pcbnew_cli" # Try to use from PATH elif not os.path.exists(pcbnew_cli): - logger.error(f"pcbnew_cli not found at {pcbnew_cli} or in PATH") + print(f"pcbnew_cli not found at {pcbnew_cli} or in PATH") return None elif system == "Linux": pcbnew_cli = shutil.which("pcbnew_cli") if not pcbnew_cli: - logger.error("pcbnew_cli not found in PATH") + print("pcbnew_cli not found in PATH") return None else: - logger.error(f"Unsupported operating system: {system}") + print(f"Unsupported operating system: {system}") return None await ctx.report_progress(30, 100) @@ -444,7 +440,7 @@ async def generate_thumbnail_with_cli(pcb_file: str, ctx: Context) -> Optional[I pcb_file ] - logger.debug(f"Running command: {' '.join(cmd)}") + print(f"Running command: {' '.join(cmd)}") await ctx.report_progress(50, 100) # Run the command @@ -452,35 +448,35 @@ async def generate_thumbnail_with_cli(pcb_file: str, ctx: Context) -> Optional[I process = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if process.returncode != 0: - logger.error(f"Command failed with code {process.returncode}") - logger.error(f"Error: {process.stderr}") + print(f"Command failed with code {process.returncode}") + print(f"Error: {process.stderr}") return None await ctx.report_progress(70, 100) # Check if the output file was created if not os.path.exists(output_file): - logger.error(f"Output file not created: {output_file}") + print(f"Output file not created: {output_file}") return None # Read the image file with open(output_file, 'rb') as f: img_data = f.read() - logger.info(f"Successfully generated thumbnail with CLI, size: {len(img_data)} bytes") + print(f"Successfully generated thumbnail with CLI, size: {len(img_data)} bytes") await ctx.report_progress(90, 100) return Image(data=img_data, format="png") except subprocess.TimeoutExpired: - logger.error("Command timed out after 30 seconds") + print("Command timed out after 30 seconds") return None except Exception as e: - logger.error(f"Error running CLI command: {str(e)}", exc_info=True) + print(f"Error running CLI command: {str(e)}", exc_info=True) return None except asyncio.CancelledError: - logger.info("CLI thumbnail generation cancelled") + print("CLI thumbnail generation cancelled") raise except Exception as e: - logger.error(f"Unexpected error in CLI thumbnail generation: {str(e)}") + print(f"Unexpected error in CLI thumbnail generation: {str(e)}") return None diff --git a/kicad_mcp/tools/netlist_tools.py b/kicad_mcp/tools/netlist_tools.py index e86bd0f..42e521c 100644 --- a/kicad_mcp/tools/netlist_tools.py +++ b/kicad_mcp/tools/netlist_tools.py @@ -6,12 +6,8 @@ from typing import Dict, Any from mcp.server.fastmcp import FastMCP, Context from kicad_mcp.utils.file_utils import get_project_files -from kicad_mcp.utils.logger import Logger from kicad_mcp.utils.netlist_parser import extract_netlist, analyze_netlist -# Create logger for this module -logger = Logger() - def register_netlist_tools(mcp: FastMCP) -> None: """Register netlist-related tools with the MCP server. @@ -33,10 +29,10 @@ def register_netlist_tools(mcp: FastMCP) -> None: Returns: Dictionary with netlist information """ - logger.info(f"Extracting netlist from schematic: {schematic_path}") + print(f"Extracting netlist from schematic: {schematic_path}") if not os.path.exists(schematic_path): - logger.error(f"Schematic file not found: {schematic_path}") + print(f"Schematic file not found: {schematic_path}") ctx.info(f"Schematic file not found: {schematic_path}") return {"success": False, "error": f"Schematic file not found: {schematic_path}"} @@ -52,7 +48,7 @@ def register_netlist_tools(mcp: FastMCP) -> None: netlist_data = extract_netlist(schematic_path) if "error" in netlist_data: - logger.error(f"Error extracting netlist: {netlist_data['error']}") + print(f"Error extracting netlist: {netlist_data['error']}") ctx.info(f"Error extracting netlist: {netlist_data['error']}") return {"success": False, "error": netlist_data['error']} @@ -85,7 +81,7 @@ def register_netlist_tools(mcp: FastMCP) -> None: return result except Exception as e: - logger.error(f"Error extracting netlist: {str(e)}") + print(f"Error extracting netlist: {str(e)}") ctx.info(f"Error extracting netlist: {str(e)}") return {"success": False, "error": str(e)} @@ -103,10 +99,10 @@ def register_netlist_tools(mcp: FastMCP) -> None: Returns: Dictionary with netlist information """ - logger.info(f"Extracting netlist for project: {project_path}") + print(f"Extracting netlist for project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} @@ -118,12 +114,12 @@ def register_netlist_tools(mcp: FastMCP) -> None: files = get_project_files(project_path) if "schematic" not in files: - logger.error("Schematic file not found in project") + print("Schematic file not found in project") ctx.info("Schematic file not found in project") return {"success": False, "error": "Schematic file not found in project"} schematic_path = files["schematic"] - logger.info(f"Found schematic file: {schematic_path}") + print(f"Found schematic file: {schematic_path}") ctx.info(f"Found schematic file: {os.path.basename(schematic_path)}") # Extract netlist @@ -139,7 +135,7 @@ def register_netlist_tools(mcp: FastMCP) -> None: return result except Exception as e: - logger.error(f"Error extracting project netlist: {str(e)}") + print(f"Error extracting project netlist: {str(e)}") ctx.info(f"Error extracting project netlist: {str(e)}") return {"success": False, "error": str(e)} @@ -157,10 +153,10 @@ def register_netlist_tools(mcp: FastMCP) -> None: Returns: Dictionary with connection analysis """ - logger.info(f"Analyzing connections in schematic: {schematic_path}") + print(f"Analyzing connections in schematic: {schematic_path}") if not os.path.exists(schematic_path): - logger.error(f"Schematic file not found: {schematic_path}") + print(f"Schematic file not found: {schematic_path}") ctx.info(f"Schematic file not found: {schematic_path}") return {"success": False, "error": f"Schematic file not found: {schematic_path}"} @@ -173,7 +169,7 @@ def register_netlist_tools(mcp: FastMCP) -> None: netlist_data = extract_netlist(schematic_path) if "error" in netlist_data: - logger.error(f"Error extracting netlist: {netlist_data['error']}") + print(f"Error extracting netlist: {netlist_data['error']}") ctx.info(f"Error extracting netlist: {netlist_data['error']}") return {"success": False, "error": netlist_data['error']} @@ -250,7 +246,7 @@ def register_netlist_tools(mcp: FastMCP) -> None: return result except Exception as e: - logger.error(f"Error analyzing connections: {str(e)}") + print(f"Error analyzing connections: {str(e)}") ctx.info(f"Error analyzing connections: {str(e)}") return {"success": False, "error": str(e)} @@ -269,10 +265,10 @@ def register_netlist_tools(mcp: FastMCP) -> None: Returns: Dictionary with component connection information """ - logger.info(f"Finding connections for component {component_ref} in project: {project_path}") + print(f"Finding connections for component {component_ref} in project: {project_path}") if not os.path.exists(project_path): - logger.error(f"Project not found: {project_path}") + print(f"Project not found: {project_path}") ctx.info(f"Project not found: {project_path}") return {"success": False, "error": f"Project not found: {project_path}"} @@ -284,12 +280,12 @@ def register_netlist_tools(mcp: FastMCP) -> None: files = get_project_files(project_path) if "schematic" not in files: - logger.error("Schematic file not found in project") + print("Schematic file not found in project") ctx.info("Schematic file not found in project") return {"success": False, "error": "Schematic file not found in project"} schematic_path = files["schematic"] - logger.info(f"Found schematic file: {schematic_path}") + print(f"Found schematic file: {schematic_path}") ctx.info(f"Found schematic file: {os.path.basename(schematic_path)}") # Extract netlist @@ -299,14 +295,14 @@ def register_netlist_tools(mcp: FastMCP) -> None: netlist_data = extract_netlist(schematic_path) if "error" in netlist_data: - logger.error(f"Failed to extract netlist: {netlist_data['error']}") + print(f"Failed to extract netlist: {netlist_data['error']}") ctx.info(f"Failed to extract netlist: {netlist_data['error']}") return {"success": False, "error": netlist_data['error']} # Check if component exists in the netlist components = netlist_data.get("components", {}) if component_ref not in components: - logger.error(f"Component {component_ref} not found in schematic") + print(f"Component {component_ref} not found in schematic") ctx.info(f"Component {component_ref} not found in schematic") return { "success": False, @@ -405,6 +401,6 @@ def register_netlist_tools(mcp: FastMCP) -> None: return result except Exception as e: - logger.error(f"Error finding component connections: {str(e)}", exc_info=True) + print(f"Error finding component connections: {str(e)}", exc_info=True) ctx.info(f"Error finding component connections: {str(e)}") return {"success": False, "error": str(e)} diff --git a/kicad_mcp/utils/drc_history.py b/kicad_mcp/utils/drc_history.py index 0337203..c31296b 100644 --- a/kicad_mcp/utils/drc_history.py +++ b/kicad_mcp/utils/drc_history.py @@ -9,12 +9,6 @@ import platform import time from datetime import datetime from typing import Dict, List, Any, Optional -from pathlib import Path - -from kicad_mcp.utils.logger import Logger - -# Create logger for this module -logger = Logger() # Directory for storing DRC history if platform.system() == "Windows": @@ -73,7 +67,7 @@ def save_drc_result(project_path: str, drc_result: Dict[str, Any]) -> None: with open(history_path, 'r') as f: history = json.load(f) except (json.JSONDecodeError, IOError) as e: - logger.error(f"Error loading DRC history: {str(e)}") + print(f"Error loading DRC history: {str(e)}") history = {"project_path": project_path, "entries": []} else: history = {"project_path": project_path, "entries": []} @@ -92,9 +86,9 @@ def save_drc_result(project_path: str, drc_result: Dict[str, Any]) -> None: try: with open(history_path, 'w') as f: json.dump(history, f, indent=2) - logger.info(f"Saved DRC history entry to {history_path}") + print(f"Saved DRC history entry to {history_path}") except IOError as e: - logger.error(f"Error saving DRC history: {str(e)}") + print(f"Error saving DRC history: {str(e)}") def get_drc_history(project_path: str) -> List[Dict[str, Any]]: @@ -109,7 +103,7 @@ def get_drc_history(project_path: str) -> List[Dict[str, Any]]: history_path = get_project_history_path(project_path) if not os.path.exists(history_path): - logger.info(f"No DRC history found for {project_path}") + print(f"No DRC history found for {project_path}") return [] try: @@ -125,7 +119,7 @@ def get_drc_history(project_path: str) -> List[Dict[str, Any]]: return entries except (json.JSONDecodeError, IOError) as e: - logger.error(f"Error reading DRC history: {str(e)}") + print(f"Error reading DRC history: {str(e)}") return [] diff --git a/kicad_mcp/utils/kicad_api_detection.py b/kicad_mcp/utils/kicad_api_detection.py index 2780b4e..ad2f89e 100644 --- a/kicad_mcp/utils/kicad_api_detection.py +++ b/kicad_mcp/utils/kicad_api_detection.py @@ -6,12 +6,8 @@ import subprocess import shutil from typing import Tuple, Optional, Literal -from kicad_mcp.utils.logger import Logger from kicad_mcp.config import system -# Create logger for this module -logger = Logger() - def check_for_cli_api() -> bool: """Check if KiCad CLI API is available. @@ -36,7 +32,7 @@ def check_for_cli_api() -> bool: result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: - logger.info(f"Found working kicad-cli: {kicad_cli}") + print(f"Found working kicad-cli: {kicad_cli}") return True # Check common installation locations if not found in PATH @@ -63,14 +59,14 @@ def check_for_cli_api() -> bool: # Check each potential path for path in potential_paths: if os.path.exists(path) and os.access(path, os.X_OK): - logger.info(f"Found kicad-cli at common location: {path}") + print(f"Found kicad-cli at common location: {path}") return True - logger.info("KiCad CLI API is not available") + print("KiCad CLI API is not available") return False except Exception as e: - logger.error(f"Error checking for KiCad CLI API: {str(e)}") + print(f"Error checking for KiCad CLI API: {str(e)}") return False @@ -83,13 +79,13 @@ def check_for_ipc_api() -> bool: try: # Try to import the kipy module import kipy - logger.info("KiCad IPC API (kicad-python) is available") + print("KiCad IPC API (kicad-python) is available") return True except ImportError: - logger.info("KiCad IPC API (kicad-python) is not available") + print("KiCad IPC API (kicad-python) is not available") return False except Exception as e: - logger.error(f"Error checking for KiCad IPC API: {str(e)}") + print(f"Error checking for KiCad IPC API: {str(e)}") return False @@ -106,9 +102,9 @@ def check_ipc_api_environment() -> Tuple[bool, Optional[str]]: socket_path = os.environ.get("KICAD_SOCKET_PATH") if is_plugin: - logger.info("Running as a KiCad plugin") + print("Running as a KiCad plugin") elif socket_path: - logger.info(f"KiCad IPC socket path found: {socket_path}") + print(f"KiCad IPC socket path found: {socket_path}") return (is_plugin, socket_path) @@ -131,5 +127,5 @@ def get_best_api_approach() -> Literal["cli", "ipc", "none"]: return "cli" # No API available - logger.warning("No KiCad API available") + print("No KiCad API available") return "none" diff --git a/kicad_mcp/utils/logger.py b/kicad_mcp/utils/logger.py deleted file mode 100644 index ed89e15..0000000 --- a/kicad_mcp/utils/logger.py +++ /dev/null @@ -1,106 +0,0 @@ -""" -Simple logger with automatic function-level context tracking for KiCad MCP Server. - -Usage examples: -# Creates logs in the "logs" directory by default -logger = Logger() - -# To disable file logging completely -logger = Logger(log_dir=None) - -# Or to specify a custom logs directory -logger = Logger(log_dir="custom_logs") -""" -import os -import sys -import logging -import inspect -from datetime import datetime -from pathlib import Path - - -class Logger: - """ - Simple logger that automatically tracks function-level context. - """ - def __init__(self, name=None, log_dir=None, console_level=logging.INFO, file_level=logging.DEBUG): - """ - Initialize a logger with automatic function-level context. - - Args: - name: Logger name (defaults to calling module name) - log_dir: Directory to store log files (default: "logs" directory) - Set to None to disable file logging - console_level: Logging level for console output - file_level: Logging level for file output - """ - # If no name provided, try to determine it from the calling module - if name is None: - frame = inspect.currentframe().f_back - module = inspect.getmodule(frame) - self.name = module.__name__ if module else "kicad_mcp" - else: - self.name = name - - # Initialize Python's logger - self.logger = logging.getLogger(self.name) - self.logger.setLevel(logging.DEBUG) # Capture all levels, filtering at handler level - - # Only configure if not already configured - if not self.logger.handlers and not logging.getLogger().handlers: - # Create formatter with detailed context - formatter = logging.Formatter( - '%(asctime)s [%(levelname)s] %(pathname)s:%(funcName)s:%(lineno)d - %(message)s', - datefmt='%Y-%m-%d %H:%M:%S' - ) - - # Set up console output - console_handler = logging.StreamHandler(sys.stdout) - console_handler.setLevel(console_level) - console_handler.setFormatter(formatter) - self.logger.addHandler(console_handler) - - # Set up file output by default unless explicitly disabled - if log_dir is not None: - log_dir_path = Path(log_dir) - log_dir_path.mkdir(parents=True, exist_ok=True) - - timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") - log_file = log_dir_path / f"kicad_mcp_{timestamp}.log" - - file_handler = logging.FileHandler(log_file) - file_handler.setLevel(file_level) - file_handler.setFormatter(formatter) - self.logger.addHandler(file_handler) - - self.info(f"Logging session started, log file: {log_file}") - - def _get_caller_info(self): - """Get information about the function that called the logger.""" - # Skip this function, the log method, and get to the actual caller - frame = inspect.currentframe().f_back.f_back - return frame - - def debug(self, message): - """Log a debug message with caller context.""" - self.logger.debug(message) - - def info(self, message): - """Log an info message with caller context.""" - self.logger.info(message) - - def warning(self, message): - """Log a warning message with caller context.""" - self.logger.warning(message) - - def error(self, message): - """Log an error message with caller context.""" - self.logger.error(message) - - def critical(self, message): - """Log a critical message with caller context.""" - self.logger.critical(message) - - def exception(self, message): - """Log an exception message with caller context and traceback.""" - self.logger.exception(message) diff --git a/kicad_mcp/utils/netlist_parser.py b/kicad_mcp/utils/netlist_parser.py index 615ac12..894eb3b 100644 --- a/kicad_mcp/utils/netlist_parser.py +++ b/kicad_mcp/utils/netlist_parser.py @@ -3,14 +3,9 @@ KiCad schematic netlist extraction utilities. """ import os import re -from typing import Dict, List, Set, Tuple, Any, Optional +from typing import Any, Dict, List from collections import defaultdict -from kicad_mcp.utils.logger import Logger - -# Create logger for this module -logger = Logger() - class SchematicParser: """Parser for KiCad schematic files to extract netlist information.""" @@ -44,15 +39,15 @@ class SchematicParser: def _load_schematic(self) -> None: """Load the schematic file content.""" if not os.path.exists(self.schematic_path): - logger.error(f"Schematic file not found: {self.schematic_path}") + print(f"Schematic file not found: {self.schematic_path}") raise FileNotFoundError(f"Schematic file not found: {self.schematic_path}") try: with open(self.schematic_path, 'r') as f: self.content = f.read() - logger.info(f"Successfully loaded schematic: {self.schematic_path}") + print(f"Successfully loaded schematic: {self.schematic_path}") except Exception as e: - logger.error(f"Error reading schematic file: {str(e)}") + print(f"Error reading schematic file: {str(e)}") raise def parse(self) -> Dict[str, Any]: @@ -61,7 +56,7 @@ class SchematicParser: Returns: Dictionary with parsed netlist information """ - logger.info("Starting schematic parsing") + print("Starting schematic parsing") # Extract symbols (components) self._extract_components() @@ -96,7 +91,7 @@ class SchematicParser: "net_count": len(self.nets) } - logger.info(f"Schematic parsing complete: found {len(self.component_info)} components and {len(self.nets)} nets") + print(f"Schematic parsing complete: found {len(self.component_info)} components and {len(self.nets)} nets") return result def _extract_s_expressions(self, pattern: str) -> List[str]: @@ -143,7 +138,7 @@ class SchematicParser: def _extract_components(self) -> None: """Extract component information from schematic.""" - logger.info("Extracting components") + print("Extracting components") # Extract all symbol expressions (components) symbols = self._extract_s_expressions(r'\(symbol\s+') @@ -157,7 +152,7 @@ class SchematicParser: ref = component.get('reference', 'Unknown') self.component_info[ref] = component - logger.info(f"Extracted {len(self.components)} components") + print(f"Extracted {len(self.components)} components") def _parse_component(self, symbol_expr: str) -> Dict[str, Any]: """Parse a component from a symbol S-expression. @@ -220,7 +215,7 @@ class SchematicParser: def _extract_wires(self) -> None: """Extract wire information from schematic.""" - logger.info("Extracting wires") + print("Extracting wires") # Extract all wire expressions wires = self._extract_s_expressions(r'\(wire\s+') @@ -240,11 +235,11 @@ class SchematicParser: } }) - logger.info(f"Extracted {len(self.wires)} wires") + print(f"Extracted {len(self.wires)} wires") def _extract_junctions(self) -> None: """Extract junction information from schematic.""" - logger.info("Extracting junctions") + print("Extracting junctions") # Extract all junction expressions junctions = self._extract_s_expressions(r'\(junction\s+') @@ -258,11 +253,11 @@ class SchematicParser: 'y': float(xy_match.group(2)) }) - logger.info(f"Extracted {len(self.junctions)} junctions") + print(f"Extracted {len(self.junctions)} junctions") def _extract_labels(self) -> None: """Extract label information from schematic.""" - logger.info("Extracting labels") + print("Extracting labels") # Extract local labels local_labels = self._extract_s_expressions(r'\(label\s+') @@ -317,11 +312,11 @@ class SchematicParser: } }) - logger.info(f"Extracted {len(self.labels)} local labels, {len(self.global_labels)} global labels, and {len(self.hierarchical_labels)} hierarchical labels") + print(f"Extracted {len(self.labels)} local labels, {len(self.global_labels)} global labels, and {len(self.hierarchical_labels)} hierarchical labels") def _extract_power_symbols(self) -> None: """Extract power symbol information from schematic.""" - logger.info("Extracting power symbols") + print("Extracting power symbols") # Extract all power symbol expressions power_symbols = self._extract_s_expressions(r'\(symbol\s+\(lib_id\s+"power:') @@ -341,11 +336,11 @@ class SchematicParser: } }) - logger.info(f"Extracted {len(self.power_symbols)} power symbols") + print(f"Extracted {len(self.power_symbols)} power symbols") def _extract_no_connects(self) -> None: """Extract no-connect information from schematic.""" - logger.info("Extracting no-connects") + print("Extracting no-connects") # Extract all no-connect expressions no_connects = self._extract_s_expressions(r'\(no_connect\s+') @@ -359,11 +354,11 @@ class SchematicParser: 'y': float(xy_match.group(2)) }) - logger.info(f"Extracted {len(self.no_connects)} no-connects") + print(f"Extracted {len(self.no_connects)} no-connects") def _build_netlist(self) -> None: """Build the netlist from extracted components and connections.""" - logger.info("Building netlist from schematic data") + print("Building netlist from schematic data") # TODO: Implement netlist building algorithm # This is a complex task that involves: @@ -391,8 +386,8 @@ class SchematicParser: # and detect connected pins # For demonstration, we'll add a placeholder note - logger.info("Note: Full netlist building requires complex connectivity tracing") - logger.info(f"Found {len(self.nets)} potential nets from labels and power symbols") + print("Note: Full netlist building requires complex connectivity tracing") + print(f"Found {len(self.nets)} potential nets from labels and power symbols") def extract_netlist(schematic_path: str) -> Dict[str, Any]: @@ -408,7 +403,7 @@ def extract_netlist(schematic_path: str) -> Dict[str, Any]: parser = SchematicParser(schematic_path) return parser.parse() except Exception as e: - logger.error(f"Error extracting netlist: {str(e)}") + print(f"Error extracting netlist: {str(e)}") return { "error": str(e), "components": {}, diff --git a/kicad_mcp/utils/python_path.py b/kicad_mcp/utils/python_path.py index 5ae8c5e..93440c2 100644 --- a/kicad_mcp/utils/python_path.py +++ b/kicad_mcp/utils/python_path.py @@ -6,10 +6,6 @@ import sys import glob import platform -from kicad_mcp.utils.logger import Logger - -logger = Logger() - def setup_kicad_python_path(): """ Add KiCad Python modules to the Python path by detecting the appropriate version. @@ -18,14 +14,14 @@ def setup_kicad_python_path(): bool: True if successful, False otherwise """ system = platform.system() - logger.info(f"Setting up KiCad Python path for {system}") + print(f"Setting up KiCad Python path for {system}") # Define search paths based on operating system if system == "Darwin": # macOS from kicad_mcp.config import KICAD_APP_PATH if not os.path.exists(KICAD_APP_PATH): - logger.error(f"KiCad application not found at {KICAD_APP_PATH}") + print(f"KiCad application not found at {KICAD_APP_PATH}") return False # Base path to Python framework @@ -37,7 +33,7 @@ def setup_kicad_python_path(): # If 'Current' symlink doesn't work, find all available Python versions if not site_packages: - logger.debug("'Current' symlink not found, searching for numbered versions") + print("'Current' symlink not found, searching for numbered versions") # Look for numbered versions like 3.9, 3.10, etc. version_dirs = glob.glob(os.path.join(python_base, "[0-9]*")) for version_dir in version_dirs: @@ -74,7 +70,7 @@ def setup_kicad_python_path(): site_packages = expanded_packages else: - logger.error(f"Unsupported operating system: {system}") + print(f"Unsupported operating system: {system}") return False # Pick the first valid path found @@ -89,20 +85,20 @@ def setup_kicad_python_path(): if os.path.exists(pcbnew_path): if path not in sys.path: sys.path.append(path) - logger.info(f"Added KiCad Python path: {path}") - logger.info(f"Found pcbnew module at: {pcbnew_path}") + print(f"Added KiCad Python path: {path}") + print(f"Found pcbnew module at: {pcbnew_path}") # Try to actually import it to verify compatibility try: import pcbnew - logger.info(f"Successfully imported pcbnew module version: {getattr(pcbnew, 'GetBuildVersion', lambda: 'unknown')()}") + print(f"Successfully imported pcbnew module version: {getattr(pcbnew, 'GetBuildVersion', lambda: 'unknown')()}") return True except ImportError as e: - logger.error(f"Found pcbnew but failed to import: {str(e)}") + print(f"Found pcbnew but failed to import: {str(e)}") # Remove from path as it's not usable sys.path.remove(path) else: - logger.debug(f"Found site-packages at {path} but no pcbnew module") + print(f"Found site-packages at {path} but no pcbnew module") - logger.error("Could not find a valid KiCad Python site-packages directory with pcbnew module") + print("Could not find a valid KiCad Python site-packages directory with pcbnew module") return False diff --git a/main.py b/main.py index 736de8b..55efb21 100644 --- a/main.py +++ b/main.py @@ -9,28 +9,25 @@ import sys from kicad_mcp.config import KICAD_USER_DIR, ADDITIONAL_SEARCH_PATHS from kicad_mcp.server import create_server from kicad_mcp.utils.env import load_dotenv -from kicad_mcp.utils.logger import Logger # Load environment variables from .env file if present load_dotenv() -logger = Logger() - if __name__ == "__main__": try: - logger.info("Starting KiCad MCP server") + print("Starting KiCad MCP server") # Log search paths from config - logger.info(f"Using KiCad user directory: {KICAD_USER_DIR}") + print(f"Using KiCad user directory: {KICAD_USER_DIR}") if ADDITIONAL_SEARCH_PATHS: - logger.info(f"Additional search paths: {', '.join(ADDITIONAL_SEARCH_PATHS)}") + print(f"Additional search paths: {', '.join(ADDITIONAL_SEARCH_PATHS)}") else: - logger.info("No additional search paths configured") + print("No additional search paths configured") # Create and run server server = create_server() - logger.info("Running server with stdio transport") + print("Running server with stdio transport") server.run(transport='stdio') except Exception as e: - logger.exception(f"Unhandled exception: {str(e)}") + print(f"Unhandled exception: {str(e)}") raise