| """ |
| Wire symbolic recursion system to compute resources and engines. |
| Writes config/resources.json with live connection status. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| import json |
| import os |
| import subprocess |
| import sys |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| import httpx |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CONFIG_PATH = ROOT / "config" / "resources.json" |
|
|
|
|
| async def probe_url(url: str, path: str = "/health") -> dict: |
| try: |
| async with httpx.AsyncClient(timeout=5.0) as c: |
| r = await c.get(f"{url}{path}") |
| return {"url": url, "reachable": r.status_code == 200, "response": r.json() if r.status_code == 200 else None} |
| except Exception as e: |
| return {"url": url, "reachable": False, "error": str(e)} |
|
|
|
|
| def _gcloud_bin() -> str: |
| candidates = [ |
| "gcloud", |
| r"C:\Users\stlta\google-cloud-sdk\bin\gcloud.cmd", |
| ] |
| for c in candidates: |
| if c == "gcloud" or Path(c).exists(): |
| return c |
| return "gcloud" |
|
|
|
|
| def probe_gcloud() -> dict: |
| gcloud = _gcloud_bin() |
| try: |
| r = subprocess.run( |
| f'"{gcloud}" config get-value project', |
| capture_output=True, text=True, timeout=10, shell=True, |
| ) |
| project = r.stdout.strip() if r.returncode == 0 else None |
| account_r = subprocess.run( |
| f'"{gcloud}" config get-value account', |
| capture_output=True, text=True, timeout=10, shell=True, |
| ) |
| account = account_r.stdout.strip() if account_r.returncode == 0 else None |
| return {"available": bool(project), "project": project, "account": account} |
| except Exception as e: |
| return {"available": False, "error": str(e)} |
|
|
|
|
| def probe_engines() -> dict: |
| sys.path.insert(0, str(ROOT)) |
| sys.path.insert(0, str(ROOT.parent / "primal-trading")) |
| primallang_root = ( |
| ROOT.parent / "primal-lang-evolution" / "sources" / "local-downloads" |
| / "extracted" / "primallang_v3" / "primallang_v3" |
| ) |
| if primallang_root.exists(): |
| sys.path.insert(0, str(primallang_root)) |
|
|
| engines = {} |
| try: |
| from primal_trading.primal_kernel import EWICIntegrator |
| engines["primal_trading"] = {"path": str(ROOT.parent / "primal-trading"), "loaded": True} |
| except ImportError as e: |
| engines["primal_trading"] = {"loaded": False, "error": str(e)} |
|
|
| try: |
| from primallang.physics import QuantumState, plate_eigenfrequency, PlateConfig |
| engines["primallang_physics"] = { |
| "path": str(primallang_root), |
| "loaded": True, |
| "sample_eigenfreq_hz": plate_eigenfrequency(1, 1, PlateConfig()), |
| } |
| except ImportError as e: |
| engines["primallang_physics"] = {"loaded": False, "error": str(e)} |
|
|
| return engines |
|
|
|
|
| async def main() -> None: |
| os.environ.setdefault("GATEWAY_URL", "http://localhost:3000") |
| os.environ.setdefault("GATEWAY_TOKEN", "pilot-console-primary-jwt-fallback-x992z") |
|
|
| resources = { |
| "wired_at": datetime.now(timezone.utc).isoformat(), |
| "symbolic_recursion_api": await probe_url("http://localhost:8080"), |
| "physics_gateway": await probe_url("http://localhost:3000", "/v1/health"), |
| "engines": probe_engines(), |
| "atlas": { |
| "source": str(ROOT / "atlas-src" / "main.tex"), |
| "available": (ROOT / "atlas-src" / "main.tex").exists(), |
| }, |
| "gcloud": probe_gcloud(), |
| "public_feeds": { |
| "open_meteo": "https://api.open-meteo.com/v1/forecast", |
| "noaa_kp": "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", |
| }, |
| "env": { |
| "GATEWAY_URL": os.environ.get("GATEWAY_URL"), |
| "GATEWAY_TOKEN": "***" if os.environ.get("GATEWAY_TOKEN") else None, |
| "SENSOR_LAT": os.environ.get("SENSOR_LAT", "38.6270"), |
| "SENSOR_LON": os.environ.get("SENSOR_LON", "-90.1994"), |
| }, |
| } |
|
|
| CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) |
| CONFIG_PATH.write_text(json.dumps(resources, indent=2)) |
| print(json.dumps(resources, indent=2)) |
| print(f"\nWired → {CONFIG_PATH}") |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
