| """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) |
