GitHub Viewer
"""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()