Start a server with walker
(
walker: PygWalker,
port: Optional[int],
*,
auto_open: bool,
auto_shutdown: bool,
)
| 121 | |
| 122 | |
| 123 | def _start_server( |
| 124 | walker: PygWalker, |
| 125 | port: Optional[int], |
| 126 | *, |
| 127 | auto_open: bool, |
| 128 | auto_shutdown: bool, |
| 129 | ): |
| 130 | """Start a server with walker""" |
| 131 | state = _GlobalState(auto_shutdown=auto_shutdown) |
| 132 | walker._init_callback(BaseCommunication(str(walker.gid))) |
| 133 | |
| 134 | handler = _create_handler_with_walker(walker, state) |
| 135 | if port is None: |
| 136 | port = find_free_port() |
| 137 | address = f"http://localhost:{port}" |
| 138 | |
| 139 | def _listen_shutdown(): |
| 140 | while 1: |
| 141 | time.sleep(1) |
| 142 | if time.time() - state.last_health_time > _MAX_HEALTH_TIMEOUT_SECONDS: |
| 143 | httpd.shutdown() |
| 144 | break |
| 145 | |
| 146 | try: |
| 147 | with CustomTCPServer(("127.0.0.1", port), handler) as httpd: |
| 148 | if auto_open: |
| 149 | threading.Thread(target=_open_browser, args=(address,)).start() |
| 150 | if auto_shutdown: |
| 151 | threading.Thread(target=_listen_shutdown).start() |
| 152 | httpd.serve_forever() |
| 153 | except KeyboardInterrupt: |
| 154 | pass |
| 155 | |
| 156 | |
| 157 | def walk( |
no test coverage detected