"""ohmatic — the agent-friendly CLI for the hosted Ohmatic API. One stdlib-only file, no dependencies: a shell-capable agent (or a human) drives the whole verify/repair loop without speaking HTTP or MCP. Everything an agent needs to behave well is surfaced explicitly: * STABLE EXIT CODES (the contract): 0 circuit passed 1 circuit failed checks (repairs printed) 2 input/transport/server error 3 auth or payment required (message + signup/checkout URL printed) 4 round cap reached (killswitch: STOP looping, improve the prompt) * GENERATION ID discipline: one circuit's repair loop bills as ONE generation only when every round carries the same generation_id. `verify` mints one when absent, prints it on every result, and the re-run command is echoed verbatim so a retrying agent cannot accidentally open a fresh billed generation. * Line-oriented output (greppable, diff-stable), or `--json` for the raw response. * stdin everywhere: `-` reads the circuit/netlist from a pipe. Credentials resolve in order: --api-key flag, OHMATIC_API_KEY, the config file (~/.ohmatic/config.json, or the path in OHMATIC_CONFIG). `ohmatic init` mints a free anonymous key and stores it there; a raw circuit string is accepted wherever a circuit is expected (the server tolerates fenced/prose model output). py cli/ohmatic.py init py cli/ohmatic.py verify circuit.json py cli/ohmatic.py verify circuit.json --gen-id ab12cd34ef56 # the retry form cat circuit.json | py cli/ohmatic.py verify - --json py cli/ohmatic.py import-kicad board.net -o circuit.json py cli/ohmatic.py export-kicad circuit.json --format schematic -o out.kicad_sch """ from __future__ import annotations import argparse import json import os import secrets import sys import urllib.error import urllib.request from pathlib import Path from typing import Any DEFAULT_URL = "https://ohmatic.dev" _UA = "ohmatic-cli/0.1" EXIT_PASS, EXIT_FAIL, EXIT_ERROR, EXIT_AUTH, EXIT_CAPPED = 0, 1, 2, 3, 4 class CliError(Exception): """Human-readable failure with an exit code; main() prints it to stderr.""" def __init__(self, message: str, code: int = EXIT_ERROR): super().__init__(message) self.code = code # ── config / credential resolution ───────────────────────────────────────────── def _config_path() -> Path: override = os.environ.get("OHMATIC_CONFIG", "").strip() return Path(override) if override else Path.home() / ".ohmatic" / "config.json" def _load_config() -> dict[str, Any]: try: data = json.loads(_config_path().read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except (OSError, ValueError): return {} def _save_config(update: dict[str, Any]) -> Path: path = _config_path() path.parent.mkdir(parents=True, exist_ok=True) merged = {**_load_config(), **update} path.write_text(json.dumps(merged, indent=2) + "\n", encoding="utf-8") return path def _api_url(args: argparse.Namespace) -> str: url = (getattr(args, "url", None) or os.environ.get("OHMATIC_API_URL", "").strip() or _load_config().get("api_url") or DEFAULT_URL) return str(url).rstrip("/") def _api_key(args: argparse.Namespace, *, required: bool = True) -> str: key = (getattr(args, "api_key", None) or os.environ.get("OHMATIC_API_KEY", "").strip() or _load_config().get("api_key") or "") if required and not key: raise CliError("no API key. Run `ohmatic init` for a free anonymous key, or set " "OHMATIC_API_KEY / --api-key.", EXIT_AUTH) return str(key) # ── transport ────────────────────────────────────────────────────────────────── def _request(args: argparse.Namespace, method: str, path: str, *, body: dict[str, Any] | None = None, key: str = "") -> dict[str, Any]: """One API call. Returns the parsed JSON body; raises CliError with the mapped exit code (3 for auth/payment, 2 for everything else) and the server's own message, so an agent sees the actionable text, not a stack trace.""" url = _api_url(args) + path headers = {"User-Agent": _UA, "Accept": "application/json"} data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode("utf-8") if key: headers["Authorization"] = f"Bearer {key}" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=float(getattr(args, "timeout", 120))) as resp: payload = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: raise CliError(_http_error_text(exc), EXIT_AUTH if exc.code in (401, 402) else EXIT_ERROR) except (urllib.error.URLError, OSError, ValueError) as exc: raise CliError(f"cannot reach the Ohmatic API at {url}: {exc}", EXIT_ERROR) if not isinstance(payload, dict): raise CliError("unexpected non-object response from the API", EXIT_ERROR) return payload def _http_error_text(exc: urllib.error.HTTPError) -> str: """The server's FastAPI `detail` (string or structured), flattened to one actionable message: text first, then any URLs it carried (signup/checkout).""" try: detail = json.loads(exc.read().decode("utf-8")).get("detail") except Exception: # noqa: BLE001 — error-path robustness: any body shape reduces to the status line detail = None if isinstance(detail, dict): parts = [str(detail.get("message") or detail.get("error") or "").strip()] parts += [str(v) for k, v in sorted(detail.items()) if k.endswith("_url") and isinstance(v, str) and v] text = " | ".join(p for p in parts if p) else: text = str(detail or "").strip() return f"HTTP {exc.code}: {text or exc.reason}" # ── input helpers ────────────────────────────────────────────────────────────── def _read_input(source: str) -> str: if source == "-": return sys.stdin.read() try: return Path(source).read_text(encoding="utf-8-sig") except OSError as exc: raise CliError(f"cannot read {source}: {exc}") def _circuit_value(text: str) -> Any: """A circuit as the API accepts it: parsed JSON when the input parses (a full request body is unwrapped to its `circuit`), otherwise the raw string — the server tolerates fenced/prose model output.""" try: parsed = json.loads(text) except ValueError: return text if isinstance(parsed, dict) and "circuit" in parsed and "components" not in parsed: return parsed["circuit"] return parsed def _round_body(args: argparse.Namespace, circuit: Any) -> tuple[dict[str, Any], str]: gen_id = getattr(args, "gen_id", None) or secrets.token_hex(8) body: dict[str, Any] = {"circuit": circuit, "generation_id": gen_id} if getattr(args, "max_rounds", None): body["max_rounds"] = args.max_rounds return body, gen_id # ── output helpers ───────────────────────────────────────────────────────────── def _out(line: str = "") -> None: print(line) def _emit_json(payload: dict[str, Any]) -> None: print(json.dumps(payload, indent=2)) def _print_repairs(feedback: dict[str, Any]) -> None: for i, rep in enumerate(feedback.get("repairs", []), 1): where = " ".join(x for x in (rep.get("component_id"), rep.get("pin_ref"), rep.get("net_name")) if x) _out(f"[{i}] {rep.get('code', '?')} ({rep.get('severity', 'error')})" + (f" @ {where}" if where else "")) for label, field in (("problem", "problem"), ("why", "why"), ("expected", "expected"), ("actual", "actual"), ("fix", "repair_hint")): value = str(rep.get(field, "") or "").strip() if value: _out(f" {label}: {value}") def _print_capped(payload: dict[str, Any], gen_id: str) -> int: kill = bool(payload.get("killswitch")) _out(f"CAPPED round={payload.get('round')} max_rounds={payload.get('max_rounds')} " f"killswitch={str(kill).lower()} generation_id={gen_id}") _out(str(payload.get("message", ""))) return EXIT_CAPPED def _print_verdict(payload: dict[str, Any], gen_id: str, prog: str, source: str) -> int: passed = bool(payload.get("passed")) head = "PASS" if passed else "FAIL" _out(f"{head} findings={payload.get('diagnostic_count', 0)} " f"round={payload.get('round', '?')} billed_usd={payload.get('billed_usd', 0)} " f"free_remaining={payload.get('free_generations_remaining', '?')} " f"generation_id={gen_id}") if passed: return EXIT_PASS _print_repairs(payload.get("feedback") or {}) _out() _out(f"re-verify THIS circuit after fixes (same billed generation): " f"{prog} verify {source} --gen-id {gen_id}") return EXIT_FAIL # ── subcommands ──────────────────────────────────────────────────────────────── def _cmd_health(args: argparse.Namespace) -> int: payload = _request(args, "GET", "/v1/health") if args.json: _emit_json(payload) else: _out(f"ok build={payload.get('build', '')} schema={payload.get('schema_version', '')}") return EXIT_PASS if payload.get("status") == "ok" else EXIT_ERROR def _cmd_init(args: argparse.Namespace) -> int: if _api_key(args, required=False) and not args.force: raise CliError("an API key is already configured; use --force to mint a fresh " "anonymous key anyway (free anon keys are rate-limited per network).") payload = _request(args, "POST", "/v1/anon-key", body={}) key = payload.get("api_key") if not key: raise CliError("key minting did not return a key; try again in a minute.") path = _save_config({"api_key": key, "api_url": _api_url(args)}) if args.json: _emit_json({"config": str(path), **payload}) else: _out(f"anonymous key stored in {path}") _out(f"free generations on this key: {payload.get('free_generations', '?')} " f"(lifetime; sign up for a monthly allowance)") return EXIT_PASS def _cmd_verify(args: argparse.Namespace) -> int: body, gen_id = _round_body(args, _circuit_value(_read_input(args.circuit))) payload = _request(args, "POST", "/v1/verify", body=body, key=_api_key(args)) if args.json: _emit_json({**payload, "generation_id": gen_id}) return (EXIT_CAPPED if payload.get("capped") else EXIT_PASS if payload.get("passed") else EXIT_FAIL) if payload.get("capped"): return _print_capped(payload, gen_id) return _print_verdict(payload, gen_id, args.prog_name, args.circuit) def _cmd_feedback(args: argparse.Namespace) -> int: body, gen_id = _round_body(args, _circuit_value(_read_input(args.circuit))) payload = _request(args, "POST", "/v1/repair-feedback", body=body, key=_api_key(args)) if args.json: _emit_json({**payload, "generation_id": gen_id}) return EXIT_CAPPED if payload.get("capped") else EXIT_PASS if payload.get("capped"): return _print_capped(payload, gen_id) _out(f"valid={str(bool(payload.get('valid'))).lower()} " f"errors={payload.get('error_count', 0)} warnings={payload.get('warning_count', 0)} " f"round={payload.get('round', '?')} generation_id={gen_id}") _print_repairs(payload) return EXIT_PASS def _cmd_credits(args: argparse.Namespace) -> int: payload = _request(args, "GET", "/v1/credits", key=_api_key(args)) if args.json: _emit_json(payload) else: _out(f"free_generations_remaining={payload.get('free_generations_remaining', '?')}") return EXIT_PASS def _cmd_status(args: argparse.Namespace) -> int: payload = _request(args, "GET", "/v1/billing/status", key=_api_key(args)) if args.json: _emit_json(payload) else: for k in sorted(payload): _out(f"{k}={payload[k]}") return EXIT_PASS def _cmd_import_kicad(args: argparse.Namespace) -> int: netlist = _read_input(args.netlist) gen_id = args.gen_id or secrets.token_hex(8) body: dict[str, Any] = {"netlist": netlist, "generation_id": gen_id} if args.max_rounds: body["max_rounds"] = args.max_rounds payload = _request(args, "POST", "/v1/kicad/import", body=body, key=_api_key(args)) if payload.get("capped") and not args.json: return _print_capped(payload, gen_id) circuit = payload.get("circuit") if args.out and circuit is not None: Path(args.out).write_text(json.dumps(circuit, indent=2) + "\n", encoding="utf-8") if args.json: _emit_json({**payload, "generation_id": gen_id}) else: if args.out: _out(f"circuit written to {args.out}") advisory = (payload.get("pinout_advisory") or {}).get("message") or "" if advisory: _out(f"note: {advisory}") return _print_verdict(payload, gen_id, args.prog_name, args.out if args.out else "") return EXIT_PASS if payload.get("passed") else EXIT_FAIL def _cmd_export_kicad(args: argparse.Namespace) -> int: body = {"circuit": _circuit_value(_read_input(args.circuit)), "format": args.format} payload = _request(args, "POST", "/v1/kicad/export", body=body, key=_api_key(args)) if args.json: _emit_json(payload) return EXIT_PASS if args.format == "both": if not args.out: _emit_json(payload) return EXIT_PASS outdir = Path(args.out) outdir.mkdir(parents=True, exist_ok=True) for doc in (payload.get("netlist"), payload.get("schematic")): if isinstance(doc, dict): content = doc.get("netlist") or doc.get("schematic") or "" target = outdir / str(doc.get("filename", "ohmatic.out")) target.write_text(content, encoding="utf-8") _out(f"wrote {target}") return EXIT_PASS content = payload.get("netlist") or payload.get("schematic") or "" if args.out: Path(args.out).write_text(content, encoding="utf-8") _out(f"wrote {args.out}") else: sys.stdout.write(content) return EXIT_PASS def _cmd_redeem(args: argparse.Namespace) -> int: payload = _request(args, "POST", "/v1/billing/redeem", body={"code": args.code}, key=_api_key(args)) _emit_json(payload) if args.json else _out(json.dumps(payload)) return EXIT_PASS def _cmd_checkout(args: argparse.Namespace) -> int: payload = _request(args, "POST", "/v1/billing/checkout", body={}, key=_api_key(args)) if args.json: _emit_json(payload) else: url = next((v for k, v in payload.items() if k.endswith("url") and isinstance(v, str)), "") _out(url or json.dumps(payload)) return EXIT_PASS # ── argument wiring ──────────────────────────────────────────────────────────── def _build_parser(prog: str) -> argparse.ArgumentParser: common = argparse.ArgumentParser(add_help=False) common.add_argument("--url", help="API base URL (default: OHMATIC_API_URL / config / hosted)") common.add_argument("--api-key", help="API key (default: OHMATIC_API_KEY / config)") common.add_argument("--json", action="store_true", help="print the raw JSON response") common.add_argument("--timeout", type=float, default=120.0, help="request timeout seconds") rounds = argparse.ArgumentParser(add_help=False) rounds.add_argument("--gen-id", help="generation id: REUSE the printed one when " "re-verifying the same circuit after fixes") rounds.add_argument("--max-rounds", type=int, help="per-circuit spend guard (rounds)") p = argparse.ArgumentParser( prog=prog, description="Agent-friendly CLI for the Ohmatic verification API. " "Exit codes: 0 pass, 1 fail, 2 error, 3 auth/payment, 4 round cap.") sub = p.add_subparsers(dest="command", required=True) sub.add_parser("health", parents=[common], help="API liveness + build tag") sp = sub.add_parser("init", parents=[common], help="mint a free anonymous key into the config") sp.add_argument("--force", action="store_true", help="mint even if a key is configured") sp = sub.add_parser("verify", parents=[common, rounds], help="verify a circuit (file or -) and print repairs on failure") sp.add_argument("circuit", help="circuit JSON path, or - for stdin (raw model text accepted)") sp = sub.add_parser("feedback", parents=[common, rounds], help="repair feedback only (same billing round semantics as verify)") sp.add_argument("circuit", help="circuit JSON path, or - for stdin") sub.add_parser("credits", parents=[common], help="free generations remaining on this key") sub.add_parser("status", parents=[common], help="billing/account status") sp = sub.add_parser("import-kicad", parents=[common, rounds], help="KiCad .net -> Ohmatic circuit JSON + first verify round") sp.add_argument("netlist", help=".net path, or - for stdin") sp.add_argument("-o", "--out", help="write the imported circuit JSON here") sp = sub.add_parser("export-kicad", parents=[common], help="Ohmatic circuit JSON -> KiCad netlist/schematic") sp.add_argument("circuit", help="circuit JSON path, or - for stdin") sp.add_argument("--format", choices=("netlist", "schematic", "both"), default="netlist") sp.add_argument("-o", "--out", help="output file (or directory for --format both)") sp = sub.add_parser("redeem", parents=[common], help="redeem a credit code") sp.add_argument("code") sub.add_parser("checkout", parents=[common], help="get a top-up checkout link") return p _COMMANDS = { "health": _cmd_health, "init": _cmd_init, "verify": _cmd_verify, "feedback": _cmd_feedback, "credits": _cmd_credits, "status": _cmd_status, "import-kicad": _cmd_import_kicad, "export-kicad": _cmd_export_kicad, "redeem": _cmd_redeem, "checkout": _cmd_checkout, } def main(argv: list[str] | None = None, prog: str = "ohmatic") -> int: parser = _build_parser(prog) args = parser.parse_args(argv) args.prog_name = prog try: return _COMMANDS[args.command](args) except CliError as exc: print(f"error: {exc}", file=sys.stderr) return exc.code if __name__ == "__main__": raise SystemExit(main())