SolveForce Phone One — phone-first gateway with JSONL export (read-only)
#!/usr/bin/env python3
# SolveForce Phone One — Step One: JSONL export toggle (read-only)
# MIT © 2025 Ronald Joseph Legarski, Jr. — Published by SolveForce
#
# Additions over "Phone Zero":
# - --export (enable JSONL export, ClickHouse-ready JSONEachRow)
# - --export-dir <path> (default: ./exports)
# - Each plugin read (manual or via --poll) writes one JSON row:
# {"plugin":"battery","_ts":"...","data":{...}}
# - /health now reports export status and counters
# - /metrics exposes export counters
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
STARTED = time.time()
def now_utc_iso() -> str:
return datetime.now(timezone.utc).isoformat()
# ---------- Export config & helpers ----------
EXPORT_ENABLED = False
EXPORT_DIR = "exports"
EXPORT_WRITES = 0
EXPORT_ERRORS = 0
EXPORT_OK = True
EXPORT_NOTE = ""
def _ensure_export_dir():
global EXPORT_OK, EXPORT_NOTE
try:
os.makedirs(EXPORT_DIR, exist_ok=True)
# writability test
testp = os.path.join(EXPORT_DIR, "._writetest")
with open(testp, "w", encoding="utf-8") as f:
f.write("ok\n")
os.remove(testp)
EXPORT_OK = True
EXPORT_NOTE = "ready"
except Exception as e:
EXPORT_OK = False
EXPORT_NOTE = f"unwritable: {e}"
def write_jsonl(plugin: str, data: dict):
global EXPORT_WRITES, EXPORT_ERRORS, EXPORT_OK
if not EXPORT_ENABLED:
return
if not EXPORT_OK:
return
try:
row = {"plugin": plugin, "_ts": now_utc_iso(), "data": data}
path = os.path.join(EXPORT_DIR, f"{plugin}.jsonl")
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
EXPORT_WRITES += 1
except Exception as e:
EXPORT_ERRORS += 1
# ---------- Minimal plugin framework ----------
class Plugin:
NAME = "plugin"
def __init__(self, **kwargs): pass
def read(self):
"""Return a JSON-serializable dict (read-only)."""
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):
# Requires Termux:API 'termux-battery-status' command; otherwise returns unavailable
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):
# Try to gather basic IPs using `ip addr` (Termux) or `getprop` as fallback
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)}
# Fallback to Android getprop for basic hints
if have_cmd("getprop"):
try:
wifi_ip = subprocess.check_output(["getprop", "dhcp.wlan0.ipaddress"], timeout=2).decode().strip()
data = {"available": True, "wifi_ip": wifi_ip}
return data
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):
self.plugins = {cls.NAME: cls() for cls in BUILTIN_PLUGINS}
self.latest = {}
self.lock = threading.Lock()
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()
entry = {"plugin": name, "_ts": now_utc_iso(), "data": res}
with self.lock:
self.latest[name] = entry
write_jsonl(name, res) # step one: export on read
return entry
def read_all(self):
out = {}
for n in self.names():
out[n] = self.read_one(n) # each call exports if enabled
return out
def state(self):
with self.lock:
return {"plugins": self.names(), "latest": dict(self.latest)}
REG = Registry()
# ---------- Optional background poller ----------
class Poller(threading.Thread):
def __init__(self, interval_sec: int):
super().__init__(daemon=True); self.interval = max(0, int(interval_sec)); self.stop_evt = threading.Event()
def run(self):
if self.interval <= 0: return
while not self.stop_evt.is_set():
try:
REG.read_all() # exports included
except Exception:
pass
self.stop_evt.wait(self.interval)
def stop(self): self.stop_evt.set()
# ---------- Mobile UI (root interface) ----------
MOBILE_UI = """<!doctype html>
<html><head><meta charset="utf-8"/><meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>SolveForce Phone — Step 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 — Step One (JSONL)</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>
"""
# ---------- Translator (operators + currencies) ----------
MATH_OPS = {
"+": "+", "＋": "+",
"-": "-", "−": "-", "–": "-", "—": "-",
"×": "*", "∙": "*", "·": "*", "⋅": "*", "*": "*",
"÷": "/", "∕": "/", "/": "/",
"=": "=", "≈": "~", "≃": "~", "≅": "~",
"<": "<", "≤": "<=", "⩽": "<=",
">": ">", "≥": ">=", "⩾": ">=",
"^": "^", "%": "%"
}
CURRENCIES = {
"A$": "AUD", "C$": "CAD", "NZ$": "NZD", "HK$": "HKD", "S$": "SGD", "R$": "BRL", "MX$": "MXN",
"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₩": "KRW", "₹": "INR", "₽": "RUS", "₺": "TRY",
"₫": "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}
# ---------- HTTP surface ----------
MANAGER_POLL = None
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":
self._send(200, {
"status":"ok",
"time":now_utc_iso(),
"uptime_sec": int(time.time()-STARTED),
"export": {"enabled": EXPORT_ENABLED, "dir": EXPORT_DIR, "ok": EXPORT_OK, "note": EXPORT_NOTE, "writes": EXPORT_WRITES, "errors": EXPORT_ERRORS}
});
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)}",
f"solveforce_export_enabled {1 if EXPORT_ENABLED else 0}",
f"solveforce_export_writes_total {EXPORT_WRITES}",
f"solveforce_export_errors_total {EXPORT_ERRORS}"]
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")
ap.add_argument("--export-dir", default="exports", help="export directory")
args = ap.parse_args()
# configure export
global EXPORT_ENABLED, EXPORT_DIR
EXPORT_ENABLED = bool(args.export)
EXPORT_DIR = args.export_dir
if EXPORT_ENABLED:
_ensure_export_dir()
# start poller
poll_int = int(args.poll)
global MANAGER_POLL
MANAGER_POLL = Poller(interval_sec=poll_int)
MANAGER_POLL.start()
# HTTP
host, port = args.host, int(args.port)
httpd = HTTPServer((host, port), Handler)
print(f"[SolveForce] Phone One UI → http://{host}:{port}/ui")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
MANAGER_POLL.stop()
httpd.server_close()
if __name__ == "__main__":
main()
SolveForce Phone One — Step One: JSONL export toggle (read-only)
#!/usr/bin/env python3
# SolveForce Phone One — Step One: JSONL export toggle (read-only)
# MIT © 2025 Ronald Joseph Legarski, Jr. — Published by SolveForce
#
# Additions over "Phone Zero":
# - --export (enable JSONL export, ClickHouse-ready JSONEachRow)
# - --export-dir <path> (default: ./exports)
# - Each plugin read (manual or via --poll) writes one JSON row:
# {"plugin":"battery","_ts":"...","data":{...}}
# - /health now reports export status and counters
# - /metrics exposes export counters
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
STARTED = time.time()
def now_utc_iso() -> str:
return datetime.now(timezone.utc).isoformat()
# ---------- Export config & helpers ----------
EXPORT_ENABLED = False
EXPORT_DIR = "exports"
EXPORT_WRITES = 0
EXPORT_ERRORS = 0
EXPORT_OK = True
EXPORT_NOTE = ""
def _ensure_export_dir():
global EXPORT_OK, EXPORT_NOTE
try:
os.makedirs(EXPORT_DIR, exist_ok=True)
# writability test
testp = os.path.join(EXPORT_DIR, "._writetest")
with open(testp, "w", encoding="utf-8") as f:
f.write("ok\n")
os.remove(testp)
EXPORT_OK = True
EXPORT_NOTE = "ready"
except Exception as e:
EXPORT_OK = False
EXPORT_NOTE = f"unwritable: {e}"
def write_jsonl(plugin: str, data: dict):
global EXPORT_WRITES, EXPORT_ERRORS, EXPORT_OK
if not EXPORT_ENABLED:
return
if not EXPORT_OK:
return
try:
row = {"plugin": plugin, "_ts": now_utc_iso(), "data": data}
path = os.path.join(EXPORT_DIR, f"{plugin}.jsonl")
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
EXPORT_WRITES += 1
except Exception as e:
EXPORT_ERRORS += 1
# ---------- Minimal plugin framework ----------
class Plugin:
NAME = "plugin"
def __init__(self, **kwargs): pass
def read(self):
"""Return a JSON-serializable dict (read-only)."""
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):
# Requires Termux:API 'termux-battery-status' command; otherwise returns unavailable
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):
# Try to gather basic IPs using `ip addr` (Termux) or `getprop` as fallback
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)}
# Fallback to Android getprop for basic hints
if have_cmd("getprop"):
try:
wifi_ip = subprocess.check_output(["getprop", "dhcp.wlan0.ipaddress"], timeout=2).decode().strip()
data = {"available": True, "wifi_ip": wifi_ip}
return data
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):
self.plugins = {cls.NAME: cls() for cls in BUILTIN_PLUGINS}
self.latest = {}
self.lock = threading.Lock()
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()
entry = {"plugin": name, "_ts": now_utc_iso(), "data": res}
with self.lock:
self.latest[name] = entry
write_jsonl(name, res) # step one: export on read
return entry
def read_all(self):
out = {}
for n in self.names():
out[n] = self.read_one(n) # each call exports if enabled
return out
def state(self):
with self.lock:
return {"plugins": self.names(), "latest": dict(self.latest)}
REG = Registry()
# ---------- Optional background poller ----------
class Poller(threading.Thread):
def __init__(self, interval_sec: int):
super().__init__(daemon=True); self.interval = max(0, int(interval_sec)); self.stop_evt = threading.Event()
def run(self):
if self.interval <= 0: return
while not self.stop_evt.is_set():
try:
REG.read_all() # exports included
except Exception:
pass
self.stop_evt.wait(self.interval)
def stop(self): self.stop_evt.set()
# ---------- Mobile UI (root interface) ----------
MOBILE_UI = """<!doctype html>
<html><head><meta charset="utf-8"/><meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>SolveForce Phone — Step 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 — Step One (JSONL)</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>
"""
# ---------- Translator (operators + currencies) ----------
MATH_OPS = {
"+": "+", "＋": "+",
"-": "-", "−": "-", "–": "-", "—": "-",
"×": "*", "∙": "*", "·": "*", "⋅": "*", "*": "*",
"÷": "/", "∕": "/", "/": "/",
"=": "=", "≈": "~", "≃": "~", "≅": "~",
"<": "<", "≤": "<=", "⩽": "<=",
">": ">", "≥": ">=", "⩾": ">=",
"^": "^", "%": "%"
}
CURRENCIES = {
"A$": "AUD", "C$": "CAD", "NZ$": "NZD", "HK$": "HKD", "S$": "SGD", "R$": "BRL", "MX$": "MXN",
"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₩": "KRW", "₹": "INR", "₽": "RUS", "₺": "TRY",
"₫": "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}
# ---------- HTTP surface ----------
MANAGER_POLL = None
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":
self._send(200, {
"status":"ok",
"time":now_utc_iso(),
"uptime_sec": int(time.time()-STARTED),
"export": {"enabled": EXPORT_ENABLED, "dir": EXPORT_DIR, "ok": EXPORT_OK, "note": EXPORT_NOTE, "writes": EXPORT_WRITES, "errors": EXPORT_ERRORS}
});
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)}",
f"solveforce_export_enabled {1 if EXPORT_ENABLED else 0}",
f"solveforce_export_writes_total {EXPORT_WRITES}",
f"solveforce_export_errors_total {EXPORT_ERRORS}"]
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")
ap.add_argument("--export-dir", default="exports", help="export directory")
args = ap.parse_args()
# configure export
global EXPORT_ENABLED, EXPORT_DIR
EXPORT_ENABLED = bool(args.export)
EXPORT_DIR = args.export_dir
if EXPORT_ENABLED:
_ensure_export_dir()
# start poller
poll_int = int(args.poll)
global MANAGER_POLL
MANAGER_POLL = Poller(interval_sec=poll_int)
MANAGER_POLL.start()
# HTTP
host, port = args.host, int(args.port)
httpd = HTTPServer((host, port), Handler)
print(f"[SolveForce] Phone One UI → http://{host}:{port}/ui")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
MANAGER_POLL.stop()
httpd.server_close()
if __name__ == "__main__":
main()
Create “SolveForce Phone One” with JSONL export toggles and provide a downloadable file.
# Create "SolveForce Phone One" with JSONL export toggles and provide a downloadable file.
import os, json, hashlib, py_compile
from pathlib import Path
BASE = Path("/mnt/data")
fpath = BASE / "solveforce_phone_one.py"
code = r'''#!/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)
# - NEW: --export (JSONL, ClickHouse-ready JSONEachRow), --export-dir <dir>
# - Writes one line per read: {"plugin": "...", "_ts": "...", "data": {...}}
#
# Usage:
# python solveforce_phone_one.py --host 127.0.0.1 --port 8080 --poll 0
# python solveforce_phone_one.py --host 0.0.0.0 --port 8080 --poll 5 --export --export-dir /sdcard/solveforce/exports
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:
# Do not crash the server; just record the error
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 a JSON-serializable dict (read-only)."""
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):
# Requires Termux:API 'termux-battery-status' command; otherwise returns unavailable
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):
# Try to gather basic IPs using `ip addr` (Termux) or `getprop` as fallback
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)}
# Fallback to Android getprop for basic hints
if have_cmd("getprop"):
try:
wifi_ip = subprocess.check_output(["getprop", "dhcp.wlan0.ipaddress"], timeout=2).decode().strip()
data = {"available": True, "wifi_ip": wifi_ip}
return data
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
# Export safely
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)}
# Registry and poller get created in main() once EXPORTER is configured
REG: Registry = None # type: ignore
# ---------- Optional background poller ----------
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()
MANAGER_POLL: Poller = None # type: ignore
# ---------- Mobile UI (root interface) ----------
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>
:root{--b:#111;--t:#111;--c:#0a84ff}
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>
"""
# ---------- Translator (operators + currencies) ----------
MATH_OPS = {
"+": "+", "＋": "+",
"-": "-", "−": "-", "–": "-", "—": "-",
"×": "*", "∙": "*", "·": "*", "⋅": "*", "*": "*",
"÷": "/", "∕": "/", "/": "/",
"=": "=", "≈": "~", "≃": "~", "≅": "~",
"<": "<", "≤": "<=", "⩽": "<=",
">": ">", "≥": ">=", "⩾": ">=",
"^": "^", "%": "%"
}
CURRENCIES = {
"A$": "AUD", "C$": "CAD", "NZ$": "NZD", "HK$": "HKD", "S$": "SGD", "R$": "BRL", "MX$": "MXN",
"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₩": "KRW", "₹": "INR", "₽": "RUS", "₺": "TRY",
"₫": "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}
# ---------- HTTP surface ----------
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()
# Configure exporter
global EXPORTER
EXPORTER = JsonlExporter(enabled=args.export, directory=args.export_dir)
# Instantiate registry with exporter
global REG
REG = Registry(exporter=EXPORTER)
# Optional poller
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()
'''
# Write and compile
fpath.write_text(code, encoding="utf-8")
py_compile.compile(str(fpath), doraise=True)
# Hash
import hashlib
sha = hashlib.sha256(fpath.read_bytes()).hexdigest()
{"path": str(fpath), "sha256": sha}
SolveForce Phone One — phone-first gateway with JSONL exports (read-only)
#!/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)
# - NEW: --export (JSONL, ClickHouse-ready JSONEachRow), --export-dir <dir>
# - Writes one line per read: {"plugin": "...", "_ts": "...", "data": {...}}
#
# Usage:
# python solveforce_phone_one.py --host 127.0.0.1 --port 8080 --poll 0
# python solveforce_phone_one.py --host 0.0.0.0 --port 8080 --poll 5 --export --export-dir /sdcard/solveforce/exports
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:
# Do not crash the server; just record the error
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 a JSON-serializable dict (read-only)."""
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):
# Requires Termux:API 'termux-battery-status' command; otherwise returns unavailable
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):
# Try to gather basic IPs using `ip addr` (Termux) or `getprop` as fallback
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)}
# Fallback to Android getprop for basic hints
if have_cmd("getprop"):
try:
wifi_ip = subprocess.check_output(["getprop", "dhcp.wlan0.ipaddress"], timeout=2).decode().strip()
data = {"available": True, "wifi_ip": wifi_ip}
return data
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
# Export safely
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)}
# Registry and poller get created in main() once EXPORTER is configured
REG: Registry = None # type: ignore
# ---------- Optional background poller ----------
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()
MANAGER_POLL: Poller = None # type: ignore
# ---------- Mobile UI (root interface) ----------
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>
:root{--b:#111;--t:#111;--c:#0a84ff}
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>
"""
# ---------- Translator (operators + currencies) ----------
MATH_OPS = {
"+": "+", "＋": "+",
"-": "-", "−": "-", "–": "-", "—": "-",
"×": "*", "∙": "*", "·": "*", "⋅": "*", "*": "*",
"÷": "/", "∕": "/", "/": "/",
"=": "=", "≈": "~", "≃": "~", "≅": "~",
"<": "<", "≤": "<=", "⩽": "<=",
">": ">", "≥": ">=", "⩾": ">=",
"^": "^", "%": "%"
}
CURRENCIES = {
"A$": "AUD", "C$": "CAD", "NZ$": "NZD", "HK$": "HKD", "S$": "SGD", "R$": "BRL", "MX$": "MXN",
"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₩": "KRW", "₹": "INR", "₽": "RUS", "₺": "TRY",
"₫": "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}
# ---------- HTTP surface ----------
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()
# Configure exporter
global EXPORTER
EXPORTER = JsonlExporter(enabled=args.export, directory=args.export_dir)
# Instantiate registry with exporter
global REG
REG = Registry(exporter=EXPORTER)
# Optional poller
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()
