# Build "SolveForce Phone Zero" — a minimal, phone-first gateway starting from zero.
# It includes a mobile UI, plugin loader (with a Termux battery plugin auto-detected),
# and an optional background poller. Pure standard library.
import os, json, hashlib, time, textwrap, py_compile, shutil
from pathlib import Path
BASE = Path("/mnt/data")
fname = BASE / "solveforce_phone_zero.py"
code = r'''#!/usr/bin/env python3
# SolveForce Phone Zero — start-from-zero, phone-first gateway (read-only)
# MIT © 2025 Ronald Joseph Legarski, Jr. — Published by SolveForce
#
# - Single file
# - Pure standard library (Android-friendly: Termux or Pydroid)
# - Mobile UI at /ui
# - Minimal plugin system with auto-detected Termux battery plugin
# - Optional background poller via --poll N (seconds)
#
# Usage:
# python solveforce_phone_zero.py --host 127.0.0.1 --port 8080 --poll 0
# python solveforce_phone_zero.py --host 0.0.0.0 --port 8080 --poll 5 # LAN + poller
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()
# ---------- 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()
with self.lock:
self.latest[name] = {"data": res, "_ts": now_utc_iso()}
return self.latest[name]
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)}
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()
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 Zero</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 Zero — 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 ----------
MANAGER_POLL = None
class Handler(BaseHTTPRequestHandler):
server_version = "SolveForcePhoneZero/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)}); 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}"]
lines.append(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")
args = ap.parse_args()
poll_int = int(args.poll)
global MANAGER_POLL
MANAGER_POLL = Poller(interval_sec=poll_int)
MANAGER_POLL.start()
host, port = args.host, int(args.port)
httpd = HTTPServer((host, port), Handler)
print(f"[SolveForce] Phone Zero UI → http://{host}:{port}/ui")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
finally:
MANAGER_POLL.stop()
httpd.server_close()
if __name__ == "__main__":
main()
'''
# Write file
fname.write_text(code, encoding="utf-8")
# Compile to verify no syntax errors
py_compile.compile(str(fname), doraise=True)
# Compute sha256
sha256 = hashlib.sha256()
with open(fname, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256.update(chunk)
digest = sha256.hexdigest()
{"path": str(fname), "sha256": digest}