pacai.ui.web
1import base64 2import errno 3import http 4import http.server 5import io 6import logging 7import mimetypes 8import os 9import re 10import socket 11import threading 12import time 13import typing 14import urllib.parse 15import webbrowser 16 17import PIL.Image 18import edq.util.dirent 19import edq.util.json 20 21import pacai.core.action 22import pacai.core.board 23import pacai.core.gamestate 24import pacai.core.ui 25 26THIS_DIR: str = os.path.join(os.path.dirname(os.path.realpath(__file__))) 27STATIC_DIR: str = os.path.join(THIS_DIR, '..', 'resources', 'webui') 28 29START_PORT: int = 30000 30END_PORT: int = 40000 31 32INITIAL_SLEEP_TIME_SEC: float = 0.05 33SOCKET_SLEEP_TIME_SECS: float = 0.10 34SERVER_WAIT_TIME_SECS: float = 0.25 35REAP_TIME_SECS: float = 0.25 36COMPLETE_WAIT_TIME_SECS: float = 0.15 37 38SERVER_POLL_INTERVAL_SECS: float = 0.1 39 40RequestHandlerResult: typing.TypeAlias = tuple[dict | str | bytes | None, int | None, dict | None] | None 41 42class WebUserInputDevice(pacai.core.ui.UserInputDevice): 43 """ 44 A user input device that gets input from the same web page used by a WebUI. 45 """ 46 47 def __init__(self, 48 char_mapping: dict[str, pacai.core.action.Action] | None = None, 49 **kwargs: typing.Any) -> None: 50 self._actions: list[pacai.core.action.Action] = [] 51 """ The actions stored from the web page. """ 52 53 self._lock: threading.Lock = threading.Lock() 54 """ A lock to protect the user actions. """ 55 56 if (char_mapping is None): 57 char_mapping = pacai.core.ui.DUAL_CHAR_MAPPING 58 59 self._char_mapping: dict[str, pacai.core.action.Action] = char_mapping 60 """ Map characters to actions. """ 61 62 def add_keys(self, keys: list[str]) -> None: 63 """ Load key inputs from the UI into this device. """ 64 65 actions = [] 66 for key in keys: 67 if (key in self._char_mapping): 68 actions.append(self._char_mapping[key]) 69 70 with self._lock: 71 self._actions += actions 72 73 def get_inputs(self) -> list[pacai.core.action.Action]: 74 with self._lock: 75 actions = self._actions 76 self._actions = [] 77 return actions 78 79class HTTPHandler(http.server.BaseHTTPRequestHandler): 80 """ Handle HTTP requests for the web UI. """ 81 82 _lock: threading.Lock = threading.Lock() 83 """ A lock to protect the data sent to the web page. """ 84 85 _user_input_device: WebUserInputDevice | None = None 86 """ Put all user input into this device. """ 87 88 _fps: int | None = None 89 """ The FPS the web UI should run at. """ 90 91 _state: pacai.core.gamestate.GameState | None = None 92 """ The current game state. """ 93 94 _image_url: str | None = None 95 """ The data URL for the current image. """ 96 97 @classmethod 98 def ui_setup(cls, fps: int, user_input_device: WebUserInputDevice) -> None: 99 """ Initialize this handler with information from the UI. """ 100 101 cls._fps = fps 102 cls._user_input_device = user_input_device 103 104 @classmethod 105 def set_data(cls, state: pacai.core.gamestate.GameState, image: PIL.Image.Image) -> None: 106 """ Set the data passed back to the web page. """ 107 108 buffer = io.BytesIO() 109 image.save(buffer, format = 'png') 110 data_64 = base64.b64encode(buffer.getvalue()).decode(edq.util.dirent.DEFAULT_ENCODING) 111 data_url = f"data:image/png;base64,{data_64}" 112 113 with cls._lock: 114 cls._state = state.copy() 115 cls._image_url = data_url 116 117 @classmethod 118 def get_data(cls) -> tuple[pacai.core.gamestate.GameState | None, str | None]: 119 """ Get the data passed back to the web page. """ 120 121 with cls._lock: 122 return cls._state, cls._image_url 123 124 def log_message(self, *args: typing.Any) -> None: 125 """ 126 Reduce the logging noise. 127 """ 128 129 def handle(self) -> None: 130 """ 131 Override handle() to ignore dropped connections. 132 """ 133 134 try: 135 http.server.BaseHTTPRequestHandler.handle(self) 136 except BrokenPipeError: 137 logging.info("Connection closed on the client side.") 138 139 def do_POST(self) -> None: # pylint: disable=invalid-name 140 """ Handle POST requests. """ 141 142 self._handle_request(self._get_post_data) 143 144 def do_GET(self) -> None: # pylint: disable=invalid-name 145 """ Handle GET requests. """ 146 147 self._handle_request(self._get_get_data) 148 149 def _handle_request(self, data_handler: typing.Callable) -> None: 150 logging.trace("Serving: '%s'.", self.path) # type: ignore[attr-defined] # pylint: disable=no-member 151 152 code: int = http.HTTPStatus.OK 153 headers: dict[str, typing.Any] = {} 154 155 result = None 156 try: 157 data = data_handler() 158 result = self._route(self.path, data) 159 except Exception as ex: 160 # An error occured during data handling (routing captures their own errors). 161 logging.debug("Error handling '%s'.", self.path, exc_info = ex) 162 result = (str(ex), http.HTTPStatus.BAD_REQUEST, None) 163 164 if (result is None): 165 # All handling was done internally, the response is complete. 166 return 167 168 # A standard response structure was returned, continue processing. 169 payload, response_code, response_headers = result 170 171 if (isinstance(payload, dict)): 172 payload = edq.util.json.dumps(payload) 173 headers['Content-Type'] = 'application/json' 174 175 if (isinstance(payload, str)): 176 payload = payload.encode(edq.util.dirent.DEFAULT_ENCODING) 177 178 if (payload is not None): 179 headers['Content-Length'] = len(payload) 180 181 if (response_headers is not None): 182 for key, value in response_headers.items(): 183 headers[key] = value 184 185 if (response_code is not None): 186 code = response_code 187 188 self.send_response(code) 189 190 for (key, value) in headers.items(): 191 self.send_header(key, value) 192 self.end_headers() 193 194 if (payload is not None): 195 self.wfile.write(payload) 196 197 def _route(self, path: str, params: dict[str, typing.Any]) -> RequestHandlerResult: 198 path = path.strip() 199 200 target = _handler_not_found 201 for (regex, handler_func) in ROUTES: 202 if (re.search(regex, path) is not None): 203 target = handler_func 204 break 205 206 try: 207 return target(self, path, params) 208 except Exception as ex: 209 logging.error("Error on path '%s', handler '%s'.", path, str(target), exc_info = ex) 210 return str(ex), http.HTTPStatus.INTERNAL_SERVER_ERROR, None 211 212 def _get_get_data(self) -> dict[str, typing.Any]: 213 path = self.path.strip().rstrip('/') 214 url = urllib.parse.urlparse(path) 215 216 raw_params = urllib.parse.parse_qs(url.query) 217 params: dict[str, typing.Any] = {} 218 219 for (key, values) in raw_params.items(): 220 if ((len(values) == 0) or (values[0] == '')): 221 continue 222 223 if (len(values) == 1): 224 params[key] = values[0] 225 else: 226 params[key] = values 227 228 return params 229 230 def _get_post_data(self) -> dict[str, typing.Any]: 231 length = int(self.headers['Content-Length']) 232 payload = self.rfile.read(length).decode(edq.util.dirent.DEFAULT_ENCODING) 233 234 try: 235 request = edq.util.json.loads(payload) 236 except Exception as ex: 237 raise ValueError("Payload is not valid json.") from ex 238 239 return request # type: ignore[no-any-return] 240 241class WebUI(pacai.core.ui.UI): 242 """ 243 A UI that starts a web server and launches a brower window to serve a UI. 244 The web server will accept requests that contains user inputs, 245 and respond with the current game state and a visual representation of the game (a base64 encoded png). 246 """ 247 248 def __init__(self, 249 **kwargs: typing.Any) -> None: 250 input_device = WebUserInputDevice(**kwargs) 251 super().__init__(user_input_device = input_device, **kwargs) 252 253 self._port: int = -1 254 """ 255 The port to start the web server on. 256 The first open port in [START_PORT, END_PORT] will be used. 257 """ 258 259 self._startup_barrier: threading.Barrier = threading.Barrier(2) 260 """ Use a threading barrier to wait for the server thread to start. """ 261 262 self._server_thread: threading.Thread | None = None 263 """ The thread the server will be run on. """ 264 265 self._server: http.server.HTTPServer | None = None 266 """ The HTTP server. """ 267 268 def game_start(self, 269 initial_state: pacai.core.gamestate.GameState, 270 board_highlights: list[pacai.core.board.Highlight] | None = None, 271 **kwargs: typing.Any) -> None: 272 self._start_server() 273 274 super().game_start(initial_state, board_highlights = board_highlights) 275 276 self._launch_page(initial_state) 277 278 def game_complete(self, 279 final_state: pacai.core.gamestate.GameState, 280 board_highlights: list[pacai.core.board.Highlight] | None = None, 281 ) -> None: 282 super().game_complete(final_state, board_highlights = board_highlights) 283 284 # Wait for the UI to make a final request. 285 time.sleep(COMPLETE_WAIT_TIME_SECS) 286 287 self._stop_server() 288 289 def draw(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> None: 290 image = self.draw_image(state) 291 HTTPHandler.set_data(state, image) 292 293 def _start_server(self) -> None: 294 """ Start the HTTP server on another thread. """ 295 296 # Fetch the port. 297 self._port = _find_open_port() 298 299 # Setup the barrier to wait for the server thread to start. 300 self._startup_barrier.reset() 301 302 # Create, but don't start the server. 303 self._server = http.server.ThreadingHTTPServer(('', self._port), HTTPHandler) 304 305 # Setup the handler. 306 HTTPHandler.ui_setup(self._fps, typing.cast(WebUserInputDevice, self._user_input_device)) 307 308 self._server_thread = threading.Thread(target = _run_server, args = (self._server, self._startup_barrier)) 309 self._server_thread.start() 310 311 # Wait for the server to startup. 312 self._startup_barrier.wait() 313 time.sleep(INITIAL_SLEEP_TIME_SEC) 314 315 def _stop_server(self) -> None: 316 """ Stop the HTTP server and thread. """ 317 318 if ((self._server is None) or (self._server_thread is None)): 319 return 320 321 self._server.shutdown() 322 time.sleep(SERVER_WAIT_TIME_SECS) 323 324 if (self._server_thread.is_alive()): 325 self._server_thread.join(REAP_TIME_SECS) 326 327 self._server = None 328 self._server_thread = None 329 330 def _launch_page(self, initial_state: pacai.core.gamestate.GameState) -> None: 331 """ Open the browser window to the web UI page. """ 332 333 image = self.draw_image(initial_state) 334 HTTPHandler.set_data(initial_state, image) 335 336 logging.info("Starting web UI on port %d.", self._port) 337 logging.info("If a browser window does not open, you may use the following link:") 338 logging.info("http://127.0.0.1:%d", self._port) 339 340 webbrowser.open(f"http://127.0.0.1:{self._port}/static/index.html") 341 342def _find_open_port() -> int: 343 """ Go through [START_PORT, END_PORT] looking for open ports. """ 344 345 for port in range(START_PORT, END_PORT + 1): 346 try: 347 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 348 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 349 sock.bind(('127.0.0.1', port)) 350 351 # Explicitly close the port and wait a short amount of time for the port to clear. 352 # This should not be required because of the socket option above, 353 # but the cost is small. 354 sock.close() 355 time.sleep(SOCKET_SLEEP_TIME_SECS) 356 357 return port 358 except socket.error as ex: 359 sock.close() 360 361 if (ex.errno == errno.EADDRINUSE): 362 continue 363 364 # Unknown error. 365 raise ex 366 367 raise ValueError(f"Could not find open port in [{START_PORT}, {END_PORT}].") 368 369def _run_server(server: http.server.HTTPServer, startup_barrier: threading.Barrier) -> None: 370 """ Run the http server on this curent thread. """ 371 372 startup_barrier.wait() 373 server.serve_forever(poll_interval = SERVER_POLL_INTERVAL_SECS) 374 server.server_close() 375 376@typing.runtime_checkable 377class RequestHandler(typing.Protocol): 378 """ Functions that can be used as HTTP request handlers by HTTPHandler. """ 379 380 def __call__(self, handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 381 ... 382 383def _handler_not_found(handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 384 """ Handle a 404. """ 385 386 return (f"404 route not found: '{path}'.", http.HTTPStatus.NOT_FOUND, None) 387 388def _handler_redirect(target: str) -> RequestHandler: 389 """ Get a handler that redirects to the specific target. """ 390 391 def handler_func(handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 392 return (None, http.HTTPStatus.MOVED_PERMANENTLY, {'Location': target}) 393 394 return handler_func 395 396def _handler_static(handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 397 """ Get a static (bundled) file. """ 398 399 # Note that the path is currently a URL path, and therefore separated with slashes. 400 parts = path.strip().lstrip('/').split('/') 401 402 # Remove the static prefix. 403 if (parts[0] == 'static'): 404 parts.pop(0) 405 406 static_path = os.path.join(STATIC_DIR, *parts) 407 logging.trace("Serving static path: '%s'.", static_path) # type: ignore[attr-defined] # pylint: disable=no-member 408 409 if (not os.path.isfile(static_path)): 410 return (f"404 static path not found '{path}'.", http.HTTPStatus.NOT_FOUND, None) 411 412 data = edq.util.dirent.read_file_bytes(static_path) 413 414 code = http.HTTPStatus.OK 415 headers = {} 416 417 mime_info = mimetypes.guess_type(path) 418 if (mime_info is not None): 419 headers['Content-Type'] = mime_info[0] 420 421 return data, code, headers 422 423def _handler_init(handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 424 """ Handle a request by the browser to initialize. """ 425 426 data = { 427 'title': 'pacai', 428 'fps': handler._fps, 429 } 430 431 return (data, None, None) 432 433def _handler_update(handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 434 """ Handle a request by the browser for more data. """ 435 436 state, image_url = handler.get_data() 437 data = { 438 'state': state, 439 'image_url': image_url, 440 } 441 442 keys = params.get('keys', []) 443 if (handler._user_input_device is not None): 444 handler._user_input_device.add_keys(keys) 445 446 return (data, None, None) 447 448ROUTES: list[tuple[str, RequestHandler]] = [ 449 (r'^/$', _handler_redirect('/static/index.html')), 450 (r'^/index.html$', _handler_redirect('/static/index.html')), 451 (r'^/static$', _handler_redirect('/static/index.html')), 452 (r'^/static/$', _handler_redirect('/static/index.html')), 453 454 (r'^/favicon.ico$', _handler_redirect('/static/favicon.ico')), 455 456 (r'^/static/', _handler_static), 457 458 (r'^/api/init$', _handler_init), 459 (r'^/api/update$', _handler_update), 460]
43class WebUserInputDevice(pacai.core.ui.UserInputDevice): 44 """ 45 A user input device that gets input from the same web page used by a WebUI. 46 """ 47 48 def __init__(self, 49 char_mapping: dict[str, pacai.core.action.Action] | None = None, 50 **kwargs: typing.Any) -> None: 51 self._actions: list[pacai.core.action.Action] = [] 52 """ The actions stored from the web page. """ 53 54 self._lock: threading.Lock = threading.Lock() 55 """ A lock to protect the user actions. """ 56 57 if (char_mapping is None): 58 char_mapping = pacai.core.ui.DUAL_CHAR_MAPPING 59 60 self._char_mapping: dict[str, pacai.core.action.Action] = char_mapping 61 """ Map characters to actions. """ 62 63 def add_keys(self, keys: list[str]) -> None: 64 """ Load key inputs from the UI into this device. """ 65 66 actions = [] 67 for key in keys: 68 if (key in self._char_mapping): 69 actions.append(self._char_mapping[key]) 70 71 with self._lock: 72 self._actions += actions 73 74 def get_inputs(self) -> list[pacai.core.action.Action]: 75 with self._lock: 76 actions = self._actions 77 self._actions = [] 78 return actions
A user input device that gets input from the same web page used by a WebUI.
48 def __init__(self, 49 char_mapping: dict[str, pacai.core.action.Action] | None = None, 50 **kwargs: typing.Any) -> None: 51 self._actions: list[pacai.core.action.Action] = [] 52 """ The actions stored from the web page. """ 53 54 self._lock: threading.Lock = threading.Lock() 55 """ A lock to protect the user actions. """ 56 57 if (char_mapping is None): 58 char_mapping = pacai.core.ui.DUAL_CHAR_MAPPING 59 60 self._char_mapping: dict[str, pacai.core.action.Action] = char_mapping 61 """ Map characters to actions. """
63 def add_keys(self, keys: list[str]) -> None: 64 """ Load key inputs from the UI into this device. """ 65 66 actions = [] 67 for key in keys: 68 if (key in self._char_mapping): 69 actions.append(self._char_mapping[key]) 70 71 with self._lock: 72 self._actions += actions
Load key inputs from the UI into this device.
74 def get_inputs(self) -> list[pacai.core.action.Action]: 75 with self._lock: 76 actions = self._actions 77 self._actions = [] 78 return actions
Get any inputs that have occurred since the last call to this method. This method is responsible for not returning the same input instance in subsequent calls. The last input in the returned list should be the most recent input.
Inherited Members
80class HTTPHandler(http.server.BaseHTTPRequestHandler): 81 """ Handle HTTP requests for the web UI. """ 82 83 _lock: threading.Lock = threading.Lock() 84 """ A lock to protect the data sent to the web page. """ 85 86 _user_input_device: WebUserInputDevice | None = None 87 """ Put all user input into this device. """ 88 89 _fps: int | None = None 90 """ The FPS the web UI should run at. """ 91 92 _state: pacai.core.gamestate.GameState | None = None 93 """ The current game state. """ 94 95 _image_url: str | None = None 96 """ The data URL for the current image. """ 97 98 @classmethod 99 def ui_setup(cls, fps: int, user_input_device: WebUserInputDevice) -> None: 100 """ Initialize this handler with information from the UI. """ 101 102 cls._fps = fps 103 cls._user_input_device = user_input_device 104 105 @classmethod 106 def set_data(cls, state: pacai.core.gamestate.GameState, image: PIL.Image.Image) -> None: 107 """ Set the data passed back to the web page. """ 108 109 buffer = io.BytesIO() 110 image.save(buffer, format = 'png') 111 data_64 = base64.b64encode(buffer.getvalue()).decode(edq.util.dirent.DEFAULT_ENCODING) 112 data_url = f"data:image/png;base64,{data_64}" 113 114 with cls._lock: 115 cls._state = state.copy() 116 cls._image_url = data_url 117 118 @classmethod 119 def get_data(cls) -> tuple[pacai.core.gamestate.GameState | None, str | None]: 120 """ Get the data passed back to the web page. """ 121 122 with cls._lock: 123 return cls._state, cls._image_url 124 125 def log_message(self, *args: typing.Any) -> None: 126 """ 127 Reduce the logging noise. 128 """ 129 130 def handle(self) -> None: 131 """ 132 Override handle() to ignore dropped connections. 133 """ 134 135 try: 136 http.server.BaseHTTPRequestHandler.handle(self) 137 except BrokenPipeError: 138 logging.info("Connection closed on the client side.") 139 140 def do_POST(self) -> None: # pylint: disable=invalid-name 141 """ Handle POST requests. """ 142 143 self._handle_request(self._get_post_data) 144 145 def do_GET(self) -> None: # pylint: disable=invalid-name 146 """ Handle GET requests. """ 147 148 self._handle_request(self._get_get_data) 149 150 def _handle_request(self, data_handler: typing.Callable) -> None: 151 logging.trace("Serving: '%s'.", self.path) # type: ignore[attr-defined] # pylint: disable=no-member 152 153 code: int = http.HTTPStatus.OK 154 headers: dict[str, typing.Any] = {} 155 156 result = None 157 try: 158 data = data_handler() 159 result = self._route(self.path, data) 160 except Exception as ex: 161 # An error occured during data handling (routing captures their own errors). 162 logging.debug("Error handling '%s'.", self.path, exc_info = ex) 163 result = (str(ex), http.HTTPStatus.BAD_REQUEST, None) 164 165 if (result is None): 166 # All handling was done internally, the response is complete. 167 return 168 169 # A standard response structure was returned, continue processing. 170 payload, response_code, response_headers = result 171 172 if (isinstance(payload, dict)): 173 payload = edq.util.json.dumps(payload) 174 headers['Content-Type'] = 'application/json' 175 176 if (isinstance(payload, str)): 177 payload = payload.encode(edq.util.dirent.DEFAULT_ENCODING) 178 179 if (payload is not None): 180 headers['Content-Length'] = len(payload) 181 182 if (response_headers is not None): 183 for key, value in response_headers.items(): 184 headers[key] = value 185 186 if (response_code is not None): 187 code = response_code 188 189 self.send_response(code) 190 191 for (key, value) in headers.items(): 192 self.send_header(key, value) 193 self.end_headers() 194 195 if (payload is not None): 196 self.wfile.write(payload) 197 198 def _route(self, path: str, params: dict[str, typing.Any]) -> RequestHandlerResult: 199 path = path.strip() 200 201 target = _handler_not_found 202 for (regex, handler_func) in ROUTES: 203 if (re.search(regex, path) is not None): 204 target = handler_func 205 break 206 207 try: 208 return target(self, path, params) 209 except Exception as ex: 210 logging.error("Error on path '%s', handler '%s'.", path, str(target), exc_info = ex) 211 return str(ex), http.HTTPStatus.INTERNAL_SERVER_ERROR, None 212 213 def _get_get_data(self) -> dict[str, typing.Any]: 214 path = self.path.strip().rstrip('/') 215 url = urllib.parse.urlparse(path) 216 217 raw_params = urllib.parse.parse_qs(url.query) 218 params: dict[str, typing.Any] = {} 219 220 for (key, values) in raw_params.items(): 221 if ((len(values) == 0) or (values[0] == '')): 222 continue 223 224 if (len(values) == 1): 225 params[key] = values[0] 226 else: 227 params[key] = values 228 229 return params 230 231 def _get_post_data(self) -> dict[str, typing.Any]: 232 length = int(self.headers['Content-Length']) 233 payload = self.rfile.read(length).decode(edq.util.dirent.DEFAULT_ENCODING) 234 235 try: 236 request = edq.util.json.loads(payload) 237 except Exception as ex: 238 raise ValueError("Payload is not valid json.") from ex 239 240 return request # type: ignore[no-any-return]
Handle HTTP requests for the web UI.
98 @classmethod 99 def ui_setup(cls, fps: int, user_input_device: WebUserInputDevice) -> None: 100 """ Initialize this handler with information from the UI. """ 101 102 cls._fps = fps 103 cls._user_input_device = user_input_device
Initialize this handler with information from the UI.
105 @classmethod 106 def set_data(cls, state: pacai.core.gamestate.GameState, image: PIL.Image.Image) -> None: 107 """ Set the data passed back to the web page. """ 108 109 buffer = io.BytesIO() 110 image.save(buffer, format = 'png') 111 data_64 = base64.b64encode(buffer.getvalue()).decode(edq.util.dirent.DEFAULT_ENCODING) 112 data_url = f"data:image/png;base64,{data_64}" 113 114 with cls._lock: 115 cls._state = state.copy() 116 cls._image_url = data_url
Set the data passed back to the web page.
118 @classmethod 119 def get_data(cls) -> tuple[pacai.core.gamestate.GameState | None, str | None]: 120 """ Get the data passed back to the web page. """ 121 122 with cls._lock: 123 return cls._state, cls._image_url
Get the data passed back to the web page.
130 def handle(self) -> None: 131 """ 132 Override handle() to ignore dropped connections. 133 """ 134 135 try: 136 http.server.BaseHTTPRequestHandler.handle(self) 137 except BrokenPipeError: 138 logging.info("Connection closed on the client side.")
Override handle() to ignore dropped connections.
242class WebUI(pacai.core.ui.UI): 243 """ 244 A UI that starts a web server and launches a brower window to serve a UI. 245 The web server will accept requests that contains user inputs, 246 and respond with the current game state and a visual representation of the game (a base64 encoded png). 247 """ 248 249 def __init__(self, 250 **kwargs: typing.Any) -> None: 251 input_device = WebUserInputDevice(**kwargs) 252 super().__init__(user_input_device = input_device, **kwargs) 253 254 self._port: int = -1 255 """ 256 The port to start the web server on. 257 The first open port in [START_PORT, END_PORT] will be used. 258 """ 259 260 self._startup_barrier: threading.Barrier = threading.Barrier(2) 261 """ Use a threading barrier to wait for the server thread to start. """ 262 263 self._server_thread: threading.Thread | None = None 264 """ The thread the server will be run on. """ 265 266 self._server: http.server.HTTPServer | None = None 267 """ The HTTP server. """ 268 269 def game_start(self, 270 initial_state: pacai.core.gamestate.GameState, 271 board_highlights: list[pacai.core.board.Highlight] | None = None, 272 **kwargs: typing.Any) -> None: 273 self._start_server() 274 275 super().game_start(initial_state, board_highlights = board_highlights) 276 277 self._launch_page(initial_state) 278 279 def game_complete(self, 280 final_state: pacai.core.gamestate.GameState, 281 board_highlights: list[pacai.core.board.Highlight] | None = None, 282 ) -> None: 283 super().game_complete(final_state, board_highlights = board_highlights) 284 285 # Wait for the UI to make a final request. 286 time.sleep(COMPLETE_WAIT_TIME_SECS) 287 288 self._stop_server() 289 290 def draw(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> None: 291 image = self.draw_image(state) 292 HTTPHandler.set_data(state, image) 293 294 def _start_server(self) -> None: 295 """ Start the HTTP server on another thread. """ 296 297 # Fetch the port. 298 self._port = _find_open_port() 299 300 # Setup the barrier to wait for the server thread to start. 301 self._startup_barrier.reset() 302 303 # Create, but don't start the server. 304 self._server = http.server.ThreadingHTTPServer(('', self._port), HTTPHandler) 305 306 # Setup the handler. 307 HTTPHandler.ui_setup(self._fps, typing.cast(WebUserInputDevice, self._user_input_device)) 308 309 self._server_thread = threading.Thread(target = _run_server, args = (self._server, self._startup_barrier)) 310 self._server_thread.start() 311 312 # Wait for the server to startup. 313 self._startup_barrier.wait() 314 time.sleep(INITIAL_SLEEP_TIME_SEC) 315 316 def _stop_server(self) -> None: 317 """ Stop the HTTP server and thread. """ 318 319 if ((self._server is None) or (self._server_thread is None)): 320 return 321 322 self._server.shutdown() 323 time.sleep(SERVER_WAIT_TIME_SECS) 324 325 if (self._server_thread.is_alive()): 326 self._server_thread.join(REAP_TIME_SECS) 327 328 self._server = None 329 self._server_thread = None 330 331 def _launch_page(self, initial_state: pacai.core.gamestate.GameState) -> None: 332 """ Open the browser window to the web UI page. """ 333 334 image = self.draw_image(initial_state) 335 HTTPHandler.set_data(initial_state, image) 336 337 logging.info("Starting web UI on port %d.", self._port) 338 logging.info("If a browser window does not open, you may use the following link:") 339 logging.info("http://127.0.0.1:%d", self._port) 340 341 webbrowser.open(f"http://127.0.0.1:{self._port}/static/index.html")
A UI that starts a web server and launches a brower window to serve a UI. The web server will accept requests that contains user inputs, and respond with the current game state and a visual representation of the game (a base64 encoded png).
249 def __init__(self, 250 **kwargs: typing.Any) -> None: 251 input_device = WebUserInputDevice(**kwargs) 252 super().__init__(user_input_device = input_device, **kwargs) 253 254 self._port: int = -1 255 """ 256 The port to start the web server on. 257 The first open port in [START_PORT, END_PORT] will be used. 258 """ 259 260 self._startup_barrier: threading.Barrier = threading.Barrier(2) 261 """ Use a threading barrier to wait for the server thread to start. """ 262 263 self._server_thread: threading.Thread | None = None 264 """ The thread the server will be run on. """ 265 266 self._server: http.server.HTTPServer | None = None 267 """ The HTTP server. """
269 def game_start(self, 270 initial_state: pacai.core.gamestate.GameState, 271 board_highlights: list[pacai.core.board.Highlight] | None = None, 272 **kwargs: typing.Any) -> None: 273 self._start_server() 274 275 super().game_start(initial_state, board_highlights = board_highlights) 276 277 self._launch_page(initial_state)
Initialize the UI with the game's initial state.
279 def game_complete(self, 280 final_state: pacai.core.gamestate.GameState, 281 board_highlights: list[pacai.core.board.Highlight] | None = None, 282 ) -> None: 283 super().game_complete(final_state, board_highlights = board_highlights) 284 285 # Wait for the UI to make a final request. 286 time.sleep(COMPLETE_WAIT_TIME_SECS) 287 288 self._stop_server()
Update the UI with the game's final state.
290 def draw(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> None: 291 image = self.draw_image(state) 292 HTTPHandler.set_data(state, image)
Visualize the state of the game to the UI. This is the typically the main override point for children. Note that how this method visualizes the game completely unrelated to how the draw_image() method works. draw() will render to whatever the specific UI for the child class is, while draw_image() specifically creates an image which will be used for animations. If the child UI is also image-based than it can leverage draw_image(), but there is no requirement to do that.
Inherited Members
377@typing.runtime_checkable 378class RequestHandler(typing.Protocol): 379 """ Functions that can be used as HTTP request handlers by HTTPHandler. """ 380 381 def __call__(self, handler: HTTPHandler, path: str, params: dict) -> RequestHandlerResult: 382 ...
Functions that can be used as HTTP request handlers by HTTPHandler.
1953def _no_init_or_replace_init(self, *args, **kwargs): 1954 cls = type(self) 1955 1956 if cls._is_protocol: 1957 raise TypeError('Protocols cannot be instantiated') 1958 1959 # Already using a custom `__init__`. No need to calculate correct 1960 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1961 if cls.__init__ is not _no_init_or_replace_init: 1962 return 1963 1964 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1965 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1966 # searches for a proper new `__init__` in the MRO. The new `__init__` 1967 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1968 # instantiation of the protocol subclass will thus use the new 1969 # `__init__` and no longer call `_no_init_or_replace_init`. 1970 for base in cls.__mro__: 1971 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1972 if init is not _no_init_or_replace_init: 1973 cls.__init__ = init 1974 break 1975 else: 1976 # should not happen 1977 cls.__init__ = object.__init__ 1978 1979 cls.__init__(self, *args, **kwargs)