"""Manufacturer spec pages → claims on hardware entities (tier 1). One class per manufacturer domain (source provenance), module `spec_pages`. * apple_specs Mac Studio / Mac mini / MacBook Pro / MacBook Air / iMac tech-spec pages. The "Chip" and "Memory" rows are parsed per model column: chip name + "NNNGB/s memory bandwidth" (+ configurable chip variants with their own bandwidth), memory sizes (+ the chip variant a size requires, in parentheses). Chips get `memory_bandwidth_gbs` (the highest stated for that chip) and `memory_bandwidth_options_gbs`; memory *configurations* are claimed on a product entity ("Mac mini (Apple M6)", kind computer) that `uses` the chip — the same chip ships with different memory options per product, so chip-level `memory_gb` from five pages would flip-flop. Only what the page states; the iMac page (older template) is parsed from text. * nvidia_specs H100 / H200 (two-column spec tables SXM | NVL), DGX B200 and DGX Spark (key/value tables): memory, bandwidth, TDP, Tensor Core TFLOPS (NVIDIA publishes them with sparsity → `compute_*_tflops` + `compute_sparsity: true`). Entities are addressed with `{"registry_hardware": key}` when the key exists in registry/hardware.yaml (+ hardware.d fragments); a page value that differs from the curated registry is left to the writer's conflict handling (never edited silently). """ from __future__ import annotations import re from typing import Any from aiatlas.registry import load, org_ref from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext from aiatlas.sdk.facts import EntityRef, Facts, Target from aiatlas.sdk.fetch import FetchResult APPLE_PAGES = {"mac-studio": "Mac Studio", "mac-mini": "Mac mini", "macbook-pro": "MacBook Pro", "macbook-air": "MacBook Air", "imac": "iMac"} CHIP = re.compile(r"\bApple\s*(M\d+(?:\s*(?:Pro|Max|Ultra))?)\s*chip", re.IGNORECASE) CHIP_SHORT = re.compile(r"\b(M\d+(?:\s*(?:Pro|Max|Ultra))?)\b") BANDWIDTH = re.compile(r"(\d+(?:\.\d+)?)\s*(GB|TB)/s memory bandwidth", re.IGNORECASE) GB = re.compile(r"(\d+)\s*GB\b") PAREN = re.compile(r"\(([^)]*)\)") def hardware_ref(key: str, fallback_name: str, *, manufacturer: str, kind: str, scheme: str = "spec_key") -> EntityRef: entry = next((h for h in load("hardware") if h["key"] == key), None) org = org_ref(manufacturer) if entry: return EntityRef(entity_type="hardware", name=entry["name"], identifiers={"registry_hardware": key}, aliases=list(entry.get("aliases", [])), slug_hint=key, organization=org) return EntityRef(entity_type="hardware", name=fallback_name, identifiers={scheme: key}, slug_hint=key, organization=org, attributes={"kind": kind}) def _norm_chip(s: str) -> str: s = re.sub(r"\s+", " ", s.replace("\xa0", " ")).strip() m = CHIP_SHORT.search(s) return f"Apple {m.group(1)}".replace(" ", " ") if m else s def _bw_gbs(num: str, unit: str) -> float: v = float(num) return round(v * 1000 if unit.upper() == "TB" else v, 1) class _Specs(BaseConnector): version = "1" parser_version = "1" interval_seconds = 86400 min_interval_seconds = 12 * 3600 max_interval_seconds = 14 * 86400 tier = 1 priority = 2 concurrency = 1 # ================================================================================================ Apple class AppleSpecsConnector(_Specs): name = "apple_specs" label = "Apple — Mac tech specs (chips, memory bandwidth, memory configurations)" description = "apple.com//specs/ pages: Apple Silicon chip per model column, unified memory options and memory bandwidth." source_key = "apple.com" rate_per_min = 10 expected_min_records = 5 async def discover(self, ctx: RunContext) -> list[Target]: return [Target(url=f"https://www.apple.com/{slug}/specs/", doc_type="spec_page", key=f"apple:{slug}", min_bytes=20000, priority=1, meta={"product": product}) for slug, product in APPLE_PAGES.items()] async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: facts = Facts() html = parsed.html if not html: return facts product = target.meta.get("product") or (html.title or "").split(" - ")[0].strip() apple = org_ref("apple") facts.entities.append(apple) chips: dict[str, set[float]] = {} # chip name → bandwidths stated products: dict[str, dict[str, Any]] = {} # chip name → {"sizes": set, "bw": {size: bw}} rows = html.css(".techspecs-row") if rows: self._parse_rows(rows, chips, products) else: self._parse_text(html.text, chips, products) if not chips: ctx.log.warning("apple specs: no chip rows found", extra={"url": target.url}) for chip, bws in chips.items(): ref = self._chip_ref(facts, chip) if bws: facts.claim(ref, "memory_bandwidth_gbs", max(bws), unit="GB/s") if len(bws) > 1: facts.claim(ref, "memory_bandwidth_options_gbs", sorted(bws), unit="GB/s") # no chip-level `spec_url`: the same chip is described on several product pages (the registry keeps its curated one) facts.relate(apple, "manufactures", ref) for chip, info in products.items(): if not info["sizes"]: continue chip_ref = self._chip_ref(facts, chip) key = f"{target.key.split(':', 1)[1] if target.key else product.lower()}:{_key(chip)}" pref = facts.entity("hardware", f"{product} ({chip})", identifiers={"apple_product_chip": key}, organization=apple, aliases=[f"{product} {chip}"]) facts.claim(pref, "kind", "computer") facts.claim(pref, "product_line", product) facts.claim(pref, "memory_gb", sorted(info["sizes"]), unit="GB") facts.claim(pref, "memory_type", "unified memory") bws = set(chips.get(chip) or set()) | set(info["bw"].values()) if bws: facts.claim(pref, "memory_bandwidth_gbs", max(bws), unit="GB/s") if info["bw"]: facts.claim(pref, "memory_configurations", [{"memory_gb": s, "memory_bandwidth_gbs": b} for s, b in sorted(info["bw"].items())]) facts.claim(pref, "spec_url", target.url) facts.relate(pref, "uses", chip_ref) facts.relate(apple, "manufactures", pref) facts.document_entity = apple facts.document_title = html.title return facts def _chip_ref(self, facts: Facts, chip: str) -> EntityRef: key = _key(chip) for e in facts.entities: if e.entity_type == "hardware" and (e.identifiers.get("registry_hardware") == key or e.identifiers.get("apple_chip") == key): return e ref = hardware_ref(key, chip, manufacturer="apple", kind="soc", scheme="apple_chip") facts.entities.append(ref) return ref def _parse_rows(self, rows: list[Any], chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None: column_chips: list[str | None] = [] for row in rows: head_node = row.css_first(".techspecs-rowheader") head = re.sub(r"\d+$", "", head_node.text(strip=True)).strip().lower() if head_node else "" cols = row.css(".techspecs-column") if head == "chip": column_chips = [] for col in cols: lines = _column_lines(col) chip = next((f"Apple {m.group(1)}" for t in lines for m in [CHIP.search(t)] if m), None) chip = _norm_chip(chip) if chip else None column_chips.append(chip) if not chip: continue chips.setdefault(chip, set()) products.setdefault(chip, {"sizes": set(), "bw": {}}) for t in lines: m = BANDWIDTH.search(t) if not m: continue bw = _bw_gbs(m.group(1), m.group(2)) cm = CHIP_SHORT.search(re.sub(r"^(?:or|and)\s+", "", t.strip())) target_chip = _norm_chip(cm.group(1)) if cm and " with " in f" {t} " else chip chips.setdefault(target_chip, set()).add(bw) products.setdefault(target_chip, {"sizes": set(), "bw": {}}) elif head == "memory" and column_chips: for i, col in enumerate(cols): chip = column_chips[i] if i < len(column_chips) else None if not chip: continue products.setdefault(chip, {"sizes": set(), "bw": {}}) for t in _column_lines(col): self._memory_item(t, chip, chips, products) @staticmethod def _memory_item(text: str, column_chip: str, chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None: """'24GB or 32GB (170GB/s memory bandwidth)' · '36GB (M5 Max with 32-core GPU)' · 'or 48GB (M5 Pro or M5 Max with 40-core GPU)'.""" outside = PAREN.sub("", text) sizes = [int(g) for g in GB.findall(outside)] if not sizes: return parens = PAREN.findall(text) bw_m = next((BANDWIDTH.search(p) for p in parens if BANDWIDTH.search(p)), None) chip_targets = {_norm_chip(c) for p in parens for c in CHIP_SHORT.findall(p)} chip_targets = {c for c in chip_targets if re.fullmatch(r"Apple M\d+( Pro| Max| Ultra)?", c)} or {column_chip} for chip in chip_targets: info = products.setdefault(chip, {"sizes": set(), "bw": {}}) info["sizes"].update(sizes) if bw_m: bw = _bw_gbs(bw_m.group(1), bw_m.group(2)) chips.setdefault(chip, set()).add(bw) for s in sizes: info["bw"][s] = bw def _parse_text(self, text: str, chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None: """Older template (iMac): sequential text. Safe only when the page describes a single chip.""" found = {f"Apple {m.group(1)}" for m in CHIP.finditer(text)} if len(found) != 1: for chip in found: chips.setdefault(chip, set()) return chip = next(iter(found)) bws = {_bw_gbs(m.group(1), m.group(2)) for m in BANDWIDTH.finditer(text)} chips[chip] = bws info = products.setdefault(chip, {"sizes": set(), "bw": {}}) mem = re.search(r"\nMemory\n(.*?)\n(?:Display|Storage)\n", text, flags=re.DOTALL) if mem: info["sizes"].update(int(g) for g in re.findall(r"(\d+)GB(?: or \d+GB)? unified memory", mem.group(1))) info["sizes"].update(int(g) for g in re.findall(r"(\d+)GB or \d+GB unified memory", mem.group(1))) info["sizes"].update(int(g) for g in re.findall(r"\d+GB or (\d+)GB unified memory", mem.group(1))) def _column_lines(col: Any) -> list[str]: """Text lines of one spec column (both Apple templates: /
  • and

    ).""" out: list[str] = [] for node in col.css("li, p, strong.title"): if node.tag == "p" and node.css_first("strong.title"): continue # the itself is collected t = re.sub(r"\s+", " ", node.text(separator=" ", strip=True).replace("\xa0", " ")).strip() if t and (not out or out[-1] != t): out.append(t) return out def _key(chip: str) -> str: return re.sub(r"[^a-z0-9]+", "-", chip.lower()).strip("-") # ================================================================================================ NVIDIA NV_PAGES = [ ("https://www.nvidia.com/en-us/data-center/h100/", "nvidia:h100", {"H100 SXM": "nvidia-h100-sxm", "H100 NVL": "nvidia-h100-nvl"}), ("https://www.nvidia.com/en-us/data-center/h200/", "nvidia:h200", {"H200 SXM": "nvidia-h200", "H200 NVL": "nvidia-h200-nvl"}), ("https://www.nvidia.com/en-us/data-center/dgx-b200/", "nvidia:dgx-b200", {"__single__": "nvidia-dgx-b200"}), ("https://www.nvidia.com/en-us/products/workstations/dgx-spark/", "nvidia:dgx-spark", {"__single__": "nvidia-dgx-spark"}), ] NUM = re.compile(r"(\d{1,3}(?:,\d{3})+|\d+(?:\.\d+)?)") TFLOPS_ROWS = {"fp64": "compute_fp64_tflops", "fp64 tensor core": "compute_fp64_tensor_tflops", "fp32": "compute_fp32_tflops", "tf32 tensor core": "compute_tf32_tflops", "bfloat16 tensor core": "compute_bf16_tflops", "fp16 tensor core": "compute_fp16_tflops", "fp8 tensor core": "compute_fp8_tflops", "int8 tensor core": "compute_int8_tops", "fp4 tensor core": "compute_fp4_tflops"} class NvidiaSpecsConnector(_Specs): name = "nvidia_specs" label = "NVIDIA — data-center GPU and DGX system specifications" description = "H100, H200, DGX B200 and DGX Spark spec tables: memory, bandwidth, TDP, Tensor Core throughput, form factor." source_key = "nvidia.com" rate_per_min = 10 expected_min_records = 4 async def discover(self, ctx: RunContext) -> list[Target]: return [Target(url=url, doc_type="spec_page", key=key, min_bytes=20000, priority=1, meta={"columns": cols}) for url, key, cols in NV_PAGES] async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: facts = Facts() html = parsed.html if not html: return facts nvidia = org_ref("nvidia") facts.entities.append(nvidia) columns: dict[str, str] = target.meta.get("columns") or {} for table in html.tables: rows = [r for r in table["rows"] if r] headers = table["headers"] if not headers and rows and rows[0] and rows[0][0] == "" and len(rows[0]) >= 3: headers, rows = rows[0], rows[1:] if headers and len(headers) >= 3 and headers[0] == "": names = [re.sub(r"[¹²³⁴*]+", "", h).strip() for h in headers[1:]] keys = [columns.get(n) for n in names] if not any(keys): continue for i, key in enumerate(keys): if not key: continue ref = self._ref(facts, key, names[i]) for row in rows: if len(row) > i + 1: self._spec(facts, ref, row[0], row[i + 1]) self._finish(facts, nvidia, ref, target.url) elif "__single__" in columns and rows and all(len(r) >= 2 for r in rows[:5]): ref = self._ref(facts, columns["__single__"], columns["__single__"]) for row in rows: if len(row) >= 2: self._spec(facts, ref, row[0], row[1]) self._finish(facts, nvidia, ref, target.url) break facts.document_entity = nvidia facts.document_title = html.title return facts def _ref(self, facts: Facts, key: str, name: str) -> EntityRef: for e in facts.entities: if e.identifiers.get("registry_hardware") == key or e.identifiers.get("nvidia_product") == key: return e ref = hardware_ref(key, f"NVIDIA {name}", manufacturer="nvidia", kind="gpu", scheme="nvidia_product") facts.entities.append(ref) return ref def _finish(self, facts: Facts, nvidia: EntityRef, ref: EntityRef, url: str) -> None: facts.claim(ref, "spec_url", url) facts.relate(nvidia, "manufactures", ref) if ref.identifiers.get("registry_hardware") == "nvidia-dgx-b200": b200 = hardware_ref("nvidia-b200", "NVIDIA B200", manufacturer="nvidia", kind="gpu") facts.entities.append(b200) facts.relate(ref, "uses", b200, attributes={"count": 8}) def _spec(self, facts: Facts, ref: EntityRef, label: str, value: str) -> None: key = re.sub(r"[¹²³⁴*]+", "", label).strip().lower() key = re.sub(r"\s+\d$", "", key) val = value.strip() if not val or val in ("-", "—"): return if key in ("gpu memory", "system memory") and "GB" in val: m = re.search(r"(\d{1,3}(?:,\d{3})*|\d+)\s*GB", val) if m: facts.claim(ref, "memory_gb", int(m.group(1).replace(",", "")), unit="GB") t = re.search(r"\b(HBM\d[eE]?|LPDDR\d[xX]?|GDDR\d[xX]?)\b", val) if t: facts.claim(ref, "memory_type", t.group(1)) bw = re.search(r"(\d+(?:\.\d+)?)\s*(TB|GB)/s", val) if bw: facts.claim(ref, "memory_bandwidth_gbs", _bw_gbs(bw.group(1), bw.group(2)), unit="GB/s") elif key in ("gpu memory bandwidth", "memory bandwidth"): bw = re.search(r"(\d+(?:\.\d+)?)\s*(TB|GB)/s", val) if bw: facts.claim(ref, "memory_bandwidth_gbs", _bw_gbs(bw.group(1), bw.group(2)), unit="GB/s") elif "thermal design power" in key or key in ("tdp", "gb10 tdp"): watts = [int(w) for w in re.findall(r"(\d{2,4})\s*W", val.replace(",", ""))] if watts: facts.claim(ref, "tdp_watts", max(watts), unit="W") if len(watts) > 1 or "up to" in val.lower() or "configurable" in val.lower(): facts.claim(ref, "tdp_note", val[:120]) elif key in ("system power usage", "power supply"): m = re.search(r"(\d+(?:\.\d+)?)\s*(kW|W)", val.replace(",", "")) if m: watts = float(m.group(1)) * (1000 if m.group(2) == "kW" else 1) facts.claim(ref, "system_power_watts" if "usage" in key else "power_supply_watts", int(watts), unit="W") elif key in TFLOPS_ROWS: m = NUM.search(val) if m: num = float(m.group(1).replace(",", "")) if re.search(r"\bP(?:eta)?FLOPS", val, re.IGNORECASE): num *= 1000 facts.claim(ref, TFLOPS_ROWS[key], int(num) if num.is_integer() else num, unit="TFLOPS" if "tops" not in TFLOPS_ROWS[key] else "TOPS") if "*" in label or "²" in label or "sparsity" in val.lower(): facts.claim(ref, "compute_sparsity", True) elif key == "performance": for prec, num, unit in re.findall(r"(FP\d)\s*Tensor Core:\s*(\d+(?:\.\d+)?)\s*(PFLOPS|TFLOPS)", val): tflops = float(num) * (1000 if unit == "PFLOPS" else 1) facts.claim(ref, f"compute_{prec.lower()}_tflops", int(tflops), unit="TFLOPS") facts.claim(ref, "compute_sparsity", True) elif key.startswith("tensor performance"): m = re.search(r"(\d+(?:\.\d+)?)\s*(PFLOP|TFLOP)S?\s*(FP\d)", val, re.IGNORECASE) if m: tflops = float(m.group(1)) * (1000 if m.group(2).upper() == "PFLOP" else 1) facts.claim(ref, f"compute_{m.group(3).lower()}_tflops", int(tflops), unit="TFLOPS") elif key == "form factor": facts.claim(ref, "form_factor", val[:80]) elif key == "interconnect": m = re.search(r"NVLink[^\d]*(\d+)\s*GB/s", val) if m: facts.claim(ref, "nvlink_bandwidth_gbs", int(m.group(1)), unit="GB/s") elif key == "gpu": m = re.match(r"(\d+)x\s+(.*)", val) if m: facts.claim(ref, "gpu_count", int(m.group(1))) facts.claim(ref, "gpu_model", m.group(2)[:80]) elif "architecture" in val.lower(): facts.claim(ref, "gpu_architecture", val[:80]) elif key == "architecture": facts.claim(ref, "architecture", val[:80]) elif key == "cpu": facts.claim(ref, "cpu", val[:160]) elif key in ("storage",): facts.claim(ref, "storage", val[:160]) elif key == "multi-instance gpus": facts.claim(ref, "mig_support", val[:80]) elif key == "nvidia nvlink bandwidth": m = re.search(r"(\d+(?:\.\d+)?)\s*TB/s", val) if m: facts.claim(ref, "nvlink_bandwidth_gbs", int(float(m.group(1)) * 1000), unit="GB/s") CONNECTORS = [AppleSpecsConnector, NvidiaSpecsConnector]