SolveForce Phone Gateway — Transparency & Source
Language‑first, read‑only, phone‑native. From zero to one—spelled and auditable.
This page publishes the SolveForce phone‑first gateway openly: what it does, how to run it on Android, how to verify integrity, and the entire source code. The design is read‑only by default, so safety is visible in the spelling of the code itself.
🔎 What this is
- Phone‑native gateway you run on Android (Termux or Pydroid).
- HTTP UI at
/uifor a fingertip console. - Adapters (read‑only) for battery (if Termux:API is present) and basic network info.
- Translator that normalizes math operators and currency symbols into canonical tokens (
× → *,÷ → /,≤ → <=,$ → <USD>,€ → <EUR>,₽ → <RUB>, etc.). - JSONL export (opt‑in) for ClickHouse ingestion or audit trails.
Scope (truthful constraints):
- No writes, no control ops, no device actuation.
- No external calls; it only serves HTTP locally (or your LAN if you bind it).
- Background polling is off unless you pass
--poll N.
⬇️ Downloads
Upload the file(s) to your Media Library or repository and link here:
- Phone Zero (minimal seed) — optional reference
- Phone One (JSONL exports) — preferred for posting
solveforce_phone_one.py— [your link here]- SHA‑256: (publish after you upload; see integrity section below)
If you don’t want to host a file, you can copy the Canonical Source below into a new file named
solveforce_phone_one.pyon the phone.
✅ Integrity & verification (Android)
Compute the SHA‑256 before running:
sha256sum solveforce_phone_one.py
# Compare with the SHA you publish on this page
Or via Python:
python - << 'PY'
import hashlib, sys
p="solveforce_phone_one.py"
print(hashlib.sha256(open(p,'rb').read()).hexdigest())
PY
📱 Android install (Termux)
pkg update -y && pkg upgrade -y
pkg install -y python
termux-setup-storage # allow access to /sdcard if exporting
# optional battery plugin:
pkg install -y termux-api
Run (local only):
python solveforce_phone_one.py --host 127.0.0.1 --port 8080 --poll 0
# Open on the phone: http://127.0.0.1:8080/ui
Run (LAN visible) with JSONL exports every 5s:
mkdir -p /sdcard/solveforce/exports
python solveforce_phone_one.py --host 0.0.0.0 --port 8080 --poll 5 --export --export-dir /sdcard/solveforce/exports
# From laptop on same Wi‑Fi: http://PHONE_IP:8080/ui
Auto‑start (Termux:Boot):
mkdir -p ~/.termux/boot
cat > ~/.termux/boot/start-gateway.sh << 'SH'
#!/data/data/com.termux/files/usr/bin/sh
cd ~/solveforce
termux-wake-lock
nohup python solveforce_phone_one.py --host 0.0.0.0 --port 8080 --poll 5 --export --export-dir /sdcard/solveforce/exports >/sdcard/solveforce/phone.log 2>&1 &
SH
chmod +x ~/.termux/boot/start-gateway.sh
🧭 Endpoints
/ui— mobile console/health— uptime + export status/state— plugins + last reads/read?plugin=battery— Termux:API battery (if available)/read?plugin=net— IP hints (ip -o -4orgetprop)/read?plugin=all— read all plugins/translate?q=$199.99+VAT=€210≈— normalized token stream/metrics— Prometheus text (minimal counters)
🔐 Security posture (tell it like it is)
- Read‑only by design. There is no control surface.
- Local by default. Binding
--host 127.0.0.1keeps it on the phone. - LAN exposure is your choice. Use
--host 0.0.0.0only on trusted Wi‑Fi. - Public access not advised without an auth layer. For remote access, use a secure overlay (e.g., Tailscale).
- No secrets in code. If you ever add credentials, keep them in env/keystore.
- Exports are explicit. Nothing is written unless you pass
--export.
🗃️ JSONL exports (ClickHouse‑ready)
- One file per plugin (e.g.,
/sdcard/solveforce/exports/battery.jsonl). - Each line is a
JSONEachRowobject:
{"plugin":"battery","_ts":"2025-08-19T07:34:22.123456+00:00","data":{...}}
Example ClickHouse tables:
CREATE TABLE gw_battery (plugin String, _ts DateTime64(3,'UTC'), data JSON)
ENGINE=MergeTree ORDER BY _ts;
CREATE TABLE gw_net (plugin String, _ts DateTime64(3,'UTC'), data JSON)
ENGINE=MergeTree ORDER BY _ts;
🧾 Canonical Source (Phone One, corrected)
Save this as
solveforce_phone_one.py. This is the exact code the page describes—no external dependencies, read‑only, and includes JSONL export toggles.
Note: The ruble symbol maps to the ISO code RUB (corrected).
#!/usr/bin/env python3
# SolveForce Phone One — phone-first gateway with JSONL exports (read-only)
# MIT © 2025 Ronald Joseph Legarski, Jr. — Published by SolveForce
#
# Zero → One:
# - Single-file HTTP gateway, mobile UI at /ui
# - Plugins: battery (Termux:API if present), net (basic IP discovery)
# - --export (JSONL, ClickHouse-ready JSONEachRow), --export-dir <dir>
# - Writes one line per read: {"plugin": "...", "_ts": "...", "data": {...}}
import os, sys, argparse, json, time, threading, shutil, subprocess
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
from datetime import datetime, timezone
from typing import Optional
STARTED = time.time()
def now_utc_iso() -> str:
return datetime.now(timezone.utc).isoformat()
# ---------- Exporter (JSONL) ----------
class JsonlExporter:
def __init__(self, enabled: bool = False, directory: str = "exports"):
self.enabled = bool(enabled)
self.dir = directory
self.last_error: Optional[str] = None
if self.enabled:
try:
os.makedirs(self.dir, exist_ok=True)
except Exception as e:
self.last_error = f"mkdir failed: {e}"
self.enabled = False
def write(self, plugin: str, data: dict):
if not self.enabled:
return
try:
line = json.dumps({"plugin": plugin, "_ts": now_utc_iso(), "data": data}, ensure_ascii=False)
path = os.path.join(self.dir, f"{plugin}.jsonl")
with open(path, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception as e:
self.last_error = f"write failed: {e}"
EXPORTER = JsonlExporter(False, "exports")
# ---------- Minimal plugin framework ----------
class Plugin:
NAME = "plugin"
def __init__(self, **kwargs): pass
def read(self):
return {"ok": True}
def have_cmd(cmd: str) -> bool:
return shutil.which(cmd) is not None
def run_json_cmd(args, timeout=3):
try:
out = subprocess.check_output(args, timeout=timeout, stderr=subprocess.STDOUT)
txt = out.decode("utf-8", "replace").strip()
return json.loads(txt)
except Exception as e:
return {"error": str(e), "args": args}
class BatteryTermux(Plugin):
NAME = "battery"
def read(self):
if not have_cmd("termux-battery-status"):
return {"available": False, "note": "termux-battery-status not found"}
data = run_json_cmd(["termux-battery-status"], timeout=2)
data["available"] = "error" not in data
return data
class NetIfacesLite(Plugin):
NAME = "net"
def read(self):
if have_cmd("ip"):
try:
txt = subprocess.check_output(["ip", "-o", "-4", "addr", "show"], timeout=2).decode("utf-8","replace")
v4 = []
for line in txt.splitlines():
parts = line.split()
if len(parts) >= 4:
iface = parts[1]
cidr = parts[3]
v4.append({"iface": iface, "cidr": cidr})
return {"available": True, "ipv4": v4}
except Exception as e:
return {"available": False, "error": str(e)}
if have_cmd("getprop"):
try:
wifi_ip = subprocess.check_output(["getprop", "dhcp.wlan0.ipaddress"], timeout=2).decode().strip()
return {"available": True, "wifi_ip": wifi_ip}
except Exception as e:
return {"available": False, "error": str(e)}
return {"available": False, "note": "no ip/getprop available"}
BUILTIN_PLUGINS = [BatteryTermux, NetIfacesLite]
class Registry:
def __init__(self, exporter: JsonlExporter):
self.plugins = {cls.NAME: cls() for cls in BUILTIN_PLUGINS}
self.latest = {}
self.lock = threading.Lock()
self.exporter = exporter
def names(self):
return list(self.plugins.keys())
def read_one(self, name):
p = self.plugins.get(name)
if not p: return {"error": f"unknown plugin '{name}'"}
res = p.read()
row = {"data": res, "_ts": now_utc_iso()}
with self.lock:
self.latest[name] = row
try:
self.exporter.write(name, res)
except Exception:
pass
return row
def read_all(self):
out = {}
for n in self.names():
out[n] = self.read_one(n)
return out
def state(self):
with self.lock:
return {"plugins": self.names(), "latest": dict(self.latest)}
class Poller(threading.Thread):
def __init__(self, interval_sec: int, registry: Registry):
super().__init__(daemon=True); self.interval = max(0, int(interval_sec)); self.stop_evt = threading.Event()
self.registry = registry
def run(self):
if self.interval <= 0: return
while not self.stop_evt.is_set():
try:
self.registry.read_all()
except Exception:
pass
self.stop_evt.wait(self.interval)
def stop(self): self.stop_evt.set()
MOBILE_UI = """<!doctype html>
<html><head><meta charset="utf-8"/><meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>SolveForce Phone One</title>
<style>
body{font-family:system-ui,Segoe UI,Roboto,Helvetica,Arial,sans-serif;margin:1rem;line-height:1.4}
header{font-weight:700;margin-bottom:.8rem}
.row{display:flex;gap:.5rem;flex-wrap:wrap}
button{padding:.6rem 1rem;border-radius:.6rem;border:1px solid #888;background:#fff}
.card{border:1px solid #ccc;border-radius:.8rem;padding:1rem;margin:.6rem 0}
code,pre{font-family:ui-monospace,Consolas,Menlo,monospace}
.mono{white-space:pre-wrap}
input[type=text]{padding:.5rem;border:1px solid #aaa;border-radius:.5rem;min-width:240px}
</style></head>
<body>
<header>📱 SolveForce Phone One — Root UI</header>
<div class="row">
<button onclick="hit('/health')">/health</button>
<button onclick="hit('/state')">/state</button>
<button onclick="hit('/read?plugin=battery')">battery</button>
<button onclick="hit('/read?plugin=net')">net</button>
<button onclick="hit('/metrics', true)">/metrics</button>
</div>
<div class="card">
<form onsubmit="tx();return false;">
<label>Translate:</label>
<input id="q" value="$199.99+VAT=€210≈"/>
<button>Go</button>
</form>
</div>
<div id="out" class="card mono">Ready.</div>
<script>
async function hit(path, raw){ const r=await fetch(path); const t=raw? await r.text(): await r.json(); show(t); }
async function tx(){ const q=document.getElementById('q').value; const r=await fetch('/translate?q='+encodeURIComponent(q)); show(await r.json()); }
function show(x){ const el=document.getElementById('out'); el.textContent = (typeof x==='string')?x:JSON.stringify(x,null,2); }
hit('/health');
</script>
</body></html>
"""
MATH_OPS = {
"+": "+", "+": "+",
"-": "-", "−": "-", "–": "-", "—": "-",
"×": "*", "∙": "*", "·": "*", "⋅": "*", "*": "*",
"÷": "/", "∕": "/", "/": "/",
"=": "=", "≈": "~", "≃": "~", "≅": "~",
"<": "<", "≤": "<=", "⩽": "<=",
">": ">", "≥": ">=", "⩾": ">=",
"^": "^", "%": "%"
}
CURRENCIES = {
"A$": "AUD", "C$": "CAD", "NZ$": "NZD", "HK$": "HKD", "S$": "SGD", "R$": "BRL", "MX$": "MXN",
"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₩": "KRW", "₹": "INR", "₽": "RUB",
"₫": "VND", "₦": "NGN", "₱": "PHP", "฿": "THB", "₵": "GHS", "₡": "CRC", "₭": "LAK",
"₴": "UAH", "₲": "PYG", "₮": "MNT", "₼": "AZN", "₸": "KZT", "₾": "GEL", "₿": "BTC"
}
CURR_KEYS = sorted(CURRENCIES.keys(), key=lambda k: len(k), reverse=True)
def translate_query(q: str):
i=0; tokens=[]; out=[]
while i < len(q):
matched=False
for key in CURR_KEYS:
if q.startswith(key, i):
iso=CURRENCIES[key]; tokens.append({"type":"currency","raw":key,"iso":iso})
out.append(f"<{iso}>"); i+=len(key); matched=True; break
if matched: continue
ch=q[i]
if ch in MATH_OPS:
canon=MATH_OPS[ch]; tokens.append({"type":"operator","raw":ch,"op":canon}); out.append(canon); i+=1; continue
if ch.isdigit() or (ch=="." and i+1<len(q) and q[i+1].isdigit()):
j=i+1
while j<len(q) and (q[j].isdigit() or q[j] in ".,_"): j+=1
num=q[i:j]; tokens.append({"type":"number","raw":num}); out.append(num.replace(",","")); i=j; continue
if ch.isalpha():
j=i+1
while j<len(q) and (q[j].isalpha() or q[j] in "-_/"): j+=1
word=q[i:j]; tokens.append({"type":"text","raw":word}); out.append(word); i=j; continue
out.append(ch); i+=1
return {"input": q, "normalized": "".join(out), "tokens": tokens}
class Handler(BaseHTTPRequestHandler):
server_version = "SolveForcePhoneOne/0.1"
def _send(self, code, payload, ct="application/json; charset=utf-8"):
data = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False, separators=(",",":"))
body = data.encode("utf-8")
self.send_response(code); self.send_header("Content-Type", ct)
self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
def log_message(self, fmt, *args): return
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/ui":
self._send(200, MOBILE_UI, ct="text/html; charset=utf-8"); return
if parsed.path == "/ping":
self._send(200, {"pong": True, "time": now_utc_iso()}); return
if parsed.path == "/health":
export = {"enabled": EXPORTER.enabled, "dir": EXPORTER.dir, "last_error": EXPORTER.last_error}
self._send(200, {"status":"ok","time":now_utc_iso(),"uptime_sec": int(time.time()-STARTED), "export": export}); return
if parsed.path == "/state":
self._send(200, REG.state()); return
if parsed.path == "/read":
name = parse_qs(parsed.query).get("plugin", [""])[0]
if name == "all": self._send(200, REG.read_all()); return
self._send(200, REG.read_one(name)); return
if parsed.path == "/translate":
q = parse_qs(parsed.query).get("q", [""])[0]
self._send(200, translate_query(q)); return
if parsed.path == "/metrics":
up = int(time.time() - STARTED)
names = REG.names()
lines = [f"solveforce_up {up}", f"solveforce_plugins {len(names)}"]
self._send(200, "\n".join(lines)+"\n", ct="text/plain; version=0.0.4; charset=utf-8"); return
self._send(404, {"error":"not found","path": parsed.path})
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", default="8080")
ap.add_argument("--poll", default="0", help="poll interval seconds; 0 = no background poll")
ap.add_argument("--export", action="store_true", help="enable JSONL export (JSONEachRow)")
ap.add_argument("--export-dir", default="exports", help="directory for JSONL files")
args = ap.parse_args()
global EXPORTER
EXPORTER = JsonlExporter(enabled=args.export, directory=args.export_dir)
global REG
REG = Registry(exporter=EXPORTER)
poll_int = int(args.poll)
poller = Poller(interval_sec=poll_int, registry=REG)
poller.start()
host, port = args.host, int(args.port)
httpd = HTTPServer((host, port), Handler)
print(f"[SolveForce] Phone One UI → http://{host}:{port}/ui")
if EXPORTER.enabled:
print(f"[SolveForce] Exporting JSONL → {EXPORTER.dir}")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
poller.stop()
httpd.server_close()
if __name__ == "__main__":
main()
🧭 Changelog (public)
- v0.1.0 — Phone Zero (seed): single‑file UI + translator + minimal plugins, no exports.
- v0.1.1 — Phone One (current): adds JSONL export toggles (
--export,--export-dir),/healthnow reports export status.
🪪 License & attribution
- MIT License — © 2025 Ronald Joseph Legarski, Jr.
- Published by: SolveForce — Nationwide & International Services.
- Contact: (888) 765‑8301
This is read‑only software published in good faith, without warranty. Use on trusted networks, with care, and with the discipline of spelled semantics.
🧠 Why we publish this (the recursive promise)
Language orders the machine; the machine carries the message. By spelling every symbol into a canonical grammar and keeping writes off by default, we make infrastructure legible, auditable, and consistent—from the phone in your hand to the grid at the edge.
Want me to add Step Two (plugin scaffolder & auto‑loader) and update this page with the new sections?
Done—Step Two is now real: dynamic plugins + scaffolder + admin refresh. – SolveForce Communications