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]
THIS_DIR: str = '/home/runner/work/pacai/pacai/pacai/ui'
STATIC_DIR: str = '/home/runner/work/pacai/pacai/pacai/ui/../resources/webui'
START_PORT: int = 30000
END_PORT: int = 40000
INITIAL_SLEEP_TIME_SEC: float = 0.05
SOCKET_SLEEP_TIME_SECS: float = 0.1
SERVER_WAIT_TIME_SECS: float = 0.25
REAP_TIME_SECS: float = 0.25
COMPLETE_WAIT_TIME_SECS: float = 0.15
SERVER_POLL_INTERVAL_SECS: float = 0.1
RequestHandlerResult: TypeAlias = tuple[dict | str | bytes | None, int | None, dict | None] | None
class WebUserInputDevice(pacai.core.ui.UserInputDevice):
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.

WebUserInputDevice( char_mapping: dict[str, pacai.core.action.Action] | None = None, **kwargs: Any)
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. """
def add_keys(self, keys: list[str]) -> None:
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.

def get_inputs(self) -> list[pacai.core.action.Action]:
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.

class HTTPHandler(http.server.BaseHTTPRequestHandler):
 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.

@classmethod
def ui_setup( cls, fps: int, user_input_device: WebUserInputDevice) -> None:
 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.

@classmethod
def set_data( cls, state: pacai.core.gamestate.GameState, image: PIL.Image.Image) -> None:
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.

@classmethod
def get_data(cls) -> tuple[pacai.core.gamestate.GameState | None, str | None]:
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.

def log_message(self, *args: Any) -> None:
125    def log_message(self, *args: typing.Any) -> None:
126        """
127        Reduce the logging noise.
128        """

Reduce the logging noise.

def handle(self) -> None:
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.

def do_POST(self) -> None:
140    def do_POST(self) -> None:  # pylint: disable=invalid-name
141        """ Handle POST requests. """
142
143        self._handle_request(self._get_post_data)

Handle POST requests.

def do_GET(self) -> None:
145    def do_GET(self) -> None:  # pylint: disable=invalid-name
146        """ Handle GET requests. """
147
148        self._handle_request(self._get_get_data)

Handle GET requests.

class WebUI(pacai.core.ui.UI):
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).

WebUI(**kwargs: Any)
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. """
def game_start( self, initial_state: pacai.core.gamestate.GameState, board_highlights: list[pacai.core.board.Highlight] | None = None, **kwargs: Any) -> None:
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.

def game_complete( self, final_state: pacai.core.gamestate.GameState, board_highlights: list[pacai.core.board.Highlight] | None = None) -> None:
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.

def draw(self, state: pacai.core.gamestate.GameState, **kwargs: Any) -> None:
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.

@typing.runtime_checkable
class RequestHandler(typing.Protocol):
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.

RequestHandler(*args, **kwargs)
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)
ROUTES: list[tuple[str, RequestHandler]] = [('^/$', <function _handler_redirect.<locals>.handler_func>), ('^/index.html$', <function _handler_redirect.<locals>.handler_func>), ('^/static$', <function _handler_redirect.<locals>.handler_func>), ('^/static/$', <function _handler_redirect.<locals>.handler_func>), ('^/favicon.ico$', <function _handler_redirect.<locals>.handler_func>), ('^/static/', <function _handler_static>), ('^/api/init$', <function _handler_init>), ('^/api/update$', <function _handler_update>)]