Python SDK
Online HTTP client + offline HULYAS 1.287 Hz reference implementation.
Downloads
zeq_client.py online · v1.1.3stdlib only
Thin HTTP wrapper around /api/zeq/compute + /api/zeq/prove. Verified live: NM21 {m1=5.972e24, m2=7.342e22, r=3.844e8} → 1.98049e20 N.
hulyas_framework.py offline · reference1047 lines
Original reference implementation by H. Zeq and A. Zeq. Full 42-operator Kinematic Operator spectrum, KO42.1/KO42.2 metric tensioners, master-equation ODE integrator, autotune-until-pass, strict ≤0.1 % validator. Air-gapped — zero network.
Quickstart
pip install requests # optional — the core client uses only stdlib
# Online — against zeq.dev
from zeq_client import ZeqClient
client = ZeqClient(token="zeq_ak_...")
result = client.compute("NM21", {"m1": 5.972e24, "m2": 7.342e22, "r": 3.844e8})
print(result["value"], result["unit"]) # → 1.98049e+20 N
# Offline — HULYAS reference (no network)
from hulyas_framework import autotune_until_pass
cko = autotune_until_pass("Simple harmonic motion with k=200 N/m, m=0.5 kg")
print(cko["error_pct"], cko["ko_used"]) # → 0.00072 ['KO42.1']
Full source — zeq_client.py
Why inline? The full SDK source is pasted below so the mathematics is auditable.
Copy-paste it, read it, self-host it — you never have to trust the Zeq API as a black box.
Use our key for convenience, or wire it into your own backend. The v1.1.3 binding contract and ZeqProof HMAC are identical either way.
zeq_client.py 9785 bytes 247 lines Python 3.8+
"""
ZeqClient — Python SDK wrapper for Zeq SDK API
==================================================
Solves two precision problems via pre/post processing:
PROBLEM 1: Sub-1e-7 precision floor → Dimensional Normalisation Layer
PROBLEM 2: α extraction variance → Phase-Optimal Sampling + Least-Squares Fit
The API engine itself is unchanged. This wrapper normalises inputs as
dimensionless ratios against natural units before transmission, then
denormalises results using the geometric mean of scale factors.
[Zeq Daemon] HulyaPulse 1.287 Hz — τ_zeqond = 0.777 s
"""
from __future__ import annotations
import functools
import json
import math
import operator
import time
import urllib.request
import urllib.error
from typing import Any
# ── Natural unit dictionary ────────────────────────────────────────────────
NATURAL_UNITS: dict[str, float] = {
"mass_electron": 9.10938e-31, # kg
"mass_proton": 1.67262e-27, # kg
"hbar": 1.054572e-34, # J·s
"h": 6.62607e-34, # J·s
"bohr_radius": 5.29177e-11, # m
"planck_length": 1.61626e-35, # m
"planck_mass": 2.17643e-8, # kg
"e_charge": 1.60218e-19, # C
"k_B": 1.38065e-23, # J/K
"solar_mass": 1.98892e30, # kg
"solar_radius": 6.96340e8, # m
"parsec": 3.08568e16, # m
"hubble_time": 4.35e17, # s
}
ALPHA_SPEC = 0.00129
F_HULYA = 1.287
TAU_ZEQOND = 0.777
# ── Dimensional normalisation ──────────────────────────────────────────────
def normalise(inputs: dict[str, float]) -> tuple[dict[str, float], dict[str, float]]:
"""Map every input to its closest natural unit; return (normalised, scales)."""
normalised: dict[str, float] = {}
scales: dict[str, float] = {}
for key, val in inputs.items():
if not isinstance(val, (int, float)) or val == 0:
normalised[key] = float(val) if isinstance(val, (int, float)) else 0.0
scales[key] = 1.0
continue
best_unit = None
best_log = float("inf")
for uval in NATURAL_UNITS.values():
ratio = abs(val) / uval
if ratio > 0:
log_ratio = abs(math.log10(ratio))
if log_ratio < best_log:
best_log = log_ratio
best_unit = uval
if best_unit and best_log < 3:
normalised[key] = val / best_unit
scales[key] = best_unit
else:
mag = 10 ** math.floor(math.log10(abs(val)))
normalised[key] = val / mag
scales[key] = mag
return normalised, scales
def denormalise(value: float, scales: dict[str, float]) -> float:
"""Restore physical magnitude via geometric mean of scale factors."""
if not scales:
return value
product = functools.reduce(operator.mul, scales.values(), 1.0)
geo_mean = product ** (1.0 / len(scales))
return value * geo_mean
# ── HTTP client ────────────────────────────────────────────────────────────
class ZeqClient:
"""
Python SDK client for the Zeq HTTP API.
Example:
client = ZeqClient(token="zeq_ak_...")
result = client.compute("NM21", {"m1": 5.972e24, "m2": 7.342e22, "r": 3.844e8})
print(result["value"], result["unit"])
"""
def __init__(
self,
token: str | None = None,
base_url: str = "https://www.zeq.dev",
min_interval_s: float = 0.0, # client-side throttle; 0 disables
# Backward-compat aliases for pre-1.1.3 code:
api_key: str | None = None,
) -> None:
self.base_url = base_url.rstrip("/")
self.api_key = token or api_key
if not self.api_key:
raise ValueError("ZeqClient requires a `token=` Bearer API key")
self.min_interval_s = min_interval_s
self._last_call = 0.0
def _throttle(self) -> None:
delta = time.monotonic() - self._last_call
if delta < self.min_interval_s:
time.sleep(self.min_interval_s - delta)
self._last_call = time.monotonic()
# raw call -------------------------------------------------------------
def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
self._throttle()
url = f"{self.base_url}{path}"
data = json.dumps(body).encode("utf-8")
req = urllib.request.Request(
url,
data=data,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
return {"error": f"HTTP {e.code}", "body": e.read().decode("utf-8", "replace")}
except Exception as e:
return {"error": str(e)}
# Public API — v1.1.3 contract -----------------------------------------
def compute(self, operator_id: str, inputs: dict[str, float]) -> dict[str, Any]:
"""
Call POST /api/zeq/compute with the v1.1.3 contract.
Returns the full response dict, including:
value — physical result (float)
unit — SI unit string (e.g. "N", "J", "m/s")
uncertainty — relative uncertainty (float)
operator_id — canonical operator ID
solver — "operator:<ID>" (attribution)
binding_overlay — "1.1.3" or later
zeqProof — SHA-256 HMAC (compliance)
compliance — FDA/ISO/GDPR envelope
zeqState — full kernel state
cko — Computational Kernel Output
"""
return self._post("/api/zeq/compute", {"operator_id": operator_id, "inputs": inputs})
def prove(self, operator_id: str, inputs: dict[str, float]) -> dict[str, Any]:
"""Call POST /api/zeq/prove — same payload as compute(), returns ZeqProof."""
return self._post("/api/zeq/prove", {"operator_id": operator_id, "inputs": inputs})
# Legacy / research methods --------------------------------------------
def compute_raw(self, operator_id: str, inputs: dict[str, float], domain: str = "physics") -> dict[str, Any]:
"""Legacy helper retained for the dimensional-normalisation research flow below."""
return self._post("/api/zeq/compute", {"operator_id": operator_id, "inputs": inputs})
def compute_normalized(self, operator_id: str, inputs: dict[str, float]) -> dict[str, Any]:
"""PROBLEM 1 fix — normalise → call → denormalise."""
norm_inputs, scales = normalise(inputs)
raw = self.compute_raw(operator_id, norm_inputs)
result = raw.get("result") or raw
# Extract numeric scalar to denormalise
for k in ("R_t", "value", "result"):
if isinstance(result, dict) and k in result and isinstance(result[k], (int, float)):
result[f"{k}_physical"] = denormalise(result[k], scales)
raw["_scales"] = scales
raw["_normalised_inputs"] = norm_inputs
return raw
# PROBLEM 2 — α extraction --------------------------------------------
def extract_alpha_clean(
self,
n_samples: int = 20,
min_sin_threshold: float = 0.7,
operator_id: str = "KO42",
max_attempts: int = 60,
) -> dict[str, Any]:
good: list[tuple[float, float]] = []
attempts = 0
while len(good) < n_samples and attempts < max_attempts:
attempts += 1
raw = self.compute_raw(operator_id, {"x": 1.0, "y": 1.0, "z": 1.0})
res = raw.get("result", {}) or {}
zs = raw.get("zeqState", {}) or {}
S_t = res.get("S_t")
R_t = res.get("R_t")
phase = zs.get("phase", 0)
if not S_t or R_t is None:
time.sleep(0.05)
continue
sin_val = math.sin(2 * math.pi * phase)
if abs(sin_val) >= min_sin_threshold:
good.append((sin_val, R_t / S_t - 1))
else:
time.sleep(0.02)
if not good:
return {"error": "no good samples", "attempts": attempts}
sx = sum(s * s for s, _ in good)
sxy = sum(s * r for s, r in good)
alpha_ls = sxy / sx if sx else 0.0
residuals = [r - alpha_ls * s for s, r in good]
mean_res = sum(residuals) / len(residuals)
var = sum((x - mean_res) ** 2 for x in residuals) / len(residuals)
return {
"alpha": alpha_ls,
"alpha_spec": ALPHA_SPEC,
"deviation_pct": abs(alpha_ls - ALPHA_SPEC) / ALPHA_SPEC * 100,
"residual_std": math.sqrt(var),
"n_samples": len(good),
"attempts": attempts,
}
# ── CLI verification ───────────────────────────────────────────────────────
if __name__ == "__main__":
import os
import sys
token = os.environ.get("ZEQ_TOKEN")
if not token:
print("Set ZEQ_TOKEN=<your zeq_ak_...> to run the smoke test.", file=sys.stderr)
sys.exit(1)
client = ZeqClient(token=token)
print("═" * 70)
print("NM21 — Earth–Moon gravitational force (expected ≈ 1.98e20 N)")
print("═" * 70)
res = client.compute("NM21", {"m1": 5.972e24, "m2": 7.342e22, "r": 3.844e8})
print(f"value = {res.get('value')}")
print(f"unit = {res.get('unit')}")
print(f"uncertainty = {res.get('uncertainty')}")
print(f"solver = {res.get('solver')}")
print(f"binding_overlay = {res.get('binding_overlay')}")
print(f"zeqProof = {(res.get('zeqProof') or '')[:24]}…")