diff --git a/.gitignore b/.gitignore index 16cb7c581..b1d5f56e9 100644 --- a/.gitignore +++ b/.gitignore @@ -186,3 +186,5 @@ uno_r3 *.bak *.pkl +# Mac stuff +.DS_Store diff --git a/setup.py b/setup.py index 60ffa9dae..0dc675900 100644 --- a/setup.py +++ b/setup.py @@ -31,9 +31,12 @@ "sexpdata == 1.0.0", "kinparse >= 1.2.3", "kinet2pcb >= 1.1.2", - #'PySpice; python_version >= "3.0"', "graphviz", "deprecation", + "requests >= 2.31.0", + "importlib-metadata", # For importlib support + "typing-extensions", # For type hints in Python <3.8 + "openai", ] test_requirements = [ @@ -59,7 +62,8 @@ packages=setuptools.find_packages(where="src"), entry_points={ "console_scripts": [ - "netlist_to_skidl = skidl.scripts.netlist_to_skidl_main:main" + "netlist_to_skidl = skidl.scripts.netlist_to_skidl_main:main", + "kicad_skidl_llm = skidl.scripts.kicad_skidl_llm_main:main" ] }, package_dir={"": "src"}, diff --git a/src/skidl/__init__.py b/src/skidl/__init__.py index 544a8e7e8..3f6fb0de3 100644 --- a/src/skidl/__init__.py +++ b/src/skidl/__init__.py @@ -42,6 +42,7 @@ ) from .pin import Pin from .schlib import SchLib, load_backup_lib +from .circuit_analyzer import SkidlCircuitAnalyzer from .skidl import ( ERC, POWER, @@ -57,6 +58,8 @@ generate_svg, generate_xml, lib_search_paths, + get_circuit_info, + analyze_with_llm, no_files, reset, get_default_tool, diff --git a/src/skidl/circuit.py b/src/skidl/circuit.py index 2bb29f279..eafdfe2b8 100644 --- a/src/skidl/circuit.py +++ b/src/skidl/circuit.py @@ -8,7 +8,7 @@ import json import subprocess -from collections import Counter, deque +from collections import Counter, deque, defaultdict import graphviz @@ -106,13 +106,15 @@ def mini_reset(self, init=False): self.interfaces = [] self.packages = deque() self.hierarchy = "top" - self.level = 0 - self.context = [("top",)] + # self.level = 0 + # self.context = [("top",)] + self.context = [] self.erc_assertion_list = [] self.circuit_stack = ( [] ) # Stack of previous default_circuits for context manager. self.no_files = False # Allow creation of files for netlists, ERC, libs, etc. + self.subcircuit_docs = {} # Store documentation for subcircuits # Internal set used to check for duplicate hierarchical names. self._hierarchical_names = {self.hierarchy} @@ -191,9 +193,20 @@ def activate(self, name, tag): self.hierarchy = self.hierarchy + HIER_SEP + name + str(tag) self.add_hierarchical_name(self.hierarchy) + # Store subcircuit docstring if available + import inspect + frame = inspect.currentframe() + try: + # Go up 2 frames to get to the subcircuit function + subcircuit_func = frame.f_back.f_back.f_locals.get('f') + if subcircuit_func and subcircuit_func.__doc__: + self.subcircuit_docs[self.hierarchy] = subcircuit_func.__doc__.strip() + finally: + del frame # Avoid reference cycles + # Setup some globals needed in this context. builtins.default_circuit = self - builtins.NC = self.NC # pylint: disable=undefined-variable + builtins.NC = self.NC def deactivate(self): """Deactivate the current hierarchical group and return to the previous one.""" @@ -1198,3 +1211,206 @@ def no_files(self, stop): """Don't output any files if stop is True.""" self._no_files = stop stop_log_file_output(stop) + + def get_circuit_info(self, hierarchy=None, depth=None, filename="circuit_description.txt"): + """ + Save circuit information to a text file and return the description as a string. + Shows hierarchical structure of the circuit with consolidated parts and connections. + + Args: + hierarchy (str): Starting hierarchy level to analyze. If None, starts from top. + depth (int): How many levels deep to analyze. If None, analyzes all levels. + filename (str): Output filename for the circuit description. + """ + + # A list for storing lines of text describing the circuit. + circuit_info = [] + circuit_info.append("=" * 40) + circuit_info.append(f"Circuit Name: {self.name}") + + # Get hierarchy label for the starting point. + start_hier = hierarchy or self.hierarchy + start_depth = len(start_hier.split(HIER_SEP)) + circuit_info.append(f"Starting Hierarchy: {start_hier}") + + # Group parts by hierarchy and collect all hierarchical labels + # at or below the starting point. + hierarchy_parts = defaultdict(list) + hierarchies = set() + for part in self.parts: + if part.hierarchy.startswith(start_hier): + # Check depth constraint if specified + if depth is None or len(part.hierarchy.split(HIER_SEP)) - start_depth <= depth: + hierarchy_parts[part.hierarchy].append(part) + hierarchies.add(part.hierarchy) + + # Get nets and group by hierarchy. + net_hierarchies = defaultdict(list) + for net in self.get_nets(): + net_hier_connections = defaultdict(list) + for pin in net.pins: + if pin.part.hierarchy in hierarchies: + net_hier_connections[pin.part.hierarchy].append(pin) + + for hier in net_hier_connections: + net_hierarchies[hier].append((net, net_hier_connections)) + + # Print consolidated information for each hierarchy level. + first_hierarchy = True + for hier in sorted(hierarchies): + if not first_hierarchy: + circuit_info.append("_" * 53) + else: + first_hierarchy = False + + circuit_info.append(f"Hierarchy Level: {hier}") + + # Add subcircuit docstring if available + if hier in self.subcircuit_docs: + circuit_info.append("\nSubcircuit Documentation:") + circuit_info.append(self.subcircuit_docs[hier]) + circuit_info.append("") + + # Parts in this hierarchy + if hier in hierarchy_parts: + circuit_info.append("Parts:") + for part in sorted(hierarchy_parts[hier], key=lambda p: p.ref): + circuit_info.append(f" Part: {part.ref}") + circuit_info.append(f" Name: {part.name}") + circuit_info.append(f" Value: {part.value}") + circuit_info.append(f" Footprint: {part.footprint}") + # Add part docstring if available + if hasattr(part, 'description'): + circuit_info.append(f" Description: {part.description}") + # Add part purpose if available + if hasattr(part, 'purpose'): + circuit_info.append(f" Purpose: {part.purpose}") + circuit_info.append(" Pins:") + for pin in part.pins: + net_name = pin.net.name if pin.net else "unconnected" + circuit_info.append(f" {pin.num}/{pin.name}: {net_name}") + + # Nets in this hierarchy + if hier in net_hierarchies: + circuit_info.append("\nNets:") + for net, connections in sorted(net_hierarchies[hier], key=lambda x: x[0].name): + circuit_info.append(f" Net: {net.name}") + circuit_info.append(" Connections:") + for pin in connections[hier]: + circuit_info.append(f" {pin.part.ref}.{pin.name}") + + return "\n".join(circuit_info) + + def analyze_with_llm( + self, + api_key=None, + output_file="circuit_llm_analysis.txt", + hierarchy=None, + depth=None, + analyze_subcircuits=False, + save_query_only=False, + custom_prompt=None, + backend="openrouter", + model=None, + ): + """ + Analyze the circuit using LLM, with options for analyzing the whole circuit or individual subcircuits. + + Args: + api_key: API key for the LLM service (required for OpenRouter, not needed for Ollama) + output_file: File to save analysis results. If analyzing subcircuits, this will contain consolidated results. + hierarchy: Starting hierarchy level to analyze. If None, starts from top. + depth: How many levels deep to analyze. If None, analyzes all levels. + analyze_subcircuits: If True, analyzes each subcircuit separately with depth=1. + If False, analyzes from the specified hierarchy and depth. + save_query_only: If True, only saves the query that would be sent to the LLM without executing it. + custom_prompt: Optional custom prompt to append to the default analysis prompt. + This allows adding specific analysis requirements or questions. + backend: LLM backend to use ("openrouter" or "ollama"). Defaults to "openrouter". + model: Model to use for analysis. Defaults to backend's default model. + + Returns: + If analyze_subcircuits=False: + Dictionary containing single analysis results + If analyze_subcircuits=True: + Dictionary containing: + - success: Overall success status + - subcircuits: Dict of analysis results for each subcircuit + - total_time_seconds: Total analysis time + - total_tokens: Total tokens used + """ + from .circuit_analyzer import SkidlCircuitAnalyzer + + if save_query_only: + backend = None # Don't need backend for query only. + + analyzer = SkidlCircuitAnalyzer( + api_key=api_key, + custom_prompt=custom_prompt, + backend=backend, + model=model + ) + + if not analyze_subcircuits: + # Single analysis of specified hierarchy + circuit_desc = self.get_circuit_info(hierarchy=hierarchy, depth=depth) + return analyzer.analyze_circuit(circuit_desc, output_file=output_file, save_query_only=save_query_only) + + # Analyze each subcircuit separately + results = { + "success": True, + "subcircuits": {}, + "total_time_seconds": 0, + "total_tokens": 0 + } + + # Get all unique subcircuit hierarchies + hierarchies = set() + for part in self.parts: + if part.hierarchy != self.hierarchy: # Skip top level + hierarchies.add(part.hierarchy) + + # Analyze each subcircuit + for hier in sorted(hierarchies): + # Get description focused on this subcircuit + circuit_desc = self.get_circuit_info(hierarchy=hier, depth=1) + + # Analyze just this subcircuit + sub_results = analyzer.analyze_circuit( + circuit_desc, + output_file=None, # Don't write individual files + save_query_only=save_query_only + ) + + results["subcircuits"][hier] = sub_results + results["total_time_seconds"] += sub_results.get("request_time_seconds", 0) + results["total_tokens"] += sub_results.get("prompt_tokens", 0) + sub_results.get("response_tokens", 0) + + # Save consolidated results if requested + if output_file: + consolidated_text = ["=== Subcircuits Analysis ===\n"] + + for hier, analysis in results["subcircuits"].items(): + consolidated_text.append(f"\n{'='*20} {hier} {'='*20}\n") + if analysis.get("success", False): + # Include the actual analysis text + analysis_text = analysis.get("analysis", "No analysis available") + consolidated_text.append(analysis_text) + + # Include token usage info + token_info = ( + f"\nTokens used: {analysis.get('total_tokens', 0)} " + f"(Prompt: {analysis.get('prompt_tokens', 0)}, " + f"Completion: {analysis.get('completion_tokens', 0)})" + ) + consolidated_text.append(token_info) + else: + consolidated_text.append( + f"Analysis failed: {analysis.get('error', 'Unknown error')}" + ) + consolidated_text.append("\n") + + with open(output_file, "w") as f: + f.write("\n".join(consolidated_text)) + + return results diff --git a/src/skidl/circuit_analyzer.py b/src/skidl/circuit_analyzer.py new file mode 100644 index 000000000..0607b2098 --- /dev/null +++ b/src/skidl/circuit_analyzer.py @@ -0,0 +1,366 @@ +"""Module for circuit analysis using LLMs through OpenRouter or local Ollama instance.""" + +from typing import Dict, Optional, Literal +from datetime import datetime +import time +import os +import hashlib +import requests +from openai import OpenAI +from .logger import active_logger # Import the active_logger + +# API configuration +OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions" +OLLAMA_API_URL = "http://localhost:11434/api/chat" +DEFAULT_MODEL = "google/gemini-2.0-flash-001" +DEFAULT_OLLAMA_MODEL = "llama3.2:latest" +DEFAULT_TIMEOUT = 30 +DEFAULT_TEMPERATURE = 0.7 +DEFAULT_MAX_TOKENS = 20000 +MAX_RETRIES = 3 +FILE_OPERATION_RETRIES = 3 # Retries for file operations +FILE_RETRY_DELAY = 1 # Delay between retries in seconds + +# Approximate cost per 1K tokens for typical OpenRouter usage +# (These are user-defined or approximate values and might not match real billing exactly.) +DEFAULT_COST_PER_1K_TOKENS = 0.002 + +class SkidlCircuitAnalyzer: + def __init__( + self, + model: Optional[str] = None, + api_key: Optional[str] = None, + custom_prompt: Optional[str] = None, + analysis_flags: Optional[Dict[str, bool]] = None, + timeout: int = DEFAULT_TIMEOUT, + temperature: float = DEFAULT_TEMPERATURE, + max_tokens: int = DEFAULT_MAX_TOKENS, + backend: Literal["openrouter", "ollama"] = "openrouter", + cost_per_1k_tokens: float = DEFAULT_COST_PER_1K_TOKENS, + **kwargs + ): + """ + Initialize the circuit analyzer with configuration parameters. + + Args: + model: Model identifier for the LLM + api_key: API key for OpenRouter (required if using OpenRouter backend) + custom_prompt: Additional custom prompts to include in analysis + analysis_flags: Dict of analysis sections to enable/disable + timeout: Request timeout in seconds + temperature: Model temperature parameter + max_tokens: Maximum tokens for completion + backend: Either "openrouter" or "ollama" + cost_per_1k_tokens: Approximate cost to be multiplied per 1K tokens (for OpenRouter) + """ + self.backend = backend + + # Check for API key if using OpenRouter + if backend == "openrouter": + self.api_key = api_key or os.getenv("OPENROUTER_API_KEY") + if not self.api_key: + raise ValueError( + "OpenRouter API key required. Either:\n" + "1. Set OPENROUTER_API_KEY environment variable\n" + "2. Pass api_key parameter to analyze_with_llm" + ) + # Fixed model selection logic + self.model = model if model else DEFAULT_MODEL + else: + self.api_key = None + # For Ollama, use provided model or default Ollama model + self.model = model if model else DEFAULT_OLLAMA_MODEL + + self.timeout = timeout + self.temperature = temperature + self.max_tokens = max_tokens + self.custom_prompt = custom_prompt + + from .prompts import ANALYSIS_SECTIONS + self.analysis_flags = analysis_flags or { + section: True for section in ANALYSIS_SECTIONS.keys() + } + invalid_sections = set(self.analysis_flags.keys()) - set(ANALYSIS_SECTIONS.keys()) + if invalid_sections: + raise ValueError(f"Invalid analysis sections: {invalid_sections}") + + # Keep track of total cost (approximate) + self.cost_per_1k_tokens = cost_per_1k_tokens + self.total_approx_cost = 0.0 + + self.config = kwargs + + def _generate_unique_identifier(self, subcircuit_name: str, module_path: str) -> str: + """ + Generate a unique identifier for a subcircuit to avoid name collisions. + + Args: + subcircuit_name: Name of the subcircuit function + module_path: Path to the module containing the subcircuit + + Returns: + A unique identifier string + """ + # Create a unique string combining module path and subcircuit name + unique_string = f"{module_path}:{subcircuit_name}" + # Generate a hash and take first 8 characters for brevity + hash_id = hashlib.md5(unique_string.encode()).hexdigest()[:8] + return f"{subcircuit_name}_{hash_id}" + + def _save_analysis_with_retry(self, output_file: str, analysis_text: str, verbose: bool = True) -> None: + """ + Save analysis results to a file with retry mechanism for handling file locks. + + Args: + output_file: Path to save the analysis + analysis_text: Analysis content to save + verbose: Whether to print progress messages + """ + if verbose: + active_logger.info(f"\nSaving analysis to {output_file}...") + + for attempt in range(FILE_OPERATION_RETRIES): + try: + with open(output_file, "w") as f: + f.write(analysis_text) + if verbose: + active_logger.info("Analysis saved successfully") + return + except PermissionError as e: + if attempt < FILE_OPERATION_RETRIES - 1: + if verbose: + active_logger.warning(f"Retry {attempt + 1}: File locked, waiting...") + time.sleep(FILE_RETRY_DELAY) + else: + raise IOError(f"Failed to save analysis after {FILE_OPERATION_RETRIES} attempts: {str(e)}") + + def _generate_analysis_prompt(self, circuit_description: str) -> str: + """ + Generate the complete analysis prompt. + + Args: + circuit_description: Description of the circuit to analyze + + Returns: + Complete prompt string for the LLM + """ + from .prompts import get_base_prompt, ANALYSIS_SECTIONS + + # Build enabled analysis sections + enabled_sections = [] + for section, content in ANALYSIS_SECTIONS.items(): + if self.analysis_flags.get(section, True): + enabled_sections.append(content) + + analysis_sections = "\n".join(enabled_sections) + + # Generate complete prompt using base template + prompt = get_base_prompt( + circuit_description=circuit_description, + analysis_sections=analysis_sections + ) + + # Append custom prompt if provided + if self.custom_prompt: + prompt += f"\n\nAdditional Analysis Requirements:\n{self.custom_prompt}" + + return prompt + + def analyze_circuit( + self, + circuit_description: str, + output_file: Optional[str] = "circuit_llm_analysis.txt", + verbose: bool = True, + save_query_only: bool = False + ) -> Dict: + """ + Analyze the circuit using the configured LLM. + + Args: + circuit_description: Description of the circuit to analyze + output_file: File to save analysis results (None to skip saving) + verbose: Whether to print progress messages + save_query_only: If True, only save the query without executing + + Returns: + Dictionary containing analysis results and metadata + """ + start_time = time.time() + + # Show appropriate default model based on backend + display_model = self.model if self.model else (DEFAULT_OLLAMA_MODEL if self.backend == "ollama" else DEFAULT_MODEL) + if verbose: + active_logger.info(f"\n=== {'Saving Query' if save_query_only else 'Starting Circuit Analysis'} with {display_model} ===") + + try: + # Generate the analysis prompt + prompt = self._generate_analysis_prompt(circuit_description) + + # If save_query_only is True, just save the prompt and return + if save_query_only: + if output_file: + self._save_analysis_with_retry(output_file, prompt, verbose) + if verbose: + active_logger.info("\n=== Query saved successfully ===") + return { + "success": True, + "query": prompt, + "timestamp": int(datetime.now().timestamp()), + "total_time_seconds": time.time() - start_time + } + + if verbose: + active_logger.info("\nGenerating analysis...") + + # Get analysis from selected backend with retries + request_start = time.time() + + if self.backend == "openrouter": + analysis_results = self._handle_openrouter_request(prompt, request_start, verbose) + else: + analysis_results = self._handle_ollama_request(prompt, request_start) + + # Add common result fields + results = { + "success": True, + "timestamp": int(datetime.now().timestamp()), + "total_time_seconds": time.time() - start_time, + "enabled_analyses": [ + k for k, v in self.analysis_flags.items() if v + ], + **analysis_results + } + + # Save analysis to file if required + if output_file and results.get("analysis"): + self._save_analysis_with_retry(output_file, results["analysis"], verbose) + + if verbose: + active_logger.info(f"\n=== Analysis completed in {results['total_time_seconds']:.2f} seconds ===") + + return results + + except Exception as e: + error_message = f"Analysis failed: {str(e)}" + active_logger.error(f"\nERROR: {error_message}") + + error_results = { + "success": False, + "error": error_message, + "timestamp": int(datetime.now().timestamp()), + "total_time_seconds": time.time() - start_time + } + + if output_file: + try: + self._save_analysis_with_retry(output_file, error_message, verbose) + except Exception as save_error: + active_logger.error(f"Failed to save error message: {str(save_error)}") + + return error_results + + def _handle_openrouter_request(self, prompt: str, request_start: float, verbose: bool) -> Dict: + """ + Handle requests to OpenRouter API with retries, and track approximate cost. + + Args: + prompt: The analysis prompt to send + request_start: Start time of the request + verbose: Whether to print progress messages + + Returns: + Dictionary containing API response data + """ + client = OpenAI( + base_url="https://openrouter.ai/api/v1", + api_key=self.api_key, + ) + + extra_headers = { + "HTTP-Referer": "https://github.com/devbisme/skidl", + "X-Title": "SKiDL Circuit Analyzer" + } + + for attempt in range(MAX_RETRIES): + try: + completion = client.chat.completions.create( + model=self.model, + messages=[{"role": "user", "content": prompt}], + temperature=self.temperature, + max_tokens=self.max_tokens, + extra_headers=extra_headers + ) + + # Extract text and usage + analysis_text = completion.choices[0].message.content + prompt_tokens = completion.usage.prompt_tokens + completion_tokens = completion.usage.completion_tokens + total_tokens = completion.usage.total_tokens + request_time = time.time() - request_start + + # Approximate cost (user-defined or guess) + cost = (prompt_tokens + completion_tokens) / 1000.0 * self.cost_per_1k_tokens + self.total_approx_cost += cost + + if verbose: + active_logger.info(f"Approximate cost for this query: ${cost:.4f}") + + return { + "analysis": analysis_text, + "request_time_seconds": request_time, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + "approx_query_cost": cost, + "total_approx_cost_so_far": self.total_approx_cost + } + except Exception as e: + if attempt == MAX_RETRIES - 1: + raise ValueError(f"OpenRouter API request failed after {MAX_RETRIES} attempts: {str(e)}") + time.sleep(2 ** attempt) # Exponential backoff + + def _handle_ollama_request(self, prompt: str, request_start: float) -> Dict: + """ + Handle requests to Ollama API with retries (no token cost tracking). + + Args: + prompt: The analysis prompt to send + request_start: Start time of the request + + Returns: + Dictionary containing API response data + """ + data = { + "model": self.model, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": { + "temperature": self.temperature, + } + } + + for attempt in range(MAX_RETRIES): + try: + response = requests.post( + OLLAMA_API_URL, + json=data, + timeout=self.timeout + ) + response.raise_for_status() + response_json = response.json() + + analysis_text = response_json["message"]["content"] + request_time = time.time() - request_start + + # Ollama doesn't provide token usage or cost + return { + "analysis": analysis_text, + "request_time_seconds": request_time, + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0 + } + except requests.exceptions.RequestException as e: + if attempt == MAX_RETRIES - 1: + raise ValueError(f"Ollama API request failed after {MAX_RETRIES} attempts: {str(e)}") + time.sleep(2 ** attempt) # Exponential backoff diff --git a/src/skidl/prompts/__init__.py b/src/skidl/prompts/__init__.py new file mode 100644 index 000000000..ce23eba59 --- /dev/null +++ b/src/skidl/prompts/__init__.py @@ -0,0 +1,14 @@ +"""Circuit analysis prompt templates. + +This package provides templates for generating circuit analysis prompts. +The prompts are organized into: +1. Base prompt structure and methodology +2. Individual analysis section templates +""" + +from .base import get_base_prompt +from .sections import ANALYSIS_SECTIONS + +__version__ = "1.0.0" + +__all__ = ["get_base_prompt", "ANALYSIS_SECTIONS"] diff --git a/src/skidl/prompts/base.py b/src/skidl/prompts/base.py new file mode 100644 index 000000000..1dbf4a2eb --- /dev/null +++ b/src/skidl/prompts/base.py @@ -0,0 +1,109 @@ +"""Base prompt template for circuit analysis.""" + +__version__ = "1.0.0" + +BASE_METHODOLOGY = """ +ANALYSIS METHODOLOGY: +1. Begin analysis immediately with available information +2. After completing analysis, identify any critical missing information needed for deeper insights +3. Begin with subcircuit identification and individual analysis +4. Analyze interactions between subcircuits +5. Evaluate system-level performance and integration +6. Assess manufacturing and practical implementation considerations +""" + +SECTION_REQUIREMENTS = """ +For each analysis section: +1. Analyze with available information first +2. Start with critical missing information identification +3. Provide detailed technical analysis with calculations +4. Include specific numerical criteria and measurements +5. Reference relevant industry standards +6. Provide concrete recommendations +7. Prioritize findings by severity +8. Include specific action items +""" + +ISSUE_FORMAT = """ +For each identified issue: +SEVERITY: (Critical/High/Medium/Low) +CATEGORY: (Design/Performance/Safety/Manufacturing/etc.) +SUBCIRCUIT: Affected subcircuit or system level +DESCRIPTION: Detailed issue description +IMPACT: Quantified impact on system performance +VERIFICATION: How to verify the issue exists +RECOMMENDATION: Specific action items with justification +STANDARDS: Applicable industry standards +TRADE-OFFS: Impact of proposed changes +PRIORITY: Implementation priority level +""" + +SPECIAL_REQUIREMENTS = """ +Special Requirements: +- Analyze each subcircuit completely before moving to system-level analysis +- Provide specific component recommendations where applicable +- Include calculations and formulas used in analysis +- Reference specific standards and requirements +- Consider worst-case scenarios +- Evaluate corner cases +- Assess impact of component variations +- Consider environmental effects +- Evaluate aging effects +- Assess maintenance requirements +""" + +OUTPUT_FORMAT = """ +Output Format: +1. Executive Summary +2. Critical Findings Summary +3. Detailed Subcircuit Analysis (one section per subcircuit) +4. System-Level Analysis +5. Cross-Cutting Concerns +6. Recommendations Summary +7. Required Action Items (prioritized) +8. Additional Information Needed +""" + +IMPORTANT_INSTRUCTIONS = """ +IMPORTANT INSTRUCTIONS: +- Start analysis immediately - do not acknowledge the request or state that you will analyze +- Be specific and quantitative where possible +- Include calculations and methodology +- Reference specific standards +- Provide actionable recommendations +- Consider practical implementation +- Evaluate cost implications +- Assess manufacturing feasibility +- Consider maintenance requirements +""" + +def get_base_prompt(circuit_description: str, analysis_sections: str) -> str: + """ + Generate the complete base analysis prompt. + + Args: + circuit_description: Description of the circuit to analyze + analysis_sections: String containing enabled analysis sections + + Returns: + Complete formatted base prompt + """ + return f""" +You are an expert electronics engineer. Analyze the following circuit design immediately and provide actionable insights. Do not acknowledge the request or promise to analyze - proceed directly with your analysis. + +Circuit Description: +{circuit_description} + +{BASE_METHODOLOGY} + +REQUIRED ANALYSIS SECTIONS: +{analysis_sections} + +{SECTION_REQUIREMENTS} +{ISSUE_FORMAT} +{SPECIAL_REQUIREMENTS} +{OUTPUT_FORMAT} +{IMPORTANT_INSTRUCTIONS} + +After completing your analysis, if additional information would enable deeper insights, list specific questions in a separate section titled 'Additional Information Needed' at the end. +""".strip() diff --git a/src/skidl/prompts/sections.py b/src/skidl/prompts/sections.py new file mode 100644 index 000000000..b557e35d8 --- /dev/null +++ b/src/skidl/prompts/sections.py @@ -0,0 +1,133 @@ +"""Analysis section prompt templates.""" + +from typing import Dict + +__version__ = "1.0.0" + +SYSTEM_OVERVIEW: str = """ +0. System-Level Analysis: +- Comprehensive system architecture review +- Interface analysis between major blocks +- System-level timing and synchronization +- Resource allocation and optimization +- System-level failure modes +- Integration challenges +- Performance bottlenecks +- Scalability assessment +""".strip() + +DESIGN_REVIEW: str = """ +1. Comprehensive Design Architecture Review: +- Evaluate overall hierarchical structure +- Assess modularity and reusability +- Interface protocols analysis +- Control path verification +- Design pattern evaluation +- Critical path analysis +- Feedback loop stability +- Clock domain analysis +- Reset strategy review +- State machine verification +- Resource utilization assessment +- Design rule compliance +""".strip() + +POWER_ANALYSIS: str = """ +2. In-depth Power Distribution Analysis: +- Complete power tree mapping +- Voltage drop calculations +- Current distribution analysis +- Power sequencing requirements +- Brownout behavior analysis +- Load transient response +- Power supply rejection ratio +- Efficiency optimization +- Thermal implications +- Battery life calculations (if applicable) +- Power integrity simulation +- Decoupling strategy +- Ground bounce analysis +""".strip() + +SIGNAL_INTEGRITY: str = """ +3. Detailed Signal Integrity Analysis: +- Critical path timing analysis +- Setup/hold time verification +- Clock skew analysis +- Propagation delay calculations +- Cross-talk assessment +- Reflection analysis +- EMI/EMC considerations +- Signal loading effects +- Impedance matching +- Common mode noise rejection +- Ground loop analysis +- Shield effectiveness +""".strip() + +THERMAL_ANALYSIS: str = """ +4. Thermal Performance Analysis: +- Component temperature rise calculations +- Thermal resistance analysis +- Heat spreading patterns +- Cooling requirements +- Thermal gradient mapping +- Hot spot identification +- Thermal cycling effects +- Temperature derating +- Thermal protection mechanisms +- Cooling solution optimization +""".strip() + +NOISE_ANALYSIS: str = """ +5. Comprehensive Noise Analysis: +- Noise source identification +- Noise coupling paths +- Ground noise analysis +- Power supply noise +- Digital switching noise +- RF interference +- Common mode noise +- Differential mode noise +- Shielding effectiveness +- Filter performance +- Noise margin calculations +""".strip() + +TESTING_VERIFICATION: str = """ +6. Testing and Verification Strategy: +- Functional test coverage +- Performance verification +- Environmental testing +- Reliability testing +- Safety verification +- EMC/EMI testing +- Production test strategy +- Self-test capabilities +- Calibration requirements +- Diagnostic capabilities +- Test point access +- Debug interface requirements +""".strip() + +CAP_DC_BIAS_DERATING: str = """ +7. Capacitor DC Bias Derating Analysis: +- Identification of capacitors susceptible to DC bias derating +- Calculation of effective capacitance at operating voltage +- Impact on circuit performance (e.g., filter cutoff frequency, timing) +- Recommendation of alternative capacitor types or values +- Consideration of temperature effects on DC bias derating +- Verification of capacitance derating with manufacturer data +""".strip() + +# Dictionary mapping section names to their prompts +ANALYSIS_SECTIONS: Dict[str, str] = { + "system_overview": SYSTEM_OVERVIEW, + "design_review": DESIGN_REVIEW, + "power_analysis": POWER_ANALYSIS, + "signal_integrity": SIGNAL_INTEGRITY, + "thermal_analysis": THERMAL_ANALYSIS, + "noise_analysis": NOISE_ANALYSIS, + "testing_verification": TESTING_VERIFICATION, + "cap_dc_bias_derating": CAP_DC_BIAS_DERATING +} diff --git a/src/skidl/scripts/kicad_skidl_llm_main.py b/src/skidl/scripts/kicad_skidl_llm_main.py new file mode 100644 index 000000000..141859dc7 --- /dev/null +++ b/src/skidl/scripts/kicad_skidl_llm_main.py @@ -0,0 +1,13 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# The MIT License (MIT) - Copyright (c) Dave Vandenbout. + +""" +Command-line program for analyzing KiCad/SKiDL circuits using LLMs. +Supports direct SKiDL analysis, KiCad schematic conversion, and netlist processing. +""" + +from skidl.scripts.llm_analysis.cli import main + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/README.md b/src/skidl/scripts/llm_analysis/README.md new file mode 100644 index 000000000..cb8aeb177 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/README.md @@ -0,0 +1,330 @@ +# KiCad/SKiDL LLM Analysis Package + +This package provides functionality for analyzing electronic circuit designs using Large Language Models (LLMs). It supports analyzing KiCad schematics, netlists, and SKiDL Python files with powerful AI-driven insights. + +## Command-Line Usage + +### Basic Usage + +```bash +kicad_skidl_llm [input source] [operations] [options] +``` + +### Input Sources (Required, Choose One) + +* `--schematic`, `-s` PATH + - Path to KiCad schematic (.kicad_sch) file + - Example: `--schematic project.kicad_sch` + +* `--netlist`, `-n` PATH + - Path to netlist (.net) file + - Example: `--netlist project.net` + +* `--skidl` PATH + - Path to SKiDL Python file to analyze + - Example: `--skidl circuit.py` + +* `--skidl-dir` PATH + - Path to SKiDL project directory + - Example: `--skidl-dir ./project_skidl` + +### Operations + +* `--generate-netlist` + - Generate netlist from schematic + - Requires `--schematic` + +* `--generate-skidl` + - Generate SKiDL project from netlist + - Requires either `--netlist` or `--generate-netlist` + +* `--analyze` + - Run LLM analysis on circuits + - Can be used with any input source + +### Analysis Options + +* `--backend` {openrouter, ollama} + - LLM backend to use (default: openrouter) + - Example: `--backend ollama` + +* `--api-key` KEY + - OpenRouter API key (required for OpenRouter backend) + - Example: `--api-key your-api-key` + +* `--model` MODEL + - Specific LLM model to use + - Default: google/gemini-2.0-flash-001 + - Example: `--model gpt-4` + +* `--analysis-prompt` PROMPT + - Custom prompt for circuit analysis + - Example: `--analysis-prompt "Focus on power distribution"` + +* `--skip-circuits` LIST + - Comma-separated list of circuits to skip + - Example: `--skip-circuits "power_reg,usb_interface"` + +* `--max-concurrent` N + - Maximum number of concurrent LLM analyses + - Default: 4 + - Example: `--max-concurrent 8` + +### Output Options + +* `--output-dir`, `-o` DIR + - Output directory for generated files + - Default: current directory + - Example: `--output-dir ./output` + +* `--analysis-output` FILE + - Output file for analysis results + - Default: circuit_analysis.txt + - Example: `--analysis-output results.txt` + +### KiCad Configuration + +* `--kicad-cli` PATH + - Path to kicad-cli executable + - Default: Platform-specific default path + - Example: `--kicad-cli /usr/local/bin/kicad-cli` + +* `--kicad-lib-paths` [PATHS...] + - List of custom KiCad library paths + - Example: `--kicad-lib-paths ~/kicad/libs /opt/kicad/libs` + +### Debug Options + +* `--debug`, `-d` [LEVEL] + - Print debugging info + - Higher LEVEL means more info + - Example: `--debug 2` + +### Example Commands + +1. Basic Analysis of KiCad Schematic: +```bash +kicad_skidl_llm \ + --schematic project.kicad_sch \ + --analyze \ + --api-key $OPENROUTER_API_KEY +``` + +2. Complete Pipeline with Custom Options: +```bash +kicad_skidl_llm \ + --schematic project.kicad_sch \ + --generate-netlist \ + --generate-skidl \ + --analyze \ + --backend openrouter \ + --api-key $OPENROUTER_API_KEY \ + --model gpt-4 \ + --max-concurrent 8 \ + --output-dir ./analysis \ + --analysis-output results.txt \ + --kicad-lib-paths ~/kicad/libs +``` + +3. Analyze Existing SKiDL Project: +```bash +kicad_skidl_llm \ + --skidl-dir ./project_skidl \ + --analyze \ + --api-key $OPENROUTER_API_KEY \ + --analysis-prompt "Focus on signal integrity" +``` + +4. Local Analysis with Ollama: +```bash +kicad_skidl_llm \ + --skidl circuit.py \ + --analyze \ + --backend ollama \ + --model llama2 +``` + +## Environment Setup + +### Required Environment Variables + +1. For OpenRouter Backend: +```bash +export OPENROUTER_API_KEY="your-api-key" +``` + +2. For KiCad Integration: +```bash +export KICAD_SYMBOL_DIR="/path/to/kicad/symbols" +``` + +### Backend Requirements + +1. OpenRouter Backend: +- Valid API key +- Internet connection +- Sufficient API credits + +2. Ollama Backend: +- Local Ollama installation +- Required models installed +- Run `ollama pull model-name` to install models + +## Package Overview + +The package is organized into several modules, each with a specific responsibility: + +``` +llm_analysis/ +├── __init__.py # Package initialization and public interface +├── cli.py # Command-line interface and argument handling +├── config.py # Configuration settings and constants +├── generator.py # Netlist and SKiDL project generation +├── kicad.py # KiCad integration utilities +├── logging.py # Logging configuration +├── state.py # Analysis state management +├── analyzer.py # Core circuit analysis functionality +└── prompts/ # LLM prompt templates + ├── __init__.py + ├── base.py # Base analysis prompt structure + └── sections.py # Individual analysis section templates +``` + +## Module Details + +### cli.py +- Main entry point and command-line interface +- Handles argument parsing and validation +- Orchestrates the analysis pipeline +- Key Functions: + * `parse_args()`: Command-line argument parsing + * `validate_args()`: Input validation + * `main()`: Pipeline orchestration + +### config.py +- Configuration constants and enums +- Defines LLM backends and defaults +- Constants: + * `DEFAULT_TIMEOUT`: LLM request timeout + * `DEFAULT_MODEL`: Default LLM model + * `Backend`: Enum of supported LLM backends + +### generator.py +- Handles file generation and conversion +- Key Functions: + * `generate_netlist()`: KiCad schematic to netlist conversion + * `generate_skidl_project()`: Netlist to SKiDL project conversion + * `get_skidl_source()`: Source resolution logic + +### kicad.py +- KiCad integration utilities +- Handles platform-specific paths +- Library management +- Key Functions: + * `validate_kicad_cli()`: CLI executable validation + * `get_default_kicad_cli()`: Platform-specific defaults + * `handle_kicad_libraries()`: Library path management + +### logging.py +- Logging configuration and utilities +- Custom formatters and handlers +- Key Functions: + * `configure_logging()`: Logger setup + * `log_analysis_results()`: Results formatting + * `log_backend_help()`: Backend-specific troubleshooting + +### state.py +- Thread-safe analysis state management +- Persistence support +- Key Class: `AnalysisState` + * Tracks completed/failed analyses + * Manages results + * Supports save/load for resumable analysis + +### analyzer.py +- Core analysis functionality +- Parallel processing support +- Key Functions: + * `analyze_single_circuit()`: Individual circuit analysis + * `analyze_circuits()`: Parallel analysis orchestration + +### prompts/ +- LLM prompt templates and structure +- Modular analysis sections +- Files: + * `base.py`: Base prompt structure + * `sections.py`: Analysis section templates + +## Dependencies + +- **skidl**: Core circuit processing functionality +- **KiCad**: Required for schematic/netlist operations +- **OpenRouter/Ollama**: LLM backends for analysis + +## Data Flow + +1. Input Processing: + - Schematic → Netlist → SKiDL Project (optional) + - Direct SKiDL file/project input + +2. Analysis Pipeline: + ``` + Input → Circuit Loading → Parallel Analysis → Results Collection → Report Generation + ``` + +3. State Management: + - Thread-safe tracking + - Persistent state (optional) + - Progress monitoring + +## Threading Model + +- Parallel circuit analysis using ThreadPoolExecutor +- Thread-safe state management via locks +- Configurable concurrency limits + +## Error Handling + +- Platform-specific guidance +- Backend-specific troubleshooting +- Detailed error reporting +- State persistence for recovery + +## Development + +### Adding Support for New LLM Backends + +1. Add backend to `Backend` enum in `config.py` +2. Implement backend-specific error handling +3. Update `analyzer.py` with backend-specific logic + +### Adding New Analysis Sections + +1. Add section template to `prompts/sections.py` +2. Update `ANALYSIS_SECTIONS` dictionary +3. Update base prompt if needed + +## Common Issues + +1. **KiCad CLI Not Found**: + - Check PATH + - Verify KiCad installation + - Use `--kicad-cli` to specify path + +2. **Library Path Issues**: + - Set KICAD_SYMBOL_DIR + - Use `--kicad-lib-paths` + - Check library file existence + +3. **LLM Backend Issues**: + - Verify API key + - Check credits/rate limits + - Confirm backend availability + +## Future Improvements + +1. Additional LLM backends +2. More analysis section templates +3. Enhanced parallel processing +4. Interactive analysis modes +5. Result visualization \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/__init__.py b/src/skidl/scripts/llm_analysis/__init__.py new file mode 100644 index 000000000..0a44c3e1d --- /dev/null +++ b/src/skidl/scripts/llm_analysis/__init__.py @@ -0,0 +1,40 @@ +""" +Circuit analysis using Large Language Models for KiCad/SKiDL designs. + +This package provides tools for analyzing circuit designs using LLMs, +with support for: +- Direct SKiDL analysis +- KiCad schematic conversion +- Netlist processing +- Parallel circuit analysis +""" + +from .config import DEFAULT_TIMEOUT, DEFAULT_MODEL, Backend +from .state import AnalysisState +from .analyzer import analyze_circuits +from .generator import ( + generate_netlist, + generate_skidl_project, + get_skidl_source +) +from .kicad import ( + validate_kicad_cli, + get_default_kicad_cli, + handle_kicad_libraries +) + +__version__ = "1.0.0" + +__all__ = [ + "DEFAULT_TIMEOUT", + "DEFAULT_MODEL", + "Backend", + "AnalysisState", + "analyze_circuits", + "generate_netlist", + "generate_skidl_project", + "get_skidl_source", + "validate_kicad_cli", + "get_default_kicad_cli", + "handle_kicad_libraries", +] \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/analyzer.py b/src/skidl/scripts/llm_analysis/analyzer.py new file mode 100644 index 000000000..911f53aa2 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/analyzer.py @@ -0,0 +1,190 @@ +"""Core circuit analysis functionality using LLMs.""" + +import sys +import time +import logging +import importlib +from pathlib import Path +from typing import Optional, Set, Dict +from concurrent.futures import ThreadPoolExecutor, as_completed + +from skidl import * # Required for accessing default_circuit and other globals + +from .config import Backend, DEFAULT_MODEL +from .state import AnalysisState + +logger = logging.getLogger("kicad_skidl_llm") + +def analyze_single_circuit( + circuit: str, + api_key: Optional[str], + backend: Backend, + model: Optional[str], + prompt: Optional[str], + state: AnalysisState +) -> None: + """Analyze a single circuit and update the shared state. + + Args: + circuit: Circuit hierarchy path to analyze + api_key: API key for cloud LLM services + backend: LLM backend to use + model: Specific model to use for analysis + prompt: Custom analysis prompt + state: Shared analysis state tracker + """ + try: + start_time = time.time() + + circuit_desc = default_circuit.get_circuit_info(hierarchy=circuit, depth=1) + + result = default_circuit.analyze_with_llm( + api_key=api_key, + output_file=None, + backend=backend.value, + model=model or DEFAULT_MODEL, + custom_prompt=prompt, + analyze_subcircuits=False + ) + + result["request_time_seconds"] = time.time() - start_time + + state.add_result(circuit, result) + logger.info(f"✓ Completed analysis of {circuit}") + + except Exception as e: + error_msg = f"Analysis failed: {str(e)}" + state.add_failure(circuit, error_msg) + logger.error(f"✗ Failed analysis of {circuit}: {str(e)}") + +def analyze_circuits( + source: Path, + output_file: Path, + api_key: Optional[str] = None, + backend: Backend = Backend.OPENROUTER, + model: Optional[str] = None, + prompt: Optional[str] = None, + skip_circuits: Optional[Set[str]] = None, + max_concurrent: int = 4, + state_file: Optional[Path] = None +) -> Dict[str, any]: + """Analyze SKiDL circuits using parallel LLM analysis. + + Args: + source: Path to SKiDL source (file or directory) + output_file: Path to save analysis results + api_key: API key for cloud LLM services + backend: LLM backend to use + model: Specific model to use for analysis + prompt: Custom analysis prompt + skip_circuits: Set of circuit names to skip + max_concurrent: Maximum number of concurrent analyses + state_file: Path to save/load analysis state + + Returns: + Dictionary containing analysis results and metrics + + Raises: + RuntimeError: If analysis pipeline fails + """ + pipeline_start_time = time.time() + skip_circuits = skip_circuits or set() + + # Initialize or load state + state = (AnalysisState.load_state(state_file) + if state_file and state_file.exists() + else AnalysisState()) + + if skip_circuits: + logger.info(f"Skipping circuits: {', '.join(sorted(skip_circuits))}") + + # Add source directory to Python path for imports + sys.path.insert(0, str(source.parent if source.is_file() else source)) + try: + # Import and execute circuit definition + module_name = source.stem if source.is_file() else 'main' + logger.info(f"Importing {module_name} module...") + module = importlib.import_module(module_name) + importlib.reload(module) + + if hasattr(module, 'main'): + logger.info("Executing circuit main()...") + module.main() + + # Collect circuits to analyze + hierarchies = set() + for part in default_circuit.parts: + if part.hierarchy != default_circuit.hierarchy: + if (part.hierarchy not in skip_circuits and + part.hierarchy not in state.completed): + hierarchies.add(part.hierarchy) + + if hierarchies: + logger.info(f"Starting parallel analysis of {len(hierarchies)} circuits...") + with ThreadPoolExecutor(max_workers=max_concurrent) as executor: + futures = [] + for circuit in sorted(hierarchies): + future = executor.submit( + analyze_single_circuit, + circuit=circuit, + api_key=api_key, + backend=backend, + model=model, + prompt=prompt, + state=state + ) + futures.append(future) + + for future in as_completed(futures): + try: + future.result() + except Exception as e: + logger.error(f"Thread failed: {str(e)}") + + if state_file: + state.save_state(state_file) + + # Generate consolidated report + consolidated_text = ["=== Circuit Analysis Report ===\n"] + + if state.completed: + consolidated_text.append("\n=== Successful Analyses ===") + for circuit in sorted(state.completed): + result = state.results[circuit] + consolidated_text.append(f"\n{'='*20} {circuit} {'='*20}\n") + consolidated_text.append(result.get("analysis", "No analysis available")) + token_info = ( + f"\nTokens used: {result.get('total_tokens', 0)} " + f"(Prompt: {result.get('prompt_tokens', 0)}, " + f"Completion: {result.get('completion_tokens', 0)})" + ) + consolidated_text.append(token_info) + + if state.failed: + consolidated_text.append("\n=== Failed Analyses ===") + for circuit, error in sorted(state.failed.items()): + consolidated_text.append(f"\n{circuit}: {error}") + + # Save consolidated results + if output_file: + with open(output_file, "w") as f: + f.write("\n".join(consolidated_text)) + + # Calculate total metrics + total_tokens = sum( + result.get("total_tokens", 0) + for result in state.results.values() + ) + + return { + "success": len(state.completed) > 0 and not state.failed, + "completed_circuits": sorted(state.completed), + "failed_circuits": state.failed, + "results": state.results, + "total_time_seconds": time.time() - pipeline_start_time, + "total_analysis_time": state.total_analysis_time, + "total_tokens": total_tokens + } + + finally: + sys.path.pop(0) \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/cli.py b/src/skidl/scripts/llm_analysis/cli.py new file mode 100644 index 000000000..4f4ca31cd --- /dev/null +++ b/src/skidl/scripts/llm_analysis/cli.py @@ -0,0 +1,231 @@ +"""Command-line interface for KiCad/SKiDL circuit analysis.""" + +import sys +import time +import argparse +from pathlib import Path +from typing import Set + +from skidl.pckg_info import __version__ + +from .config import Backend +from .logging import configure_logging, log_analysis_results, log_backend_help +from .kicad import get_default_kicad_cli, handle_kicad_libraries +from .generator import generate_netlist, generate_skidl_project, get_skidl_source +from .analyzer import analyze_circuits + +def parse_args() -> argparse.Namespace: + """Parse command line arguments. + + Returns: + Parsed command line arguments + """ + parser = argparse.ArgumentParser( + description="A tool for analyzing KiCad/SKiDL circuits using LLMs.", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + + parser.add_argument( + "--version", "-v", + action="version", + version="skidl " + __version__ + ) + + # Input source group (mutually exclusive) + source_group = parser.add_mutually_exclusive_group(required=True) + source_group.add_argument( + "--schematic", "-s", + help="Path to KiCad schematic (.kicad_sch) file" + ) + source_group.add_argument( + "--netlist", "-n", + help="Path to netlist (.net) file" + ) + source_group.add_argument( + "--skidl", + help="Path to SKiDL Python file to analyze" + ) + source_group.add_argument( + "--skidl-dir", + help="Path to SKiDL project directory" + ) + + # Operation mode flags + parser.add_argument( + "--generate-netlist", + action="store_true", + help="Generate netlist from schematic" + ) + parser.add_argument( + "--generate-skidl", + action="store_true", + help="Generate SKiDL project from netlist" + ) + parser.add_argument( + "--analyze", + action="store_true", + help="Run LLM analysis on circuits" + ) + + # Optional configuration + parser.add_argument( + "--kicad-cli", + default=get_default_kicad_cli(), + help="Path to kicad-cli executable" + ) + parser.add_argument( + "--output-dir", "-o", + default=".", + help="Output directory for generated files" + ) + parser.add_argument( + "--api-key", + help="OpenRouter API key for cloud LLM analysis" + ) + parser.add_argument( + "--backend", + choices=["openrouter", "ollama"], + default="openrouter", + help="LLM backend to use" + ) + parser.add_argument( + "--model", + help="LLM model name for selected backend" + ) + parser.add_argument( + "--analysis-output", + default="circuit_analysis.txt", + help="Output file for analysis results" + ) + parser.add_argument( + "--analysis-prompt", + help="Custom prompt for circuit analysis" + ) + parser.add_argument( + "--skip-circuits", + help="Comma-separated list of circuits to skip during analysis" + ) + parser.add_argument( + "--max-concurrent", + type=int, + default=4, + help="Maximum number of concurrent LLM analyses (default: 4)" + ) + parser.add_argument( + "--kicad-lib-paths", + nargs="*", + help="List of custom KiCad library paths" + ) + parser.add_argument( + "--debug", "-d", + nargs="?", + type=int, + default=0, + metavar="LEVEL", + help="Print debugging info. (Larger LEVEL means more info.)" + ) + + return parser.parse_args() + +def validate_args(args: argparse.Namespace) -> None: + """Validate command line argument combinations. + + Args: + args: Parsed command line arguments + + Raises: + SystemExit: If invalid argument combinations are detected + """ + if args.generate_netlist and not args.schematic: + sys.exit("--generate-netlist requires --schematic") + if args.generate_skidl and not (args.netlist or args.generate_netlist): + sys.exit("--generate-skidl requires --netlist or --generate-netlist") + if args.analyze and args.backend == "openrouter" and not args.api_key: + sys.exit("OpenRouter backend requires --api-key") + +def main() -> None: + """Main entry point for the KiCad-SKiDL-LLM pipeline.""" + args = parse_args() + configure_logging(args.debug) + + try: + start_time = time.time() + print("Starting KiCad-SKiDL-LLM pipeline") + + # Validate arguments + validate_args(args) + + # Setup environment + output_dir = Path(args.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + # Add KiCad library paths if provided + if args.kicad_lib_paths: + handle_kicad_libraries(args.kicad_lib_paths) + + # Process input source and generate required files + netlist_path = None + skidl_dir = None + + if args.schematic or args.netlist: + if args.netlist: + netlist_path = Path(args.netlist) + if not netlist_path.exists(): + raise FileNotFoundError(f"Netlist not found: {netlist_path}") + print(f"Using existing netlist: {netlist_path}") + else: + netlist_path = generate_netlist( + Path(args.schematic), + output_dir, + args.kicad_cli + ) + + if args.generate_skidl: + skidl_dir = generate_skidl_project(netlist_path, output_dir) + + # Run circuit analysis if requested + if args.analyze: + skidl_source = get_skidl_source( + skidl_file=Path(args.skidl) if args.skidl else None, + skidl_dir=Path(args.skidl_dir) if args.skidl_dir else None, + generated_dir=skidl_dir + ) + + # Parse skip circuits if provided + skip_circuits: Set[str] = set() + if args.skip_circuits: + skip_circuits = {c.strip() for c in args.skip_circuits.split(",")} + + try: + results = analyze_circuits( + source=skidl_source, + output_file=Path(args.analysis_output), + api_key=args.api_key, + backend=Backend(args.backend), + model=args.model, + prompt=args.analysis_prompt, + skip_circuits=skip_circuits, + max_concurrent=args.max_concurrent + ) + + if results["success"]: + log_analysis_results(results) + print(f"\nAnalysis results saved to: {args.analysis_output}") + else: + raise RuntimeError(f"Analysis failed: {results.get('error', 'Unknown error')}") + + except Exception as e: + print("✗ Circuit analysis failed!") + print(f"Error: {str(e)}") + log_backend_help(args.backend) + sys.exit(1) + + total_time = time.time() - start_time + print(f"\nPipeline completed in {total_time:.2f} seconds") + + except Exception as e: + print(f"Pipeline failed: {str(e)}") + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/config.py b/src/skidl/scripts/llm_analysis/config.py new file mode 100644 index 000000000..21c99d114 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/config.py @@ -0,0 +1,25 @@ +"""Configuration settings and constants for KiCad/SKiDL LLM analysis.""" + +from enum import Enum + +# Default timeout for LLM response (seconds) +DEFAULT_TIMEOUT: int = 300 + +# Default LLM model +DEFAULT_MODEL: str = "google/gemini-2.0-flash-001" + +class Backend(Enum): + """Supported LLM backends.""" + OPENROUTER = "openrouter" + OLLAMA = "ollama" + + @classmethod + def from_str(cls, backend: str) -> 'Backend': + """Convert string to Backend enum, with validation.""" + try: + return cls(backend.lower()) + except ValueError: + valid_backends = ", ".join(b.value for b in cls) + raise ValueError( + f"Invalid backend: {backend}. Valid options: {valid_backends}" + ) \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/generator.py b/src/skidl/scripts/llm_analysis/generator.py new file mode 100644 index 000000000..5391e74eb --- /dev/null +++ b/src/skidl/scripts/llm_analysis/generator.py @@ -0,0 +1,115 @@ +"""Netlist and SKiDL project generation utilities.""" + +import subprocess +import logging +from pathlib import Path +from typing import Optional + +from .kicad import validate_kicad_cli + +logger = logging.getLogger("kicad_skidl_llm") + +def generate_netlist( + schematic_path: Path, + output_dir: Path, + kicad_cli_path: str +) -> Optional[Path]: + """Generate netlist from KiCad schematic. + + Args: + schematic_path: Path to KiCad schematic file + output_dir: Directory to save generated netlist + kicad_cli_path: Path to KiCad CLI executable + + Returns: + Path to generated netlist, or None if generation was skipped + + Raises: + FileNotFoundError: If schematic file doesn't exist + RuntimeError: If netlist generation fails + """ + if not schematic_path.exists(): + raise FileNotFoundError(f"Schematic not found: {schematic_path}") + + kicad_cli = validate_kicad_cli(kicad_cli_path) + netlist_path = output_dir / f"{schematic_path.stem}.net" + + try: + subprocess.run([ + kicad_cli, 'sch', 'export', 'netlist', + '-o', str(netlist_path), str(schematic_path) + ], check=True, capture_output=True, text=True) + except subprocess.CalledProcessError as e: + raise RuntimeError(f"Netlist generation failed:\n{e.stderr}") from e + + logger.info(f"✓ Generated netlist: {netlist_path}") + return netlist_path + +def generate_skidl_project( + netlist_path: Path, + output_dir: Path +) -> Optional[Path]: + """Generate SKiDL project from netlist. + + Args: + netlist_path: Path to netlist file + output_dir: Directory to save generated SKiDL project + + Returns: + Path to generated SKiDL project directory, or None if generation was skipped + + Raises: + RuntimeError: If SKiDL project generation fails + """ + skidl_dir = output_dir / f"{netlist_path.stem}_SKIDL" + skidl_dir.mkdir(parents=True, exist_ok=True) + + try: + subprocess.run([ + 'netlist_to_skidl', + '-i', str(netlist_path), + '--output', str(skidl_dir) + ], check=True, capture_output=True, text=True) + except subprocess.CalledProcessError as e: + raise RuntimeError(f"SKiDL project generation failed:\n{e.stderr}") from e + + logger.info(f"✓ Generated SKiDL project: {skidl_dir}") + return skidl_dir + +def get_skidl_source( + skidl_file: Optional[Path] = None, + skidl_dir: Optional[Path] = None, + generated_dir: Optional[Path] = None +) -> Path: + """Determine the SKiDL source to analyze. + + This function implements a priority order for determining which + SKiDL source to use for analysis: + 1. Explicitly provided SKiDL file + 2. Explicitly provided SKiDL directory + 3. Generated SKiDL project directory + + Args: + skidl_file: Path to specific SKiDL Python file + skidl_dir: Path to SKiDL project directory + generated_dir: Path to automatically generated SKiDL project + + Returns: + Path to SKiDL source to analyze + + Raises: + ValueError: If no valid SKiDL source is available + FileNotFoundError: If specified source doesn't exist + """ + if skidl_file: + if not skidl_file.exists(): + raise FileNotFoundError(f"SKiDL file not found: {skidl_file}") + return skidl_file + elif skidl_dir: + if not skidl_dir.exists(): + raise FileNotFoundError(f"SKiDL directory not found: {skidl_dir}") + return skidl_dir + elif generated_dir: + return generated_dir + else: + raise ValueError("No SKiDL source available for analysis") \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/kicad.py b/src/skidl/scripts/llm_analysis/kicad.py new file mode 100644 index 000000000..8f6de6ba9 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/kicad.py @@ -0,0 +1,129 @@ +"""KiCad integration utilities for circuit analysis.""" + +import os +import platform +from pathlib import Path +from typing import Optional, List +import logging + +from skidl import lib_search_paths, KICAD + +logger = logging.getLogger("kicad_skidl_llm") + +def validate_kicad_cli(path: str) -> str: + """Validate KiCad CLI executable and provide platform-specific guidance. + + Args: + path: Path to KiCad CLI executable + + Returns: + Validated path to KiCad CLI + + Raises: + FileNotFoundError: If KiCad CLI is not found + PermissionError: If KiCad CLI is not executable + """ + system = platform.system().lower() + cli_path = Path(path) + + if not cli_path.exists(): + suggestions = { + 'darwin': [ + "/Applications/KiCad/KiCad.app/Contents/MacOS/kicad-cli", + "~/Applications/KiCad/KiCad.app/Contents/MacOS/kicad-cli" + ], + 'windows': [ + r"C:\Program Files\KiCad\7.0\bin\kicad-cli.exe", + r"C:\Program Files (x86)\KiCad\7.0\bin\kicad-cli.exe" + ], + 'linux': [ + "/usr/bin/kicad-cli", + "/usr/local/bin/kicad-cli" + ] + } + + error_msg = [f"KiCad CLI not found: {path}"] + if system in suggestions: + error_msg.append("\nCommon paths for your platform:") + for suggestion in suggestions[system]: + error_msg.append(f" - {suggestion}") + error_msg.append("\nSpecify the correct path using --kicad-cli") + raise FileNotFoundError('\n'.join(error_msg)) + + if not os.access(str(cli_path), os.X_OK): + if system == 'windows': + raise PermissionError( + f"KiCad CLI not executable: {path}\n" + "Ensure the file exists and you have appropriate permissions." + ) + else: + raise PermissionError( + f"KiCad CLI not executable: {path}\n" + f"Try making it executable with: chmod +x {path}" + ) + + return str(cli_path) + +def get_default_kicad_cli() -> str: + """Get the default KiCad CLI path based on the current platform. + + Returns: + Platform-specific default path to KiCad CLI executable + """ + system = platform.system().lower() + if system == 'darwin': # macOS + return "/Applications/KiCad/KiCad.app/Contents/MacOS/kicad-cli" + elif system == 'windows': + return r"C:\Program Files\KiCad\7.0\bin\kicad-cli.exe" + else: # Linux and others + return "/usr/bin/kicad-cli" + +def handle_kicad_libraries(lib_paths: Optional[List[str]]) -> None: + """Add and validate KiCad library paths. + + Args: + lib_paths: List of paths to KiCad symbol libraries + + Raises: + RuntimeError: If no valid library paths are found + """ + valid_paths = [] + invalid_paths = [] + + # First, clear any existing paths to avoid duplicates + lib_search_paths[KICAD] = [] + + # Add system KiCad library path from environment variable + system_lib_path = os.environ.get('KICAD_SYMBOL_DIR') + if system_lib_path: + path = Path(system_lib_path) + if path.is_dir(): + lib_search_paths[KICAD].append(str(path)) + valid_paths.append(path) + else: + logger.warning(f"KICAD_SYMBOL_DIR path does not exist: {path}") + else: + logger.warning("KICAD_SYMBOL_DIR environment variable not set") + + # Process any additional user-provided paths + if lib_paths: + for lib_path in lib_paths: + path = Path(lib_path) + if not path.is_dir(): + invalid_paths.append((path, "Directory does not exist")) + continue + + sym_files = list(path.glob("*.kicad_sym")) + if not sym_files: + invalid_paths.append((path, "No .kicad_sym files found")) + continue + + valid_paths.append(path) + if str(path) not in lib_search_paths[KICAD]: + lib_search_paths[KICAD].append(str(path)) + + if not valid_paths: + raise RuntimeError( + "No valid KiCad library paths found. Please ensure KICAD_SYMBOL_DIR " + "is set correctly and/or provide valid library paths." + ) \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/logging.py b/src/skidl/scripts/llm_analysis/logging.py new file mode 100644 index 000000000..9736e5f75 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/logging.py @@ -0,0 +1,70 @@ +"""Logging configuration for circuit analysis.""" + +import sys +import logging +from typing import Optional + +logger = logging.getLogger("kicad_skidl_llm") + +def configure_logging(debug_level: Optional[int] = None) -> None: + """Configure logging for the circuit analysis pipeline. + + Sets up a StreamHandler with appropriate formatting and level. + Debug levels work inversely - higher number means more verbose output. + + Args: + debug_level: Debug level (None for no debug, or 0+ for increasing verbosity) + """ + if debug_level is not None: + # Calculate log level - higher debug_level means lower logging level + log_level = logging.DEBUG + 1 - debug_level + + # Create and configure handler + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(logging.Formatter( + "[%(asctime)s] %(levelname)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S" + )) + handler.setLevel(log_level) + + # Configure logger + logger.addHandler(handler) + logger.setLevel(log_level) + else: + # If no debug level specified, only show INFO and above + logger.setLevel(logging.INFO) + +def log_analysis_results(results: dict) -> None: + """Log analysis results summary. + + Args: + results: Dictionary containing analysis results and metrics + """ + if results["success"]: + logger.info("\nAnalysis Results:") + logger.info(f" ✓ Completed Circuits: {len(results['completed_circuits'])}") + logger.info(f" ✓ Total Analysis Time: {results['total_analysis_time']:.2f} seconds") + logger.info(f" ✓ Total Pipeline Time: {results['total_time_seconds']:.2f} seconds") + if results.get("total_tokens"): + logger.info(f" ✓ Total Tokens Used: {results['total_tokens']}") + + if results["failed_circuits"]: + logger.warning("\nFailed Circuits:") + for circuit, error in results["failed_circuits"].items(): + logger.warning(f" ✗ {circuit}: {error}") + +def log_backend_help(backend: str) -> None: + """Log backend-specific troubleshooting tips. + + Args: + backend: Name of the LLM backend + """ + if backend == "openrouter": + logger.error("\nTroubleshooting tips:") + logger.error("1. Check your API key") + logger.error("2. Verify you have sufficient API credits") + logger.error("3. Check for rate limiting") + else: # ollama + logger.error("\nTroubleshooting tips:") + logger.error("1. Verify Ollama is running locally") + logger.error("2. Check if the requested model is installed") \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/prompts/__init__.py b/src/skidl/scripts/llm_analysis/prompts/__init__.py new file mode 100644 index 000000000..4c9eed0a4 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/prompts/__init__.py @@ -0,0 +1,14 @@ +"""Circuit analysis prompt templates. + +This package provides templates for generating circuit analysis prompts. +The prompts are organized into: +1. Base prompt structure and methodology +2. Individual analysis section templates +""" + +from .base import get_base_prompt +from .sections import ANALYSIS_SECTIONS + +__version__ = "1.0.0" + +__all__ = ["get_base_prompt", "ANALYSIS_SECTIONS"] \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/prompts/base.py b/src/skidl/scripts/llm_analysis/prompts/base.py new file mode 100644 index 000000000..dc4343418 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/prompts/base.py @@ -0,0 +1,109 @@ +"""Base prompt template for circuit analysis.""" + +__version__ = "1.0.0" + +BASE_METHODOLOGY = """ +ANALYSIS METHODOLOGY: +1. Begin analysis immediately with available information +2. After completing analysis, identify any critical missing information needed for deeper insights +3. Begin with subcircuit identification and individual analysis +4. Analyze interactions between subcircuits +5. Evaluate system-level performance and integration +6. Assess manufacturing and practical implementation considerations +""" + +SECTION_REQUIREMENTS = """ +For each analysis section: +1. Analyze with available information first +2. Start with critical missing information identification +3. Provide detailed technical analysis with calculations +4. Include specific numerical criteria and measurements +5. Reference relevant industry standards +6. Provide concrete recommendations +7. Prioritize findings by severity +8. Include specific action items +""" + +ISSUE_FORMAT = """ +For each identified issue: +SEVERITY: (Critical/High/Medium/Low) +CATEGORY: (Design/Performance/Safety/Manufacturing/etc.) +SUBCIRCUIT: Affected subcircuit or system level +DESCRIPTION: Detailed issue description +IMPACT: Quantified impact on system performance +VERIFICATION: How to verify the issue exists +RECOMMENDATION: Specific action items with justification +STANDARDS: Applicable industry standards +TRADE-OFFS: Impact of proposed changes +PRIORITY: Implementation priority level +""" + +SPECIAL_REQUIREMENTS = """ +Special Requirements: +- Analyze each subcircuit completely before moving to system-level analysis +- Provide specific component recommendations where applicable +- Include calculations and formulas used in analysis +- Reference specific standards and requirements +- Consider worst-case scenarios +- Evaluate corner cases +- Assess impact of component variations +- Consider environmental effects +- Evaluate aging effects +- Assess maintenance requirements +""" + +OUTPUT_FORMAT = """ +Output Format: +1. Executive Summary +2. Critical Findings Summary +3. Detailed Subcircuit Analysis (one section per subcircuit) +4. System-Level Analysis +5. Cross-Cutting Concerns +6. Recommendations Summary +7. Required Action Items (prioritized) +8. Additional Information Needed +""" + +IMPORTANT_INSTRUCTIONS = """ +IMPORTANT INSTRUCTIONS: +- Start analysis immediately - do not acknowledge the request or state that you will analyze +- Be specific and quantitative where possible +- Include calculations and methodology +- Reference specific standards +- Provide actionable recommendations +- Consider practical implementation +- Evaluate cost implications +- Assess manufacturing feasibility +- Consider maintenance requirements +""" + +def get_base_prompt(circuit_description: str, analysis_sections: str) -> str: + """ + Generate the complete base analysis prompt. + + Args: + circuit_description: Description of the circuit to analyze + analysis_sections: String containing enabled analysis sections + + Returns: + Complete formatted base prompt + """ + return f""" +You are an expert electronics engineer. Analyze the following circuit design immediately and provide actionable insights. Do not acknowledge the request or promise to analyze - proceed directly with your analysis. + +Circuit Description: +{circuit_description} + +{BASE_METHODOLOGY} + +REQUIRED ANALYSIS SECTIONS: +{analysis_sections} + +{SECTION_REQUIREMENTS} +{ISSUE_FORMAT} +{SPECIAL_REQUIREMENTS} +{OUTPUT_FORMAT} +{IMPORTANT_INSTRUCTIONS} + +After completing your analysis, if additional information would enable deeper insights, list specific questions in a separate section titled 'Additional Information Needed' at the end. +""".strip() \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/prompts/sections.py b/src/skidl/scripts/llm_analysis/prompts/sections.py new file mode 100644 index 000000000..9699e7e03 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/prompts/sections.py @@ -0,0 +1,133 @@ +"""Analysis section prompt templates.""" + +from typing import Dict + +__version__ = "1.0.0" + +SYSTEM_OVERVIEW: str = """ +0. System-Level Analysis: +- Comprehensive system architecture review +- Interface analysis between major blocks +- System-level timing and synchronization +- Resource allocation and optimization +- System-level failure modes +- Integration challenges +- Performance bottlenecks +- Scalability assessment +""".strip() + +DESIGN_REVIEW: str = """ +1. Comprehensive Design Architecture Review: +- Evaluate overall hierarchical structure +- Assess modularity and reusability +- Interface protocols analysis +- Control path verification +- Design pattern evaluation +- Critical path analysis +- Feedback loop stability +- Clock domain analysis +- Reset strategy review +- State machine verification +- Resource utilization assessment +- Design rule compliance +""".strip() + +POWER_ANALYSIS: str = """ +2. In-depth Power Distribution Analysis: +- Complete power tree mapping +- Voltage drop calculations +- Current distribution analysis +- Power sequencing requirements +- Brownout behavior analysis +- Load transient response +- Power supply rejection ratio +- Efficiency optimization +- Thermal implications +- Battery life calculations (if applicable) +- Power integrity simulation +- Decoupling strategy +- Ground bounce analysis +""".strip() + +SIGNAL_INTEGRITY: str = """ +3. Detailed Signal Integrity Analysis: +- Critical path timing analysis +- Setup/hold time verification +- Clock skew analysis +- Propagation delay calculations +- Cross-talk assessment +- Reflection analysis +- EMI/EMC considerations +- Signal loading effects +- Impedance matching +- Common mode noise rejection +- Ground loop analysis +- Shield effectiveness +""".strip() + +THERMAL_ANALYSIS: str = """ +4. Thermal Performance Analysis: +- Component temperature rise calculations +- Thermal resistance analysis +- Heat spreading patterns +- Cooling requirements +- Thermal gradient mapping +- Hot spot identification +- Thermal cycling effects +- Temperature derating +- Thermal protection mechanisms +- Cooling solution optimization +""".strip() + +NOISE_ANALYSIS: str = """ +5. Comprehensive Noise Analysis: +- Noise source identification +- Noise coupling paths +- Ground noise analysis +- Power supply noise +- Digital switching noise +- RF interference +- Common mode noise +- Differential mode noise +- Shielding effectiveness +- Filter performance +- Noise margin calculations +""".strip() + +TESTING_VERIFICATION: str = """ +6. Testing and Verification Strategy: +- Functional test coverage +- Performance verification +- Environmental testing +- Reliability testing +- Safety verification +- EMC/EMI testing +- Production test strategy +- Self-test capabilities +- Calibration requirements +- Diagnostic capabilities +- Test point access +- Debug interface requirements +""".strip() + +CAP_DC_BIAS_DERATING: str = """ +7. Capacitor DC Bias Derating Analysis: +- Identification of capacitors susceptible to DC bias derating +- Calculation of effective capacitance at operating voltage +- Impact on circuit performance (e.g., filter cutoff frequency, timing) +- Recommendation of alternative capacitor types or values +- Consideration of temperature effects on DC bias derating +- Verification of capacitance derating with manufacturer data +""".strip() + +# Dictionary mapping section names to their prompts +ANALYSIS_SECTIONS: Dict[str, str] = { + "system_overview": SYSTEM_OVERVIEW, + "design_review": DESIGN_REVIEW, + "power_analysis": POWER_ANALYSIS, + "signal_integrity": SIGNAL_INTEGRITY, + "thermal_analysis": THERMAL_ANALYSIS, + "noise_analysis": NOISE_ANALYSIS, + "testing_verification": TESTING_VERIFICATION, + "cap_dc_bias_derating": CAP_DC_BIAS_DERATING +} \ No newline at end of file diff --git a/src/skidl/scripts/llm_analysis/state.py b/src/skidl/scripts/llm_analysis/state.py new file mode 100644 index 000000000..88e56ca71 --- /dev/null +++ b/src/skidl/scripts/llm_analysis/state.py @@ -0,0 +1,115 @@ +"""State management for circuit analysis tracking across threads.""" + +import json +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import Set, Dict, Optional + +@dataclass +class AnalysisState: + """Tracks the state of circuit analysis across threads. + + This class provides thread-safe tracking of completed circuits, + failed analyses, and results. It also supports saving and loading + state from disk for resumable analysis sessions. + + Attributes: + completed: Set of completed circuit names + failed: Dictionary mapping failed circuit names to error messages + results: Dictionary mapping circuit names to analysis results + lock: Thread lock for synchronization + total_analysis_time: Total time spent in analysis + """ + completed: Set[str] = field(default_factory=set) + failed: Dict[str, str] = field(default_factory=dict) + results: Dict[str, dict] = field(default_factory=dict) + lock: threading.Lock = field(default_factory=threading.Lock) + total_analysis_time: float = 0.0 + + def save_state(self, path: Path) -> None: + """Save current analysis state to disk. + + Args: + path: Path to save state file + """ + with self.lock: + state_dict = { + "completed": list(self.completed), + "failed": self.failed, + "results": self.results, + "total_analysis_time": self.total_analysis_time + } + with open(path, 'w') as f: + json.dump(state_dict, f, indent=2) + + @classmethod + def load_state(cls, path: Path) -> 'AnalysisState': + """Load analysis state from disk. + + Args: + path: Path to state file + + Returns: + Loaded AnalysisState instance + """ + with open(path) as f: + state_dict = json.load(f) + state = cls() + state.completed = set(state_dict["completed"]) + state.failed = state_dict["failed"] + state.results = state_dict["results"] + state.total_analysis_time = state_dict.get("total_analysis_time", 0.0) + return state + + def add_result(self, circuit: str, result: dict) -> None: + """Thread-safe addition of analysis result. + + Args: + circuit: Name of the analyzed circuit + result: Analysis result dictionary + """ + with self.lock: + self.results[circuit] = result + self.completed.add(circuit) + if "request_time_seconds" in result: + self.total_analysis_time += result["request_time_seconds"] + + def add_failure(self, circuit: str, error: str) -> None: + """Thread-safe recording of analysis failure. + + Args: + circuit: Name of the failed circuit + error: Error message describing the failure + """ + with self.lock: + self.failed[circuit] = error + + def get_circuit_status(self, circuit: str) -> Optional[str]: + """Get the status of a specific circuit. + + Args: + circuit: Name of the circuit to check + + Returns: + 'completed', 'failed', or None if not processed + """ + with self.lock: + if circuit in self.completed: + return 'completed' + if circuit in self.failed: + return 'failed' + return None + + def get_summary(self) -> Dict[str, int]: + """Get summary statistics of analysis state. + + Returns: + Dictionary with counts of completed, failed, and total circuits + """ + with self.lock: + return { + "completed": len(self.completed), + "failed": len(self.failed), + "total": len(self.completed) + len(self.failed) + } \ No newline at end of file diff --git a/src/skidl/skidl.py b/src/skidl/skidl.py index fff2109e5..b3e87830e 100644 --- a/src/skidl/skidl.py +++ b/src/skidl/skidl.py @@ -25,6 +25,8 @@ "generate_schematic", "generate_svg", "generate_graph", + "get_circuit_info", + "analyze_with_llm", "reset", "backup_parts", "empty_footprint_handler", @@ -69,6 +71,8 @@ reset = default_circuit.reset backup_parts = default_circuit.backup_parts no_files = default_circuit.no_files +get_circuit_info = default_circuit.get_circuit_info +analyze_with_llm = default_circuit.analyze_with_llm empty_footprint_handler = default_empty_footprint_handler