-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdesktop_app.py
More file actions
161 lines (122 loc) · 4.64 KB
/
Copy pathdesktop_app.py
File metadata and controls
161 lines (122 loc) · 4.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
import argparse
import sys
import threading
import time
import urllib.request
import webbrowser
from multiprocessing import freeze_support
from typing import Optional, Tuple
from werkzeug.serving import make_server
from server import app, bridge
APP_TITLE = "Ground Control System"
DEFAULT_WINDOW_WIDTH = 1600
DEFAULT_WINDOW_HEIGHT = 920
MIN_WINDOW_WIDTH = 1080
MIN_WINDOW_HEIGHT = 680
WINDOW_SCREEN_MARGIN = 80
class LocalWebServer:
def __init__(self, host: str, port: int) -> None:
self._server = make_server(host, port, app, threaded=True)
self.host = host
self.port = int(self._server.server_port)
self._thread: Optional[threading.Thread] = None
@property
def base_url(self) -> str:
connect_host = "127.0.0.1" if self.host in {"", "0.0.0.0"} else self.host
return f"http://{connect_host}:{self.port}"
def start(self) -> None:
self._thread = threading.Thread(target=self._server.serve_forever, name="gcs-web-server", daemon=True)
self._thread.start()
wait_for_server(f"{self.base_url}/api/status")
def stop(self) -> None:
self._server.shutdown()
if self._thread is not None:
self._thread.join(timeout=3)
def wait_for_server(url: str, timeout_seconds: float = 8.0) -> None:
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
try:
with urllib.request.urlopen(url, timeout=0.5) as response:
if response.status < 500:
return
except OSError:
time.sleep(0.1)
raise RuntimeError(f"Local web server did not start: {url}")
def connect_serial_on_startup(port: str, baud: int) -> None:
if not port:
return
try:
bridge.connect(port, baud)
except Exception as exc:
print(f"Serial autoconnect failed: {exc}", file=sys.stderr)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Desktop Ground Control System")
parser.add_argument("--port", default="", help="Optional serial port to connect on startup, such as COM5")
parser.add_argument("--baud", type=int, default=115200)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--web-port", type=int, default=0, help="Local web port. Use 0 to choose an available port.")
parser.add_argument("--browser", action="store_true", help="Open in the default browser instead of WebView.")
parser.add_argument("--debug-webview", action="store_true", help="Enable pywebview debug mode.")
return parser.parse_args()
def run_browser_mode(url: str) -> None:
webbrowser.open(url)
print(f"{APP_TITLE} is running at {url}")
print("Press Ctrl+C to stop.")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
def run_webview_mode(url: str, debug: bool) -> None:
try:
import webview
except ImportError as exc:
raise RuntimeError(
"pywebview is required for desktop mode. Install requirements-app.txt or run with --browser."
) from exc
window_width, window_height = window_size_for_display()
webview.create_window(
APP_TITLE,
url,
width=window_width,
height=window_height,
min_size=(MIN_WINDOW_WIDTH, MIN_WINDOW_HEIGHT),
)
webview.start(debug=debug)
def window_size_for_display() -> Tuple[int, int]:
screen_width, screen_height = display_size()
if screen_width is None or screen_height is None:
return DEFAULT_WINDOW_WIDTH, DEFAULT_WINDOW_HEIGHT
available_width = max(MIN_WINDOW_WIDTH, screen_width - WINDOW_SCREEN_MARGIN)
available_height = max(MIN_WINDOW_HEIGHT, screen_height - WINDOW_SCREEN_MARGIN)
return (
min(DEFAULT_WINDOW_WIDTH, available_width),
min(DEFAULT_WINDOW_HEIGHT, available_height),
)
def display_size() -> Tuple[Optional[int], Optional[int]]:
try:
import ctypes
user32 = ctypes.windll.user32
width = int(user32.GetSystemMetrics(0))
height = int(user32.GetSystemMetrics(1))
if width > 0 and height > 0:
return width, height
except Exception:
pass
return None, None
def main() -> None:
freeze_support()
args = parse_args()
web_server = LocalWebServer(args.host, args.web_port)
try:
web_server.start()
connect_serial_on_startup(args.port, args.baud)
if args.browser:
run_browser_mode(web_server.base_url)
else:
run_webview_mode(web_server.base_url, args.debug_webview)
finally:
bridge.disconnect()
web_server.stop()
if __name__ == "__main__":
main()