pythonnative/src/pythonnative/preview.py at v0.43.1 · pythonnative/pythonnative · GitHub

Repository navigation

History

287 lines (248 loc) · 10.3 KB

Raw

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

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

"""Development server and browser preview.

The browser is a protocol-2 renderer with Yoga WebAssembly layout. Python runs

on the standard application asyncio loop; the transport's coordinator and

network threads forward messages to their owners. The dev server watches app

sources, synchronizes mobile clients, and applies compatible Fast Refresh.

PN_PLATFORM=web must be set before importing pythonnative; the CLI launches a

child process with that environment when needed.

"""

from __future__ import annotations

import os

import sys

import threading

import time

import traceback

import webbrowser

from typing import Any, Callable, Dict, List, Optional

__all__ = ["PreviewSession", "serve"]

Logger = Callable[[str], None]

class PreviewSession:

"""Everything one ``pn start`` invocation owns; see [`serve`][pythonnative.preview.serve]."""

def __init__(

self,

project_root: str,

entry_module: str,

host: str = "0.0.0.0",

port: int = 8765,

project_name: str = "",

log: Optional[Logger] = None,

) -> None:

from .bridge.web import WebTransport

from .devserver import DevServer

self.project_root = os.path.abspath(project_root)

self.entry_module = entry_module

self.log: Logger = log or (lambda line: print(line, file=sys.stderr, flush=True))

self.server = DevServer(

self.project_root, entry_module, host=host, port=port, project_name=project_name, log=self.log

)

self.transport = WebTransport(log=self.log)

self.transport.on_peer_changed = self._on_peer_changed

self.transport.on_dev_message = self._on_dev_message

self._stop = threading.Event()

self._unsubscribe: Optional[Callable[[], None]] = None

# -- lifecycle -------------------------------------------------------

def start(self) -> None:

"""Install the transport, start the server, and begin watching."""

from . import bridge, diagnostics

from .utils import IS_WEB

if not IS_WEB:

raise RuntimeError(

"The browser preview needs PN_PLATFORM=web before pythonnative is imported "

"(`pn start` and `pn preview` set it for you)."

)

if self.project_root not in sys.path:

sys.path.insert(0, self.project_root)

os.environ.setdefault("PN_ENTRY_MODULE", self.entry_module)

os.environ.setdefault("PN_STORAGE_DIR", os.path.join(self.project_root, "build", "preview", "storage"))

diagnostics.set_dev_mode(True)

# Bottom of the reporter stack: screens register their own RedBox

# above this, so it only sees errors raised before a screen exists

# (a failing import of the entry module, say).

diagnostics.set_error_reporter(self, self._report_error)

bridge.set_transport(self.transport)

from .runtime import start as start_runtime

start_runtime()

self.server.set_preview_channel(self.transport)

self._unsubscribe = self.server.add_change_listener(self._on_sources_changed)

self.server.start()

def run(self) -> None:

"""Run the main loop until ``stop`` (or Ctrl+C)."""

try:

self.transport.run_main_loop(until=self._stop.is_set)

except KeyboardInterrupt:

pass

def stop(self) -> None:

"""Tear everything down."""

self._stop.set()

self.transport.stop()

if self._unsubscribe is not None:

self._unsubscribe()

self._unsubscribe = None

try:

self._destroy_hosts()

except Exception:

pass

self.server.stop()

from . import bridge, diagnostics

diagnostics.set_error_reporter(self, None)

bridge.set_transport(None)

# -- source changes ----------------------------------------------------

def _on_sources_changed(self, change: Any, snapshot: Any) -> None:

"""Watcher thread: schedule a reload of the preview's own app."""

from .devserver.watcher import modules_for_paths

modules = modules_for_paths(list(change.changed) + list(change.removed))

if not modules:

return

def _apply() -> None:

from .hosts import live_hosts

from .hot_reload import apply_reload

hosts = list(live_hosts())

started = time.monotonic()

result = apply_reload(modules, hosts)

elapsed_ms = (time.monotonic() - started) * 1000.0

if result.mode == "error":

self.log("[pn] reload failed:")

for line in (result.error or "").rstrip().split("\n"):

self.log(f" {line}")

self.transport.send_dev({"type": "reload", "ok": False, "error": result.error, "modules": modules})

return

if result.mode == "none":

return

label = "Fast Refresh" if result.mode == "fast_refresh" else "Remounted"

self.log(f"[pn] {label}: {', '.join(result.reloaded)} ({elapsed_ms:.0f} ms)")

self.transport.send_dev(

{"type": "reload", "ok": True, "mode": result.mode, "modules": result.reloaded, "ms": elapsed_ms}

)

self.transport.post_to_application(_apply)

# -- peers -------------------------------------------------------------

def _on_peer_changed(self, connected: bool) -> None:

if connected:

self.log("[pn] browser preview connected")

self._send_hello()

return

self.log("[pn] browser preview disconnected")

self._destroy_hosts()

def _send_hello(self) -> None:

"""Tell the page what to mount; it answers by creating the entry screen."""

self.transport.send_dev(

{

"type": "hello",

"entry": self.entry_module,

"project": self.server.project_name or os.path.basename(self.project_root),

}

)

def _report_error(self, exc: BaseException, phase: str) -> None:

"""Print an error to the terminal and mirror it into the preview page."""

text = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)).rstrip()

self.log(f"[pn] error during {phase}:")

for line in text.split("\n"):

self.log(f" {line}")

self.transport.send_dev({"type": "error", "phase": phase, "text": text})

def _destroy_hosts(self) -> None:

"""Unmount every screen the page created (it is gone, so are its views)."""

from .hosts.native import live_hosts

from .native_views import get_registry

for host in list(live_hosts()):

try:

host.on_destroy()

except Exception:

traceback.print_exc()

backend = get_registry()

reset = getattr(backend, "reset", None)

if callable(reset):

reset()

def _on_dev_message(self, payload: Dict[str, Any]) -> None:

kind = payload.get("type")

if kind == "log":

self.log(f"[browser] {payload.get('text', '')}")

elif kind == "error":

self.log(f"[browser error] {payload.get('text', '')}")

elif kind == "hello":

agent = str(payload.get("user_agent", ""))

device = payload.get("device", "")

size = f"{payload.get('width', '?')}x{payload.get('height', '?')}"

self.log(f"[pn] preview page: {device} {size} ({_browser_family(agent)})")

elif kind == "remount":

self.log("[pn] remounting the app")

self._destroy_hosts()

self._send_hello()

# -- info ----------------------------------------------------------------

def urls(self) -> List[str]:

"""Every URL the server can be reached at (local first)."""

from .devserver import lan_addresses

info = self.server.info

urls = [info.url("localhost")]

for address in lan_addresses():

url = info.url(address)

if url not in urls:

urls.append(url)

return urls

def _browser_family(user_agent: str) -> str:

agent = user_agent.lower()

for needle, name in (("firefox", "Firefox"), ("edg/", "Edge"), ("chrome", "Chrome"), ("safari", "Safari")):

if needle in agent:

return name

return "browser"

def serve(

entry_module: str,

project_root: Optional[str] = None,

host: str = "0.0.0.0",

port: int = 8765,

project_name: str = "",

open_browser: bool = False,

log: Optional[Logger] = None,

banner: bool = True,

ready: Optional[Callable[[PreviewSession], None]] = None,

) -> None:

"""Run the dev server (and browser preview) until interrupted.

Args:

entry_module: The app's entry module (``"app.main"``).

project_root: Directory containing ``app/``; defaults to the

current directory. Added to ``sys.path``.

host: Bind address (``0.0.0.0`` so devices can connect).

port: TCP port (``0`` picks a free one).

project_name: Shown in the preview page.

open_browser: Open the preview page in the default browser.

log: Where status lines go (stderr by default).

banner: Print the connection banner.

ready: Called once the server is listening (tests).

Raises:

RuntimeError: If ``PN_PLATFORM=web`` was not set before

PythonNative was imported (the CLI sets it for you).

OSError: If the port is taken.

"""

session = PreviewSession(

project_root or os.getcwd(),

entry_module,

host=host,

port=port,

project_name=project_name,

log=log,

)

session.start()

urls = session.urls()

if banner:

emit = session.log

emit("")

emit(f" PythonNative dev server for {project_name or entry_module}")

emit("")

emit(f" Browser preview: {urls[0]}")

for url in urls[1:]:

emit(f" Devices (LAN): {url}")

emit("")

emit(" Debug builds made with `pn run` connect here automatically.")

emit(" Press Ctrl+C to stop.")

emit("")

if open_browser:

try:

webbrowser.open(urls[0])

except Exception:

pass

if ready is not None:

ready(session)

try:

session.run()

finally:

session.stop()

pythonnative/src/pythonnative/preview.py at v0.43.1 · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

preview.py

Latest commit

History

Breadcrumbs

preview.py

File metadata and controls