From dab9c6848e20a9234e0da11972fa50b160770238 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Tue, 24 Feb 2026 13:33:34 -0700 Subject: [PATCH] Add Espressif product catalog integration 5 new MCP tools for chip/module discovery, comparison, and procurement planning using Espressif's public product catalog API. Data is lazily fetched and cached in-memory with a 24h TTL. Tools: esp_product_search, esp_product_info, esp_chip_compare, esp_product_recommend, esp_product_availability Resource: esp://products/catalog --- pyproject.toml | 1 + src/mcesptool/components/__init__.py | 3 + src/mcesptool/components/product_catalog.py | 619 ++++++++++++++++++++ src/mcesptool/server.py | 13 +- tests/test_product_catalog.py | 514 ++++++++++++++++ uv.lock | 2 + 6 files changed, 1151 insertions(+), 1 deletion(-) create mode 100644 src/mcesptool/components/product_catalog.py create mode 100644 tests/test_product_catalog.py diff --git a/pyproject.toml b/pyproject.toml index 0d20863..4de5438 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ "pyserial>=3.5", # Serial communication "pyserial-asyncio>=0.6", # Async serial support "thefuzz[speedup]>=0.22.1", # Fuzzy string matching + "httpx>=0.28", # Async HTTP client (product catalog) "pydantic>=2.11.7", # Data validation "click>=8.1.0", # CLI framework "rich>=13.9.4", # Rich console output diff --git a/src/mcesptool/components/__init__.py b/src/mcesptool/components/__init__.py index f6dc441..3e66bd6 100644 --- a/src/mcesptool/components/__init__.py +++ b/src/mcesptool/components/__init__.py @@ -11,6 +11,7 @@ from .firmware_builder import FirmwareBuilder from .flash_manager import FlashManager from .ota_manager import OTAManager from .partition_manager import PartitionManager +from .product_catalog import ProductCatalog from .production_tools import ProductionTools from .qemu_manager import QemuManager from .security_manager import SecurityManager @@ -26,6 +27,7 @@ COMPONENT_REGISTRY = { "production_tools": ProductionTools, "diagnostics": Diagnostics, "qemu_manager": QemuManager, + "product_catalog": ProductCatalog, } __all__ = [ @@ -38,5 +40,6 @@ __all__ = [ "ProductionTools", "Diagnostics", "QemuManager", + "ProductCatalog", "COMPONENT_REGISTRY", ] diff --git a/src/mcesptool/components/product_catalog.py b/src/mcesptool/components/product_catalog.py new file mode 100644 index 0000000..fecd06d --- /dev/null +++ b/src/mcesptool/components/product_catalog.py @@ -0,0 +1,619 @@ +""" +Product Catalog Component + +Provides access to Espressif's public product catalog API for chip/module +discovery, comparison, and procurement planning. Data is lazily fetched +and cached in-memory with a 24-hour TTL. +""" + +import json +import logging +import re +import time +from typing import Any + +import httpx +from fastmcp import Context, FastMCP + +from ..config import ESPToolServerConfig + +logger = logging.getLogger(__name__) + +CATALOG_URL = "https://products.espressif.com/api/user/products?language=en" + + +def _parse_memory_field(value: str | int | None) -> tuple[int, str]: + """Parse compound memory strings like '4, Quad' into (size_mb, interface). + + Returns (0, '') for empty/unparsable values. Handles bare ints from the API. + """ + if isinstance(value, int): + return value, "" + if not value or value in ("N/A", "NA", "-", "—"): + return 0, "" + parts = [p.strip() for p in value.split(",")] + try: + size = int(parts[0]) + except (ValueError, IndexError): + return 0, "" + interface = parts[1] if len(parts) > 1 else "" + return size, interface + + +def _parse_sram_kb(value: str | int | None) -> int: + """Parse SRAM field — may be plain int or compound string.""" + if isinstance(value, int): + return value + if not value or value in ("N/A", "NA", "-", "—"): + return 0 + parts = value.split(",") + try: + return int(parts[0].strip()) + except (ValueError, IndexError): + return 0 + + +def _has_capability(value: str) -> bool: + """Check if a feature field indicates availability.""" + if not value: + return False + v = value.strip().lower() + return v not in ("", "n/a", "na", "-", "—", "no", "0") + + +def _normalize_status(value: str) -> str: + """Normalize status strings for consistent filtering.""" + mapping = { + "mass production": "Mass Production", + "nrnd": "NRND", + "eol": "EOL", + "replaced": "Replaced", + "sample": "Sample", + } + return mapping.get(value.strip().lower(), value.strip()) + + +def _product_summary(product: dict) -> dict[str, Any]: + """Extract a concise summary of a product for search results.""" + flash_mb, flash_type = _parse_memory_field(product.get("flash", "")) + psram_mb, psram_type = _parse_memory_field(product.get("psram", "")) + return { + "name": product.get("name", ""), + "type": product.get("type", ""), + "series": product.get("seriesName", ""), + "status": _normalize_status(product.get("status", "")), + "wifi": _has_capability(product.get("wifi", "")), + "bluetooth": _has_capability(product.get("bluetooth", "")), + "thread_zigbee": _has_capability(product.get("threadZigbee", "")), + "flash_mb": flash_mb, + "psram_mb": psram_mb, + "sram_kb": _parse_sram_kb(product.get("sram", "")), + "gpio": int(product.get("gpio", 0) or 0), + "antenna": product.get("antenna", ""), + } + + +def _product_detail(product: dict) -> dict[str, Any]: + """Full formatted product record.""" + flash_mb, flash_type = _parse_memory_field(product.get("flash", "")) + psram_mb, psram_type = _parse_memory_field(product.get("psram", "")) + return { + "name": product.get("name", ""), + "name_new": product.get("nameNew", ""), + "type": product.get("type", ""), + "series": product.get("seriesName", ""), + "status": _normalize_status(product.get("status", "")), + "mpn": product.get("mpn", ""), + "dimensions": product.get("dimensions", ""), + "wifi": product.get("wifi", ""), + "wifi6": product.get("wifi6", ""), + "bluetooth": product.get("bluetooth", ""), + "thread_zigbee": product.get("threadZigbee", ""), + "pins": product.get("pins", ""), + "frequency_mhz": product.get("freq", ""), + "sram_kb": _parse_sram_kb(product.get("sram", "")), + "rom_kb": product.get("rom", ""), + "flash_mb": flash_mb, + "flash_type": flash_type, + "psram_mb": psram_mb, + "psram_type": psram_type, + "gpio": int(product.get("gpio", 0) or 0), + "operating_temp": product.get("operatingTemp", ""), + "voltage_range": product.get("voltageRange", ""), + "antenna": product.get("antenna", ""), + "size_type": product.get("sizeType", ""), + "release_time": product.get("releaseTime", ""), + "idf_supports": product.get("idfSupports", ""), + "spq": product.get("spq", ""), + "moq": product.get("moq", ""), + "lead_time": product.get("leadTime", ""), + "eccn_code": product.get("eccnCode", ""), + "ccatsz_code": product.get("ccatszCode", ""), + "hs_code": product.get("hsCode", ""), + "pre_firmware": product.get("preFirmware", ""), + "replaced_by": product.get("replacedByName", ""), + } + + +def _parse_lead_weeks(lead_time: str) -> int | None: + """Extract numeric weeks from lead time strings like '8-10 weeks'.""" + if not lead_time: + return None + m = re.search(r"(\d+)", lead_time) + return int(m.group(1)) if m else None + + +def _parse_temp_max(temp_str: str) -> int | None: + """Extract max operating temperature from strings like '-40~85C'.""" + if not temp_str: + return None + m = re.search(r"~\s*(\d+)", temp_str) + if m: + return int(m.group(1)) + m = re.search(r"(\d+)\s*[Cc]", temp_str) + return int(m.group(1)) if m else None + + +class ProductCatalog: + """Espressif product catalog — chip/module discovery, comparison, and procurement""" + + def __init__(self, app: FastMCP, config: ESPToolServerConfig) -> None: + self.app = app + self.config = config + self._products: list[dict] = [] + self._last_fetch: float = 0 + self._cache_ttl: int = 86400 # 24 hours + self._register_tools() + self._register_resources() + + async def _ensure_data(self, force_refresh: bool = False) -> list[dict]: + """Lazy-load and cache the product catalog from Espressif's API.""" + now = time.time() + if self._products and not force_refresh and (now - self._last_fetch) < self._cache_ttl: + return self._products + + logger.info("Fetching Espressif product catalog...") + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.get(CATALOG_URL) + resp.raise_for_status() + data = resp.json() + + self._products = data.get("results", []) + self._last_fetch = time.time() + logger.info("Cached %d products from Espressif catalog", len(self._products)) + return self._products + + def _fuzzy_find(self, products: list[dict], query: str, threshold: int = 60) -> list[dict]: + """Fuzzy-match products by name or MPN.""" + from thefuzz import fuzz + + scored = [] + q = query.lower() + for p in products: + name = (p.get("name") or "").lower() + mpn = (p.get("mpn") or "").lower() + if name == q or mpn == q: + scored.append((300, p)) + elif q in name or q in mpn: + scored.append((200, p)) + elif name in q or mpn in q: + scored.append((150, p)) + else: + score = max(fuzz.partial_ratio(q, name), fuzz.partial_ratio(q, mpn)) + if score >= threshold: + scored.append((score, p)) + scored.sort(key=lambda x: x[0], reverse=True) + return [p for _, p in scored] + + def _register_resources(self) -> None: + """Register MCP resources.""" + + @self.app.resource("esp://products/catalog") + async def product_catalog_resource() -> str: + """Complete Espressif product catalog — SoCs and modules with full specs""" + products = await self._ensure_data() + return json.dumps([_product_summary(p) for p in products]) + + def _register_tools(self) -> None: + """Register product catalog tools with FastMCP.""" + + @self.app.tool("esp_product_search") + async def product_search( + context: Context, + series: str | None = None, + product_type: str | None = None, + wifi: bool | None = None, + bluetooth: bool | None = None, + thread_zigbee: bool | None = None, + min_flash_mb: int | None = None, + min_psram_mb: int | None = None, + min_sram_kb: int | None = None, + min_gpio: int | None = None, + max_temp_c: int | None = None, + status: str | None = None, + antenna: str | None = None, + keyword: str | None = None, + ) -> dict[str, Any]: + """Search the Espressif product catalog by specs, capabilities, or keyword. + + Filter by chip family, connectivity, memory, GPIO count, temperature + range, production status, antenna type, or free-text keyword. All + filters are optional and combine with AND logic. Returns matching + products with key specs. + + Args: + series: Chip family filter, e.g. "ESP32-S3", "ESP32-C6" + product_type: "SoC" or "Module" + wifi: Filter to products with Wi-Fi + bluetooth: Filter to products with Bluetooth + thread_zigbee: Filter to products with Thread/Zigbee (802.15.4) + min_flash_mb: Minimum flash size in MB + min_psram_mb: Minimum PSRAM size in MB + min_sram_kb: Minimum SRAM size in KB + min_gpio: Minimum GPIO pin count + max_temp_c: Required upper operating temperature (e.g. 105 for industrial) + status: Production status — "Mass Production", "NRND", "EOL", "Sample" + antenna: Antenna type — "PCB", "IPEX", etc. + keyword: Fuzzy search on product name or MPN + """ + return await self._product_search_impl( + context, + series=series, + product_type=product_type, + wifi=wifi, + bluetooth=bluetooth, + thread_zigbee=thread_zigbee, + min_flash_mb=min_flash_mb, + min_psram_mb=min_psram_mb, + min_sram_kb=min_sram_kb, + min_gpio=min_gpio, + max_temp_c=max_temp_c, + status=status, + antenna=antenna, + keyword=keyword, + ) + + @self.app.tool("esp_product_info") + async def product_info( + context: Context, + name: str, + ) -> dict[str, Any]: + """Get detailed specifications for a specific Espressif product. + + Looks up a product by name or MPN (fuzzy-matched) and returns + the full spec sheet: memory, connectivity, GPIO, temperature + range, procurement details, ECCN/HS codes, and more. + + Args: + name: Product name or MPN, e.g. "ESP32-S3-WROOM-1" (fuzzy-matched) + """ + return await self._product_info_impl(context, name) + + @self.app.tool("esp_chip_compare") + async def chip_compare( + context: Context, + products: list[str], + ) -> dict[str, Any]: + """Compare 2-4 Espressif products side-by-side. + + Produces a comparison table highlighting differences in connectivity, + memory, GPIO, temperature range, and procurement status. Useful for + deciding between chip families or module variants. + + Args: + products: List of 2-4 product names to compare (fuzzy-matched) + """ + return await self._chip_compare_impl(context, products) + + @self.app.tool("esp_product_recommend") + async def product_recommend( + context: Context, + use_case: str, + constraints: str | None = None, + prefer_module: bool = True, + ) -> dict[str, Any]: + """Get chip/module recommendations for a use case. + + Describe what you're building and any constraints, and this tool + filters the Espressif catalog to matching products ranked by fit. + Only returns products in Mass Production by default. + + Args: + use_case: What you're building, e.g. "battery-powered BLE sensor" + constraints: Technical constraints, e.g. "needs PSRAM, temp to 105C" + prefer_module: Prefer modules over bare SoCs (default: true) + """ + return await self._product_recommend_impl(context, use_case, constraints, prefer_module) + + @self.app.tool("esp_product_availability") + async def product_availability( + context: Context, + series: str | None = None, + status: str = "Mass Production", + max_lead_weeks: int | None = None, + max_moq: int | None = None, + ) -> dict[str, Any]: + """Check product availability and procurement details. + + Filter the catalog by production status, lead time, and MOQ + to find products you can actually source. Useful for procurement + planning and identifying alternatives for EOL/NRND parts. + + Args: + series: Chip family filter, e.g. "ESP32-S3" + status: Production status filter (default: "Mass Production") + max_lead_weeks: Maximum acceptable lead time in weeks + max_moq: Maximum acceptable minimum order quantity + """ + return await self._product_availability_impl( + context, series, status, max_lead_weeks, max_moq + ) + + # ── Tool implementations ──────────────────────────────────────────── + + async def _product_search_impl( + self, + context: Context, + series: str | None = None, + product_type: str | None = None, + wifi: bool | None = None, + bluetooth: bool | None = None, + thread_zigbee: bool | None = None, + min_flash_mb: int | None = None, + min_psram_mb: int | None = None, + min_sram_kb: int | None = None, + min_gpio: int | None = None, + max_temp_c: int | None = None, + status: str | None = None, + antenna: str | None = None, + keyword: str | None = None, + ) -> dict[str, Any]: + products = await self._ensure_data() + + # Start with keyword pre-filter if provided + if keyword: + candidates = self._fuzzy_find(products, keyword, threshold=50) + else: + candidates = products + + results = [] + for p in candidates: + if series and series.lower() not in (p.get("seriesName") or "").lower(): + continue + if product_type and (p.get("type") or "").lower() != product_type.lower(): + continue + if wifi is True and not _has_capability(p.get("wifi", "")): + continue + if wifi is False and _has_capability(p.get("wifi", "")): + continue + if bluetooth is True and not _has_capability(p.get("bluetooth", "")): + continue + if bluetooth is False and _has_capability(p.get("bluetooth", "")): + continue + if thread_zigbee is True and not _has_capability(p.get("threadZigbee", "")): + continue + if thread_zigbee is False and _has_capability(p.get("threadZigbee", "")): + continue + flash_mb, _ = _parse_memory_field(p.get("flash", "")) + if min_flash_mb is not None and flash_mb < min_flash_mb: + continue + psram_mb, _ = _parse_memory_field(p.get("psram", "")) + if min_psram_mb is not None and psram_mb < min_psram_mb: + continue + sram_kb = _parse_sram_kb(p.get("sram", "")) + if min_sram_kb is not None and sram_kb < min_sram_kb: + continue + gpio = int(p.get("gpio", 0) or 0) + if min_gpio is not None and gpio < min_gpio: + continue + if max_temp_c is not None: + temp_max = _parse_temp_max(p.get("operatingTemp", "")) + if temp_max is None or temp_max < max_temp_c: + continue + if status and _normalize_status(p.get("status", "")) != _normalize_status(status): + continue + if antenna and antenna.lower() not in (p.get("antenna") or "").lower(): + continue + + results.append(_product_summary(p)) + + return { + "count": len(results), + "filters_applied": { + k: v + for k, v in { + "series": series, + "product_type": product_type, + "wifi": wifi, + "bluetooth": bluetooth, + "thread_zigbee": thread_zigbee, + "min_flash_mb": min_flash_mb, + "min_psram_mb": min_psram_mb, + "min_sram_kb": min_sram_kb, + "min_gpio": min_gpio, + "max_temp_c": max_temp_c, + "status": status, + "antenna": antenna, + "keyword": keyword, + }.items() + if v is not None + }, + "products": results, + } + + async def _product_info_impl( + self, context: Context, name: str + ) -> dict[str, Any]: + products = await self._ensure_data() + matches = self._fuzzy_find(products, name, threshold=65) + if not matches: + return {"error": f"No product found matching '{name}'", "suggestion": "Try esp_product_search with a keyword to browse available products"} + return _product_detail(matches[0]) + + async def _chip_compare_impl( + self, context: Context, products: list[str] + ) -> dict[str, Any]: + if len(products) < 2: + return {"error": "Provide at least 2 product names to compare"} + if len(products) > 4: + return {"error": "Compare up to 4 products at a time"} + + all_products = await self._ensure_data() + resolved = [] + not_found = [] + for name in products: + matches = self._fuzzy_find(all_products, name, threshold=65) + if matches: + resolved.append(_product_detail(matches[0])) + else: + not_found.append(name) + + if not_found: + return {"error": f"Could not find: {', '.join(not_found)}", "found": [r["name"] for r in resolved]} + + # Build comparison — highlight fields that differ + all_keys = list(resolved[0].keys()) + differences = {} + common = {} + for key in all_keys: + values = [r[key] for r in resolved] + if len({str(v) for v in values}) > 1: + differences[key] = {r["name"]: r[key] for r in resolved} + else: + common[key] = values[0] + + return { + "products_compared": [r["name"] for r in resolved], + "differences": differences, + "common": common, + "full_specs": {r["name"]: r for r in resolved}, + } + + async def _product_recommend_impl( + self, + context: Context, + use_case: str, + constraints: str | None, + prefer_module: bool, + ) -> dict[str, Any]: + products = await self._ensure_data() + + # Parse use case and constraints into filter criteria + text = f"{use_case} {constraints or ''}".lower() + + wants_wifi = any(w in text for w in ("wifi", "wi-fi", "wireless", "iot", "mqtt", "http")) + wants_ble = any(w in text for w in ("ble", "bluetooth", "beacon")) + wants_thread = any(w in text for w in ("thread", "zigbee", "matter", "802.15.4")) + wants_psram = "psram" in text + wants_industrial = any(w in text for w in ("industrial", "105", "automotive", "harsh")) + + # Extract numeric constraints from text + min_flash = None + m = re.search(r"(\d+)\s*mb\s*flash", text) + if m: + min_flash = int(m.group(1)) + + min_psram = None + m = re.search(r"(\d+)\s*mb\s*psram", text) + if m: + min_psram = int(m.group(1)) + + candidates = [] + for p in products: + # Only recommend actively produced parts + if _normalize_status(p.get("status", "")) != "Mass Production": + continue + if prefer_module and (p.get("type") or "").lower() != "module": + continue + if wants_wifi and not _has_capability(p.get("wifi", "")): + continue + if wants_ble and not _has_capability(p.get("bluetooth", "")): + continue + if wants_thread and not _has_capability(p.get("threadZigbee", "")): + continue + + flash_mb, _ = _parse_memory_field(p.get("flash", "")) + psram_mb, _ = _parse_memory_field(p.get("psram", "")) + + if wants_psram and psram_mb == 0: + continue + if min_flash is not None and flash_mb < min_flash: + continue + if min_psram is not None and psram_mb < min_psram: + continue + if wants_industrial: + temp_max = _parse_temp_max(p.get("operatingTemp", "")) + if temp_max is None or temp_max < 105: + continue + + candidates.append(_product_summary(p)) + + # If module preference yields nothing, fall back to including SoCs + if not candidates and prefer_module: + return await self._product_recommend_impl(context, use_case, constraints, False) + + return { + "use_case": use_case, + "constraints_parsed": { + "wifi": wants_wifi, + "bluetooth": wants_ble, + "thread_zigbee": wants_thread, + "psram_required": wants_psram, + "industrial_temp": wants_industrial, + "min_flash_mb": min_flash, + "min_psram_mb": min_psram, + "prefer_module": prefer_module, + }, + "count": len(candidates), + "recommendations": candidates[:15], + } + + async def _product_availability_impl( + self, + context: Context, + series: str | None, + status: str, + max_lead_weeks: int | None, + max_moq: int | None, + ) -> dict[str, Any]: + products = await self._ensure_data() + results = [] + + for p in products: + if series and series.lower() not in (p.get("seriesName") or "").lower(): + continue + if _normalize_status(p.get("status", "")) != _normalize_status(status): + continue + + lead_weeks = _parse_lead_weeks(p.get("leadTime", "")) + if max_lead_weeks is not None and lead_weeks is not None and lead_weeks > max_lead_weeks: + continue + + moq_val = p.get("moq", "") + moq_int = None + if moq_val: + m = re.search(r"(\d+)", str(moq_val)) + if m: + moq_int = int(m.group(1)) + if max_moq is not None and moq_int is not None and moq_int > max_moq: + continue + + results.append({ + **_product_summary(p), + "lead_time": p.get("leadTime", ""), + "moq": p.get("moq", ""), + "spq": p.get("spq", ""), + "mpn": p.get("mpn", ""), + "replaced_by": p.get("replacedByName", ""), + }) + + return { + "count": len(results), + "filters": { + "series": series, + "status": status, + "max_lead_weeks": max_lead_weeks, + "max_moq": max_moq, + }, + "products": results, + } diff --git a/src/mcesptool/server.py b/src/mcesptool/server.py index 0ebb94d..8efd5d8 100644 --- a/src/mcesptool/server.py +++ b/src/mcesptool/server.py @@ -22,6 +22,7 @@ from .components import ( FlashManager, OTAManager, PartitionManager, + ProductCatalog, ProductionTools, QemuManager, SecurityManager, @@ -81,6 +82,9 @@ class ESPToolServer: self.components["production_tools"] = ProductionTools(self.app, self.config) self.components["diagnostics"] = Diagnostics(self.app, self.config) + # Product catalog (no hardware dependency — always enabled) + self.components["product_catalog"] = ProductCatalog(self.app, self.config) + # QEMU emulation (if available) if self.config.get_qemu_available(): self.components["qemu_manager"] = QemuManager(self.app, self.config) @@ -114,7 +118,7 @@ class ESPToolServer: # Get tool and resource counts via public API tools = await self.app.get_tools() # Note: FastMCP doesn't expose get_resources(), so we count our components - resource_count = 3 # esp://server/status, esp://config, esp://capabilities + resource_count = 4 # esp://server/status, esp://config, esp://capabilities, esp://products/catalog return { "server_name": "MCP ESPTool Server", @@ -175,6 +179,13 @@ class ESPToolServer: "esp_performance_profile", "esp_diagnostic_report", ], + "product_catalog": [ + "esp_product_search", + "esp_product_info", + "esp_chip_compare", + "esp_product_recommend", + "esp_product_availability", + ], } if self.config.get_qemu_available(): diff --git a/tests/test_product_catalog.py b/tests/test_product_catalog.py new file mode 100644 index 0000000..a028d74 --- /dev/null +++ b/tests/test_product_catalog.py @@ -0,0 +1,514 @@ +""" +Tests for the Product Catalog component. + +Uses mocked HTTP responses to avoid hitting the real Espressif API. +""" + +import time +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from fastmcp import FastMCP + +from mcesptool.components.product_catalog import ( + ProductCatalog, + _has_capability, + _normalize_status, + _parse_lead_weeks, + _parse_memory_field, + _parse_sram_kb, + _parse_temp_max, +) +from mcesptool.config import ESPToolServerConfig + +# ── Fixtures ──────────────────────────────────────────────────────────── + +SAMPLE_PRODUCTS = [ + { + "id": 1, + "name": "ESP32-S3", + "nameNew": "ESP32-S3", + "type": "SoC", + "seriesName": "ESP32-S3", + "wifi": "802.11 b/g/n", + "wifi6": "", + "bluetooth": "BLE 5.0", + "threadZigbee": "", + "flash": "0", + "psram": "0", + "sram": "512", + "rom": "384", + "gpio": "45", + "pins": "56", + "freq": "240", + "operatingTemp": "-40~85C", + "voltageRange": "3.0~3.6V", + "dimensions": "7x7 mm", + "status": "Mass Production", + "antenna": "", + "mpn": "ESP32-S3", + "sizeType": "QFN56", + "releaseTime": "2020", + "idfSupports": "v4.4+", + "spq": "490", + "moq": "4900", + "leadTime": "8-10 weeks", + "eccnCode": "5A992.c", + "ccatszCode": "", + "hsCode": "8542.31", + "preFirmware": "", + "replacedByName": "", + }, + { + "id": 2, + "name": "ESP32-S3-WROOM-1-N16R8", + "nameNew": "ESP32-S3-WROOM-1-N16R8", + "type": "Module", + "seriesName": "ESP32-S3", + "wifi": "802.11 b/g/n", + "wifi6": "", + "bluetooth": "BLE 5.0", + "threadZigbee": "", + "flash": "16, Quad", + "psram": "8, Octal", + "sram": "512", + "rom": "384", + "gpio": "36", + "pins": "44", + "freq": "240", + "operatingTemp": "-40~85C", + "voltageRange": "3.0~3.6V", + "dimensions": "18x25.5x3.1 mm", + "status": "Mass Production", + "antenna": "PCB", + "mpn": "ESP32-S3-WROOM-1-N16R8", + "sizeType": "", + "releaseTime": "2021", + "idfSupports": "v4.4+", + "spq": "400", + "moq": "2000", + "leadTime": "6-8 weeks", + "eccnCode": "5A992.c", + "ccatszCode": "", + "hsCode": "8542.31", + "preFirmware": "", + "replacedByName": "", + }, + { + "id": 3, + "name": "ESP32-C6-MINI-1-N4", + "nameNew": "ESP32-C6-MINI-1-N4", + "type": "Module", + "seriesName": "ESP32-C6", + "wifi": "802.11 b/g/n/ax", + "wifi6": "Yes", + "bluetooth": "BLE 5.3", + "threadZigbee": "Available", + "flash": "4, Quad", + "psram": "", + "sram": "512", + "rom": "320", + "gpio": "22", + "pins": "53", + "freq": "160", + "operatingTemp": "-40~105C", + "voltageRange": "3.0~3.6V", + "dimensions": "13.2x16.6x2.4 mm", + "status": "Mass Production", + "antenna": "PCB", + "mpn": "ESP32-C6-MINI-1-N4", + "sizeType": "", + "releaseTime": "2023", + "idfSupports": "v5.1+", + "spq": "500", + "moq": "2500", + "leadTime": "4-6 weeks", + "eccnCode": "5A992.c", + "ccatszCode": "", + "hsCode": "8542.31", + "preFirmware": "", + "replacedByName": "", + }, + { + "id": 4, + "name": "ESP32-WROOM-32E", + "nameNew": "ESP32-WROOM-32E", + "type": "Module", + "seriesName": "ESP32", + "wifi": "802.11 b/g/n", + "wifi6": "", + "bluetooth": "BT/BLE 4.2", + "threadZigbee": "N/A", + "flash": "4, Quad", + "psram": "", + "sram": "520", + "rom": "448", + "gpio": "34", + "pins": "38", + "freq": "240", + "operatingTemp": "-40~85C", + "voltageRange": "3.0~3.6V", + "dimensions": "18x20x3.2 mm", + "status": "NRND", + "antenna": "PCB", + "mpn": "ESP32-WROOM-32E", + "sizeType": "", + "releaseTime": "2019", + "idfSupports": "v4.0+", + "spq": "500", + "moq": "5000", + "leadTime": "12-14 weeks", + "eccnCode": "5A992.c", + "ccatszCode": "", + "hsCode": "8542.31", + "preFirmware": "", + "replacedByName": "", + }, +] + + +def _mock_response() -> httpx.Response: + """Build a mock httpx response with sample product data.""" + return httpx.Response( + status_code=200, + json={"options": {}, "results": SAMPLE_PRODUCTS}, + request=httpx.Request("GET", "https://products.espressif.com/api/user/products"), + ) + + +@pytest.fixture +def app(): + return FastMCP("test-esp") + + +@pytest.fixture +def config(): + with patch.object(ESPToolServerConfig, "_check_tool_availability", return_value=True): + return ESPToolServerConfig.from_environment() + + +@pytest.fixture +def catalog(app, config): + cat = ProductCatalog(app, config) + # Pre-load the cache so tests don't hit the network + cat._products = SAMPLE_PRODUCTS + cat._last_fetch = time.time() + return cat + + +@pytest.fixture +def context(): + """Minimal mock MCP context.""" + return AsyncMock(spec=[]) + + +# ── Helper function unit tests ────────────────────────────────────────── + + +class TestParsers: + def test_parse_memory_basic(self): + assert _parse_memory_field("16, Quad") == (16, "Quad") + assert _parse_memory_field("8, Octal") == (8, "Octal") + assert _parse_memory_field("4") == (4, "") + + def test_parse_memory_bare_int(self): + assert _parse_memory_field(8) == (8, "") + assert _parse_memory_field(0) == (0, "") + + def test_parse_memory_empty(self): + assert _parse_memory_field("") == (0, "") + assert _parse_memory_field("N/A") == (0, "") + assert _parse_memory_field("NA") == (0, "") + assert _parse_memory_field(None) == (0, "") + + def test_parse_sram(self): + assert _parse_sram_kb("512") == 512 + assert _parse_sram_kb("520") == 520 + assert _parse_sram_kb("") == 0 + assert _parse_sram_kb(512) == 512 + assert _parse_sram_kb(None) == 0 + + def test_has_capability(self): + assert _has_capability("802.11 b/g/n") is True + assert _has_capability("BLE 5.0") is True + assert _has_capability("Available") is True + assert _has_capability("") is False + assert _has_capability("N/A") is False + assert _has_capability("NA") is False + + def test_normalize_status(self): + assert _normalize_status("Mass Production") == "Mass Production" + assert _normalize_status("mass production") == "Mass Production" + assert _normalize_status("NRND") == "NRND" + assert _normalize_status("nrnd") == "NRND" + + def test_parse_lead_weeks(self): + assert _parse_lead_weeks("8-10 weeks") == 8 + assert _parse_lead_weeks("4-6 weeks") == 4 + assert _parse_lead_weeks("") is None + assert _parse_lead_weeks("TBD") is None + + def test_parse_temp_max(self): + assert _parse_temp_max("-40~85C") == 85 + assert _parse_temp_max("-40~105C") == 105 + assert _parse_temp_max("") is None + + +# ── Tool integration tests ────────────────────────────────────────────── + + +@pytest.mark.asyncio +class TestProductSearch: + async def test_search_no_filters(self, catalog, context): + result = await catalog._product_search_impl(context) + assert result["count"] == 4 + assert len(result["products"]) == 4 + + async def test_search_by_series(self, catalog, context): + result = await catalog._product_search_impl(context, series="ESP32-S3") + assert result["count"] == 2 + names = [p["name"] for p in result["products"]] + assert "ESP32-S3" in names + assert "ESP32-S3-WROOM-1-N16R8" in names + + async def test_search_by_type(self, catalog, context): + result = await catalog._product_search_impl(context, product_type="SoC") + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-S3" + + async def test_search_by_wifi(self, catalog, context): + result = await catalog._product_search_impl(context, wifi=True) + assert result["count"] == 4 # all sample products have wifi + + async def test_search_by_thread_zigbee(self, catalog, context): + result = await catalog._product_search_impl(context, thread_zigbee=True) + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4" + + async def test_search_by_min_flash(self, catalog, context): + result = await catalog._product_search_impl(context, min_flash_mb=8) + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8" + + async def test_search_by_min_psram(self, catalog, context): + result = await catalog._product_search_impl(context, min_psram_mb=4) + assert result["count"] == 1 + assert result["products"][0]["flash_mb"] == 16 + + async def test_search_by_status(self, catalog, context): + result = await catalog._product_search_impl(context, status="NRND") + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-WROOM-32E" + + async def test_search_by_antenna(self, catalog, context): + result = await catalog._product_search_impl(context, antenna="PCB") + assert result["count"] == 3 + + async def test_search_by_max_temp(self, catalog, context): + result = await catalog._product_search_impl(context, max_temp_c=105) + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-C6-MINI-1-N4" + + async def test_search_keyword(self, catalog, context): + result = await catalog._product_search_impl(context, keyword="WROOM") + assert result["count"] >= 1 + names = [p["name"] for p in result["products"]] + assert any("WROOM" in n for n in names) + + async def test_search_combined_filters(self, catalog, context): + result = await catalog._product_search_impl( + context, series="ESP32-S3", product_type="Module", min_flash_mb=8 + ) + assert result["count"] == 1 + assert result["products"][0]["name"] == "ESP32-S3-WROOM-1-N16R8" + + async def test_search_no_results(self, catalog, context): + result = await catalog._product_search_impl(context, min_flash_mb=128) + assert result["count"] == 0 + assert result["products"] == [] + + +@pytest.mark.asyncio +class TestProductInfo: + async def test_info_exact_name(self, catalog, context): + result = await catalog._product_info_impl(context, "ESP32-S3-WROOM-1-N16R8") + assert result["name"] == "ESP32-S3-WROOM-1-N16R8" + assert result["flash_mb"] == 16 + assert result["psram_mb"] == 8 + assert result["series"] == "ESP32-S3" + + async def test_info_fuzzy_name(self, catalog, context): + result = await catalog._product_info_impl(context, "s3 wroom n16r8") + assert result["name"] == "ESP32-S3-WROOM-1-N16R8" + + async def test_info_not_found(self, catalog, context): + result = await catalog._product_info_impl(context, "ESP99-NONEXISTENT") + assert "error" in result + + +@pytest.mark.asyncio +class TestChipCompare: + async def test_compare_two(self, catalog, context): + result = await catalog._chip_compare_impl( + context, ["ESP32-S3-WROOM-1-N16R8", "ESP32-C6-MINI-1-N4"] + ) + assert len(result["products_compared"]) == 2 + assert "differences" in result + assert "common" in result + # These two differ in series, flash, etc. + assert "series" in result["differences"] + + async def test_compare_too_few(self, catalog, context): + result = await catalog._chip_compare_impl(context, ["ESP32-S3"]) + assert "error" in result + + async def test_compare_too_many(self, catalog, context): + result = await catalog._chip_compare_impl( + context, ["a", "b", "c", "d", "e"] + ) + assert "error" in result + + async def test_compare_not_found(self, catalog, context): + result = await catalog._chip_compare_impl( + context, ["ESP32-S3", "NONEXISTENT-CHIP"] + ) + assert "error" in result + + +@pytest.mark.asyncio +class TestProductRecommend: + async def test_recommend_ble_sensor(self, catalog, context): + result = await catalog._product_recommend_impl( + context, "battery-powered BLE sensor", None, True + ) + assert result["count"] >= 1 + # All recommendations should be modules (prefer_module=True) and have BLE + for p in result["recommendations"]: + assert p["type"] == "Module" + assert p["bluetooth"] is True + + async def test_recommend_thread(self, catalog, context): + result = await catalog._product_recommend_impl( + context, "Thread/Matter smart home device", None, True + ) + assert result["count"] >= 1 + for p in result["recommendations"]: + assert p["thread_zigbee"] is True + + async def test_recommend_with_psram(self, catalog, context): + result = await catalog._product_recommend_impl( + context, "wifi camera needs psram", None, True + ) + for p in result["recommendations"]: + assert p["psram_mb"] > 0 + + async def test_recommend_industrial(self, catalog, context): + result = await catalog._product_recommend_impl( + context, "industrial IoT gateway", "industrial temp range", True + ) + # Only ESP32-C6-MINI-1-N4 has 105C rating in our sample data + assert result["count"] >= 1 + + async def test_recommend_fallback_to_soc(self, catalog, context): + """When no modules match, falls back to including SoCs.""" + result = await catalog._product_recommend_impl( + context, "wifi device", "needs 32MB flash", True + ) + # No modules have 32MB flash — falls back to SoC search too + assert isinstance(result["count"], int) + + +@pytest.mark.asyncio +class TestProductAvailability: + async def test_availability_mass_production(self, catalog, context): + result = await catalog._product_availability_impl( + context, None, "Mass Production", None, None + ) + assert result["count"] == 3 # NRND one excluded + + async def test_availability_by_series(self, catalog, context): + result = await catalog._product_availability_impl( + context, "ESP32-S3", "Mass Production", None, None + ) + assert result["count"] == 2 + + async def test_availability_lead_time(self, catalog, context): + result = await catalog._product_availability_impl( + context, None, "Mass Production", 6, None + ) + # ESP32-C6-MINI-1-N4 (4 weeks) and ESP32-S3-WROOM-1-N16R8 (6 weeks) both pass <= 6 + assert result["count"] == 2 + names = [p["name"] for p in result["products"]] + assert "ESP32-C6-MINI-1-N4" in names + assert "ESP32-S3-WROOM-1-N16R8" in names + + async def test_availability_max_moq(self, catalog, context): + result = await catalog._product_availability_impl( + context, None, "Mass Production", None, 2500 + ) + # ESP32-S3-WROOM-1-N16R8 (2000) and ESP32-C6-MINI-1-N4 (2500) + assert result["count"] == 2 + + +# ── Cache behavior tests ──────────────────────────────────────────────── + + +@pytest.mark.asyncio +class TestCaching: + async def test_cache_hit(self, catalog): + """Cache should be used when data is fresh.""" + original_fetch_time = catalog._last_fetch + products = await catalog._ensure_data() + assert len(products) == 4 + assert catalog._last_fetch == original_fetch_time # no new fetch + + async def test_cache_expired(self, catalog): + """Cache should refresh when TTL expired.""" + catalog._last_fetch = time.time() - 90000 # older than 24h + + mock_resp = _mock_response() + with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + products = await catalog._ensure_data() + assert len(products) == 4 + mock_client.get.assert_called_once() + + async def test_force_refresh(self, catalog): + """Force refresh bypasses cache.""" + mock_resp = _mock_response() + with patch("mcesptool.components.product_catalog.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_resp + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + products = await catalog._ensure_data(force_refresh=True) + assert len(products) == 4 + mock_client.get.assert_called_once() + + +# ── Fuzzy matching tests ──────────────────────────────────────────────── + + +class TestFuzzyMatch: + def test_exact_match(self, catalog): + results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "ESP32-S3") + assert len(results) >= 1 + assert results[0]["name"] == "ESP32-S3" + + def test_substring_match(self, catalog): + results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "WROOM") + assert len(results) >= 1 + names = [p["name"] for p in results] + assert any("WROOM" in n for n in names) + + def test_no_match(self, catalog): + results = catalog._fuzzy_find(SAMPLE_PRODUCTS, "XYZNONEXISTENT", threshold=90) + assert len(results) == 0 diff --git a/uv.lock b/uv.lock index 36dc5a0..40eb50b 100644 --- a/uv.lock +++ b/uv.lock @@ -893,6 +893,7 @@ source = { editable = "." } dependencies = [ { name = "click" }, { name = "fastmcp" }, + { name = "httpx" }, { name = "pydantic" }, { name = "pyserial" }, { name = "pyserial-asyncio" }, @@ -933,6 +934,7 @@ requires-dist = [ { name = "factory-boy", marker = "extra == 'testing'", specifier = ">=3.3.0" }, { name = "fastmcp", specifier = ">=3.0.2,<4" }, { name = "gunicorn", marker = "extra == 'production'", specifier = ">=21.0.0" }, + { name = "httpx", specifier = ">=0.28" }, { name = "kconfiglib", marker = "extra == 'idf'", specifier = ">=14.1.0" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.5.0" }, { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=3.0.0" },