"""Product display server — shows current product on a tablet next to the robot.
Serves a minimal page at http://robot-ip:7860 that auto-updates
when the robot recommends a product. Just open it on a tablet.
"""
import asyncio
import json
import logging
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
logger = logging.getLogger(__name__)
_CONFIG_KEYS = {
"ZG_API_KEY",
"ZG_BASE_URL",
"ZG_MODEL",
"ZG_WHISPER_MODEL",
"ZG_CONSUMER_EMAIL",
"STORE_NAME",
"KOKORO_VOICE",
"CHAIN_RPC",
"CHAIN_PRIVATE_KEY",
"CHAIN_ID",
"TG_BOT_TOKEN",
"TG_CHAT_ID",
}
def _config_status() -> dict:
from . import config
return {
"env_file": str(config.env_file_path()),
"zg_api_key": bool(config.ZG_API_KEY),
"zg_consumer_email": bool(config.ZG_CONSUMER_EMAIL),
"chain_private_key": bool(config.CHAIN_PRIVATE_KEY),
}
def _save_config(payload: dict) -> dict:
from . import config
values = {}
for key in _CONFIG_KEYS:
value = payload.get(key)
if value is None:
continue
value = str(value).strip()
if value:
values[key] = value
if not values:
raise ValueError("No valid StoreOS config keys provided")
path = config.write_env_file(values)
config.apply_runtime_config(values)
return {
"saved": sorted(values.keys()),
"env_file": str(path),
"status": _config_status(),
}
_current = {
"mode": "idle",
"product": None,
"message": "Waiting for customers...",
"store_name": "The Good Store",
}
_listeners: list[asyncio.Queue] = []
def set_store_name(name: str):
_current["store_name"] = name
def show_product(product: dict, message: str = ""):
_current["mode"] = "product"
_current["product"] = product
_current["message"] = message
_notify()
def show_idle(message: str = "Waiting for customers..."):
_current["mode"] = "idle"
_current["product"] = None
_current["message"] = message
_notify()
def show_listening():
_current["mode"] = "listening"
_current["message"] = "Listening..."
_notify()
def show_thinking():
_current["mode"] = "thinking"
_current["message"] = "Thinking..."
_notify()
def _notify():
for q in _listeners:
try:
q.put_nowait(True)
except asyncio.QueueFull:
pass
PAGE = """
StoreOS Display
THE GOOD STORE
"""
class DisplayHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/api/state":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(_current).encode())
elif self.path == "/api/config/status":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(_config_status()).encode())
elif self.path == "/api/log":
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
try:
with open("/tmp/storeos.log", "r") as f:
self.wfile.write(f.read().encode())
except:
self.wfile.write(b"No log file yet")
else:
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(PAGE.encode())
def do_POST(self):
if self.path != "/api/config":
self.send_response(404)
self.end_headers()
return
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode() or "{}")
result = _save_config(payload)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(result).encode())
except Exception as e:
self.send_response(400)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps({"error": str(e)}).encode())
def log_message(self, *args):
pass
def start_server(port: int = 7860):
server = HTTPServer(("0.0.0.0", port), DisplayHandler)
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
logger.info(f"Display server at http://0.0.0.0:{port}")