Seth
5d48ad0bfe
- /api/repeaters: bad ?limit= falls back to 25 instead of crashing the handler
- repeaters.load(): tolerate a corrupted repeaters.json (JSONDecodeError -> [])
- tune_repeater/do_POST: catch TypeError so a null field yields {ok:false} not a reset
133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Web control panel server for the ICOM ID-5100.
|
|
|
|
Holds one CI-V serial connection (so don't run cli.py at the same time),
|
|
serves the instrument-panel UI, and exposes a tiny JSON API on top of radio.py.
|
|
|
|
python3 server.py [--port 8550] [--serial /dev/ttyUSB0] [--baud 9600]
|
|
|
|
Then browse to http://<host>:8550/ (binds 0.0.0.0 -> reachable from phone).
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
import repeaters
|
|
from radio import Radio, CIVError
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
PANEL = os.path.join(HERE, "panel.html")
|
|
REPEATERS_JSON = os.path.join(HERE, "repeaters.json")
|
|
|
|
radio = None
|
|
lock = threading.Lock()
|
|
|
|
|
|
def read_status():
|
|
with lock:
|
|
freq = radio.get_frequency()
|
|
mode, code = radio.get_mode()
|
|
return {
|
|
"ok": True,
|
|
"freq_mhz": round(freq / 1e6, 5),
|
|
"mode": mode,
|
|
"mode_code": code,
|
|
"volume": radio.get_volume(),
|
|
"squelch_level": radio.get_squelch_level(),
|
|
"squelch_open": radio.squelch_open(),
|
|
"smeter": radio.get_smeter(),
|
|
}
|
|
|
|
|
|
def apply_control(path, body):
|
|
with lock:
|
|
if path == "/api/freq":
|
|
radio.set_frequency(int(round(float(body["mhz"]) * 1e6)))
|
|
elif path == "/api/mode":
|
|
radio.set_mode(str(body["mode"]))
|
|
elif path == "/api/vol":
|
|
radio.set_volume(int(body["value"]))
|
|
elif path == "/api/sql":
|
|
radio.set_squelch_level(int(body["value"]))
|
|
elif path == "/api/tune-repeater":
|
|
steps, errors = radio.tune_repeater(body)
|
|
return {"ok": bool(steps.get("freq")), "steps": steps, "errors": errors}
|
|
else:
|
|
raise CIVError("unknown control")
|
|
return {"ok": True}
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass # quiet
|
|
|
|
def _json(self, obj, code=200):
|
|
payload = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def do_GET(self):
|
|
if self.path == "/" or self.path == "/index.html":
|
|
with open(PANEL, "rb") as f:
|
|
body = f.read()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
elif self.path == "/api/status":
|
|
try:
|
|
self._json(read_status())
|
|
except (CIVError, OSError) as e:
|
|
self._json({"ok": False, "error": str(e)}, 200)
|
|
elif self.path.startswith("/api/repeaters"):
|
|
from urllib.parse import urlparse, parse_qs
|
|
qs = parse_qs(urlparse(self.path).query)
|
|
q = qs.get("q", [""])[0]
|
|
try:
|
|
limit = int(qs.get("limit", ["25"])[0])
|
|
except ValueError:
|
|
limit = 25
|
|
recs = repeaters.search(repeaters.load(REPEATERS_JSON), q, limit)
|
|
self._json({"ok": True, "repeaters": recs})
|
|
else:
|
|
self._json({"ok": False, "error": "not found"}, 404)
|
|
|
|
def do_POST(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
try:
|
|
body = json.loads(self.rfile.read(length) or b"{}")
|
|
self._json(apply_control(self.path, body))
|
|
except (CIVError, OSError, ValueError, TypeError, KeyError) as e:
|
|
self._json({"ok": False, "error": str(e)}, 200)
|
|
|
|
|
|
def main():
|
|
global radio
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--port", type=int, default=8550)
|
|
ap.add_argument("--serial", default="/dev/ttyUSB0")
|
|
ap.add_argument("--baud", type=int, default=9600)
|
|
args = ap.parse_args()
|
|
|
|
radio = Radio(args.serial, args.baud).open()
|
|
httpd = ThreadingHTTPServer(("0.0.0.0", args.port), Handler)
|
|
print(f"ID-5100 panel on http://0.0.0.0:{args.port}/ "
|
|
f"(serial {args.serial} @ {args.baud})")
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
radio.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|