"""OHD Win11 setup UI — pick a local Topbox Pi on the LAN."""
from __future__ import annotations
import json
import socket
from typing import Callable, Optional
from extended_configparser.parser import ExtendedConfigParser
from machine_identity import read_machine_identity
from topbox_discovery import scan_local_topbox_devices
def build_topbox_info_payload(
*,
version: str,
config_path: str,
display_mode: str,
asset_number: int,
) -> dict:
config = ExtendedConfigParser()
config.read(config_path)
identity = read_machine_identity(config)
hostname = socket.gethostname()
short = hostname.split(".")[0]
return {
"version": version,
"hostname": short,
"machine_role": identity["machine_role"],
"bank_number": identity["bank_number"],
"term_number": identity["term_number"],
"asset_number": asset_number,
"display_mode": display_mode,
"topbox_host": identity["topbox_host"],
}
def register_ohd_setup_routes(
*,
get,
post,
request,
response,
config_path: str,
version: str,
web_port: Callable[[], int],
display_mode: Callable[[], str],
asset_number: Callable[[], int],
reload_config: Callable[[], None],
) -> None:
@get("/Topbox/info.json")
def topbox_info_handler():
payload = build_topbox_info_payload(
version=version,
config_path=config_path,
display_mode=display_mode(),
asset_number=asset_number(),
)
response.content_type = "application/json"
return json.dumps(payload)
@get("/ohd/api/topbox-scan")
def ohd_topbox_scan_handler():
devices, subnets = scan_local_topbox_devices(port=web_port())
response.content_type = "application/json"
return json.dumps({"devices": devices, "scanned_subnets": subnets})
@get("/ohd/setup")
def ohd_setup_page_handler():
config = ExtendedConfigParser()
config.read(config_path)
current_source = ""
current_host = ""
if config.has_option("TOPBOX", "PROGRESSIVES_SOURCE"):
current_source = config.get("TOPBOX", "PROGRESSIVES_SOURCE").strip()
identity = read_machine_identity(config)
current_host = identity["topbox_host"]
port = web_port()
return _OHD_SETUP_HTML.format(
port=port,
current_source=_html_escape(current_source),
current_host=_html_escape(current_host),
)
@post("/ohd/setup")
def ohd_setup_save_handler():
ip = str(request.forms.get("ip") or "").strip()
hostname = str(request.forms.get("hostname") or "").strip()
progressives_url = str(request.forms.get("progressives_url") or "").strip()
if not progressives_url and ip:
progressives_url = f"http://{ip}:{web_port()}/EtherGold/progressives.csv"
if not progressives_url.startswith("http"):
return _setup_result("FAIL — invalid progressives URL", ok=False)
config = ExtendedConfigParser()
config.read(config_path)
config.set("TOPBOX", "PROGRESSIVES_SOURCE", progressives_url)
if hostname:
config.set("TOPBOX", "TOPBOX_HOST", hostname)
from config_ini_persist import write_config_ini
if not write_config_ini(config, config_path):
return _setup_result(
"FAIL — could not write config.ini (check ownership; must not be root:root)",
ok=False,
)
reload_config()
return _setup_result(
f"Saved progressives_source={progressives_url}"
+ (f" topbox_host={hostname}" if hostname else "")
+ " — restart DyzineFrame if progressives do not update immediately.",
ok=True,
)
def _html_escape(text: str) -> str:
return (
text.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
def _setup_result(message: str, *, ok: bool) -> str:
color = "#1a7f37" if ok else "#cf222e"
return f"""
OHD Setup
{_html_escape(message)}
Back to Topbox picker
"""
_OHD_SETUP_HTML = """
OHD Topbox Setup
OHD — Select Topbox Pi
Scans WiFi/LAN for DyzineFrame devices whose hostname starts with topbox.
Master units (role M or hostname ending in M) are listed first.
Current: {current_host or "(none)"}
{current_source or "(progressives_source unset)"}
"""