File size: 2,725 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 | """CLI entry point for symbolic recursion coherence system."""
from __future__ import annotations
import argparse
import asyncio
import json
import sys
from .coherence import benchmark_operators
from .data.public_feeds import PublicSensorFeed
from .integrations.atlas_bridge import AtlasBridge
from .integrations.primal_bridge import PrimalBridge
from .kernel import run_worked_examples
async def _run_live(args: argparse.Namespace) -> None:
feed = PublicSensorFeed(latitude=args.lat, longitude=args.lon)
if args.hours > 1:
series = await feed.fetch_series(hours=args.hours)
bridge = PrimalBridge(a=args.a, b=args.b)
result = bridge.from_sensor_snapshots(series)
atlas = AtlasBridge()
import numpy as np
from .kernel import SymbolicRecursionKernel
q = np.array([s.q_t for s in series])
dx = SymbolicRecursionKernel(a=args.a).integrate_series(q, dt=3600.0)
fusion = atlas.fuse_with_symbolic_recursion(q, dx)
print(json.dumps({"series_points": len(series), "primal": result, "atlas": fusion}, indent=2))
else:
snap = await feed.fetch_current()
print(json.dumps({
"timestamp": snap.timestamp,
"Q_t": snap.q_t,
"DT": snap.dt_dev,
"DP": snap.dp_dev,
"DEM": snap.dem_dev,
"DW": snap.dw_dev,
"source": snap.source,
}, indent=2))
def main(argv: list[str] | None = None) -> None:
parser = argparse.ArgumentParser(description="Symbolic Recursion Coherence Kernel")
sub = parser.add_subparsers(dest="command")
sub.add_parser("examples", help="Run worked forcing-function examples")
sub.add_parser("benchmark", help="Benchmark vs 20 classical operators")
live = sub.add_parser("live", help="Fetch public sensor data and compute")
live.add_argument("--lat", type=float, default=38.6270)
live.add_argument("--lon", type=float, default=-90.1994)
live.add_argument("--hours", type=int, default=24)
live.add_argument("--a", type=float, default=1.0)
live.add_argument("--b", type=float, default=0.091)
sub.add_parser("serve", help="Start API server on port 8080")
args = parser.parse_args(argv)
if args.command == "examples":
print(json.dumps(run_worked_examples(), indent=2))
elif args.command == "benchmark":
rows = benchmark_operators()
print(json.dumps(rows, indent=2))
elif args.command == "live":
asyncio.run(_run_live(args))
elif args.command == "serve":
import uvicorn
uvicorn.run("symbolic_recursion.server:app", host="0.0.0.0", port=8080, reload=False)
else:
parser.print_help()
sys.exit(1) |