File size: 4,230 Bytes
5d07399 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """
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()) |