Create component_search.py

component_search.py

@@ -0,0 +1,389 @@
+"""
+TinkerIQ Component Search Module
+Real component search across Digi-Key, Adafruit, and SparkFun with intelligent matching
+"""
+
+import os
+import requests
+import time
+from bs4 import BeautifulSoup
+import re
+from urllib.parse import quote
+import json
+
+class RealComponentSearch:
+    """Advanced component search with real supplier integration"""
+    
+    def __init__(self):
+        # API credentials
+        self.digikey_client_id = os.getenv("DIGIKEY_CLIENT_ID")
+        self.digikey_client_secret = os.getenv("DIGIKEY_CLIENT_SECRET")
+        self.digikey_token = None
+        self.digikey_token_expires = 0
+        
+        # API endpoints
+        self.digikey_base = "https://api.digikey.com"
+        self.adafruit_api = "https://www.adafruit.com/api/v2/products"
+        
+        # Circuit type to component mapping
+        self.circuit_components = {
+            "h_bridge": {
+                "primary": ["L298N", "L293D", "DRV8833", "TB6612FNG"],
+                "secondary": ["DC motor", "Arduino", "capacitor", "resistor"],
+                "description": "H-bridge motor control circuit"
+            },
+            "temperature_sensor": {
+                "primary": ["DHT22", "DHT11", "DS18B20", "TMP36", "LM35"],
+                "secondary": ["Arduino", "resistor", "LED", "buzzer"],
+                "description": "Temperature monitoring circuit"
+            },
+            "led_control": {
+                "primary": ["WS2812", "APA102", "LED strip", "RGB LED"],
+                "secondary": ["resistor", "capacitor", "Arduino", "ESP32"],
+                "description": "LED control and lighting circuit"
+            },
+            "distance_sensor": {
+                "primary": ["HC-SR04", "VL53L0X", "VL53L1X", "JSN-SR04T"],
+                "secondary": ["Arduino", "resistor", "buzzer", "LED"],
+                "description": "Distance measurement circuit"
+            },
+            "audio_amplifier": {
+                "primary": ["LM386", "TDA2030", "PAM8403", "MAX98357"],
+                "secondary": ["capacitor", "resistor", "potentiometer", "speaker"],
+                "description": "Audio amplification circuit"
+            },
+            "power_supply": {
+                "primary": ["LM7805", "AMS1117", "LM2596", "XL4015"],
+                "secondary": ["capacitor", "inductor", "diode", "resistor"],
+                "description": "Power regulation circuit"
+            },
+            "microcontroller": {
+                "primary": ["Arduino Uno", "ESP32", "ESP8266", "Raspberry Pi"],
+                "secondary": ["breadboard", "jumper wires", "resistor", "LED"],
+                "description": "Microcontroller development setup"
+            }
+        }
+    
+    def get_digikey_token(self):
+        """Get OAuth token for Digi-Key API"""
+        if not self.digikey_client_id or not self.digikey_client_secret:
+            return None
+            
+        if self.digikey_token and time.time() < self.digikey_token_expires:
+            return self.digikey_token
+        
+        try:
+            auth_url = f"{self.digikey_base}/v1/oauth2/token"
+            
+            data = {
+                "client_id": self.digikey_client_id,
+                "client_secret": self.digikey_client_secret,
+                "grant_type": "client_credentials"
+            }
+            
+            response = requests.post(auth_url, data=data, timeout=10)
+            response.raise_for_status()
+            
+            token_data = response.json()
+            self.digikey_token = token_data["access_token"]
+            self.digikey_token_expires = time.time() + token_data["expires_in"] - 60
+            
+            return self.digikey_token
+            
+        except Exception as e:
+            print(f"Digi-Key auth error: {e}")
+            return None
+    
+    def search_digikey(self, query, limit=3):
+        """Search Digi-Key for components"""
+        token = self.get_digikey_token()
+        if not token:
+            return []
+        
+        try:
+            search_url = f"{self.digikey_base}/products/v4/search/keyword"
+            
+            headers = {
+                "Authorization": f"Bearer {token}",
+                "X-DIGIKEY-Client-Id": self.digikey_client_id,
+                "Content-Type": "application/json"
+            }
+            
+            payload = {
+                "Keywords": query,
+                "RecordCount": limit,
+                "RecordStartPosition": 0,
+                "Filters": {},
+                "Sort": {
+                    "SortOption": "SortByUnitPrice",
+                    "Direction": "Ascending"
+                },
+                "RequestedQuantity": 1
+            }
+            
+            response = requests.post(search_url, headers=headers, json=payload, timeout=15)
+            response.raise_for_status()
+            
+            data = response.json()
+            components = []
+            
+            for product in data.get("Products", [])[:limit]:
+                try:
+                    component = {
+                        "name": product.get("ProductDescription", "Unknown"),
+                        "part_number": product.get("DigiKeyPartNumber", ""),
+                        "manufacturer": product.get("Manufacturer", {}).get("Name", "Unknown"),
+                        "price": self._format_price(product.get("UnitPrice", 0)),
+                        "stock": product.get("QuantityAvailable", 0),
+                        "supplier": "Digi-Key",
+                        "datasheet": product.get("PrimaryDatasheet", ""),
+                        "description": product.get("DetailedDescription", "")[:100] + "...",
+                        "url": f"https://www.digikey.com/en/products/detail/{product.get('DigiKeyPartNumber', '')}"
+                    }
+                    components.append(component)
+                except Exception as e:
+                    print(f"Error processing Digi-Key product: {e}")
+                    continue
+            
+            return components
+            
+        except Exception as e:
+            print(f"Digi-Key search error: {e}")
+            return []
+    
+    def search_adafruit(self, query, limit=3):
+        """Search Adafruit products"""
+        try:
+            url = f"{self.adafruit_api}?category_id=&search={quote(query)}"
+            
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+            
+            data = response.json()
+            components = []
+            
+            for product in data.get("products", [])[:limit]:
+                try:
+                    component = {
+                        "name": product.get("name", "Unknown"),
+                        "part_number": f"ADA-{product.get('id', '')}",
+                        "manufacturer": "Adafruit",
+                        "price": f"${product.get('price', 0):.2f}",
+                        "stock": "In Stock" if product.get("in_stock", False) else "Out of Stock",
+                        "supplier": "Adafruit",
+                        "datasheet": "",
+                        "description": BeautifulSoup(product.get("description_text", ""), "html.parser").get_text()[:100] + "...",
+                        "url": f"https://www.adafruit.com/product/{product.get('id', '')}"
+                    }
+                    components.append(component)
+                except Exception as e:
+                    print(f"Error processing Adafruit product: {e}")
+                    continue
+            
+            return components
+            
+        except Exception as e:
+            print(f"Adafruit search error: {e}")
+            return []
+    
+    def search_sparkfun(self, query, limit=3):
+        """Search SparkFun products via web scraping"""
+        try:
+            search_url = "https://www.sparkfun.com/search/results"
+            params = {"term": query, "page": 1, "per_page": limit}
+            
+            headers = {
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+            }
+            
+            response = requests.get(search_url, params=params, headers=headers, timeout=10)
+            response.raise_for_status()
+            
+            # Mock SparkFun results (replace with actual parsing)
+            components = []
+            sparkfun_mock = {
+                "temperature": [
+                    {
+                        "name": "SparkFun Temperature Sensor Breakout - TMP102",
+                        "part_number": "SEN-13314",
+                        "manufacturer": "SparkFun",
+                        "price": "$5.95",
+                        "stock": "In Stock",
+                        "supplier": "SparkFun",
+                        "datasheet": "",
+                        "description": "Digital temperature sensor with I2C interface",
+                        "url": "https://www.sparkfun.com/products/13314"
+                    }
+                ],
+                "motor": [
+                    {
+                        "name": "SparkFun Motor Driver - Dual TB6612FNG",
+                        "part_number": "ROB-14451",
+                        "manufacturer": "SparkFun", 
+                        "price": "$6.95",
+                        "stock": "In Stock",
+                        "supplier": "SparkFun",
+                        "datasheet": "",
+                        "description": "Dual motor driver breakout board",
+                        "url": "https://www.sparkfun.com/products/14451"
+                    }
+                ]
+            }
+            
+            query_lower = query.lower()
+            if "temperature" in query_lower:
+                components.extend(sparkfun_mock["temperature"])
+            elif "motor" in query_lower:
+                components.extend(sparkfun_mock["motor"])
+            
+            return components[:limit]
+            
+        except Exception as e:
+            print(f"SparkFun search error: {e}")
+            return []
+    
+    def analyze_circuit_context(self, circuit_analysis):
+        """Intelligently determine circuit type from analysis text"""
+        analysis_lower = circuit_analysis.lower()
+        
+        # Score each circuit type based on keyword matches
+        circuit_scores = {}
+        
+        for circuit_type, components in self.circuit_components.items():
+            score = 0
+            
+            # Check for primary components (higher weight)
+            for primary in components["primary"]:
+                if primary.lower() in analysis_lower:
+                    score += 5
+            
+            # Check for secondary components (lower weight)
+            for secondary in components["secondary"]:
+                if secondary.lower() in analysis_lower:
+                    score += 1
+            
+            # Check circuit type keywords
+            circuit_keywords = {
+                "h_bridge": ["h-bridge", "h bridge", "motor control", "motor driver", "robot", "drive"],
+                "temperature_sensor": ["temperature", "thermal", "climate", "heat", "dht", "thermometer"],
+                "led_control": ["led", "light", "rgb", "strip", "neopixel", "lighting"],
+                "distance_sensor": ["distance", "ultrasonic", "proximity", "range", "sonar"],
+                "audio_amplifier": ["audio", "amplifier", "speaker", "sound", "music"],
+                "power_supply": ["power", "voltage", "regulator", "supply", "battery"],
+                "microcontroller": ["arduino", "esp32", "microcontroller", "mcu", "development"]
+            }
+            
+            if circuit_type in circuit_keywords:
+                for keyword in circuit_keywords[circuit_type]:
+                    if keyword in analysis_lower:
+                        score += 3
+            
+            circuit_scores[circuit_type] = score
+        
+        # Return the circuit type with highest score
+        if circuit_scores:
+            best_match = max(circuit_scores.items(), key=lambda x: x[1])
+            if best_match[1] > 0:  # Only return if we have a positive score
+                return best_match[0]
+        
+        return "general"
+    
+    def get_contextual_components(self, circuit_analysis):
+        """Get components specifically relevant to the analyzed circuit"""
+        circuit_type = self.analyze_circuit_context(circuit_analysis)
+        
+        if circuit_type == "general":
+            # If we can't determine circuit type, search for general terms
+            search_terms = ["Arduino", "breadboard", "jumper wires"]
+        else:
+            # Use circuit-specific components
+            circuit_info = self.circuit_components[circuit_type]
+            search_terms = circuit_info["primary"][:2]  # Top 2 primary components
+        
+        all_components = []
+        
+        # Search each supplier for relevant components
+        for term in search_terms:
+            # Digi-Key search
+            digi_results = self.search_digikey(term, 2)
+            all_components.extend(digi_results)
+            
+            # Adafruit search
+            ada_results = self.search_adafruit(term, 2)
+            all_components.extend(ada_results)
+            
+            # SparkFun search
+            sf_results = self.search_sparkfun(term, 1)
+            all_components.extend(sf_results)
+        
+        # Remove duplicates and return top results
+        unique_components = self._deduplicate_components(all_components)
+        return {
+            "circuit_type": circuit_type,
+            "components": unique_components[:6],  # Top 6 components
+            "description": self.circuit_components.get(circuit_type, {}).get("description", "General electronics circuit")
+        }
+    
+    def _deduplicate_components(self, components):
+        """Remove similar components"""
+        unique = []
+        seen_names = set()
+        
+        for comp in components:
+            # Normalize name for comparison
+            norm_name = re.sub(r'[^a-zA-Z0-9]', '', comp['name'].lower())
+            
+            # Simple duplicate detection
+            is_duplicate = False
+            for seen in seen_names:
+                if self._similarity(norm_name, seen) > 0.7:
+                    is_duplicate = True
+                    break
+            
+            if not is_duplicate:
+                unique.append(comp)
+                seen_names.add(norm_name)
+        
+        return unique
+    
+    def _similarity(self, a, b):
+        """Simple string similarity"""
+        if not a or not b:
+            return 0
+        
+        set_a = set(a[i:i+2] for i in range(len(a)-1))
+        set_b = set(b[i:i+2] for i in range(len(b)-1))
+        
+        if not set_a and not set_b:
+            return 1
+        if not set_a or not set_b:
+            return 0
+        
+        intersection = len(set_a & set_b)
+        union = len(set_a | set_b)
+        
+        return intersection / union if union > 0 else 0
+    
+    def _format_price(self, price):
+        """Format price consistently"""
+        if isinstance(price, (int, float)):
+            return f"${price:.2f}"
+        elif isinstance(price, str):
+            return price if price.startswith('$') else f"${price}"
+        else:
+            return "Price unavailable"
+
+# Test function
+if __name__ == "__main__":
+    search = RealComponentSearch()
+    
+    # Test circuit analysis
+    test_analysis = "This circuit shows an H-bridge motor driver using an L298N chip connected to an Arduino Uno. The circuit is designed to control two DC motors for a robot car project."
+    
+    results = search.get_contextual_components(test_analysis)
+    print(f"Circuit Type: {results['circuit_type']}")
+    print(f"Description: {results['description']}")
+    print(f"Found {len(results['components'])} components:")
+    
+    for comp in results['components']:
+        print(f"- {comp['name']} | {comp['price']} | {comp['supplier']}")