remove all instances of logger.py (see issue #1)
This commit is contained in:
Lama
parent
3596bae34e
commit
fda329cc8f
@ -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
|
||||
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)}",
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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."
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)}
|
||||
|
||||
@ -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 []
|
||||
|
||||
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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)
|
||||
@ -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": {},
|
||||
|
||||
@ -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
|
||||
|
||||
15
main.py
15
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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user