pythonnative/src/pythonnative/hot_reload.py at main · andevsr/pythonnative · GitHub

Repository navigation

History

653 lines (545 loc) · 24.9 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

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

"""Hot-reload support for PythonNative development.

Two cooperating pieces:

- **Host-side**: [`FileWatcher`][pythonnative.hot_reload.FileWatcher]

polls the developer's ``app/`` directory for ``.py`` changes and

triggers a callback (typically ``adb push`` on Android or a

``simctl`` file copy on iOS).

- **Device-side**:

[`ModuleReloader`][pythonnative.hot_reload.ModuleReloader] reloads

changed Python modules using ``importlib`` and asks the screen

host to re-render its current tree.

Two strategies share the device-side surface:

- **Fast Refresh** (default): after reloading the changed modules

the reconciler tree is walked and every component function whose

module was reloaded is swapped in place. Hook state, navigation

state, and even scroll positions survive because the underlying

``VNode`` objects are reused; the next render simply calls the

new function bodies through the old slots.

- **Full remount**: when the in-place swap fails (e.g. the new

module raised at import time, or a render exception bubbled out

while running the new function), the host falls back to building

a brand-new reconciler tree. State is lost but the app keeps

running.

Example:

Integrated into ``pn run --hot-reload``:

```python

from pythonnative.hot_reload import FileWatcher

def push(changed):

for path in changed:

print("changed:", path)

watcher = FileWatcher("app/", on_change=push)

watcher.start()

```

"""

import importlib

import importlib.util

import json

import os

import sys

import threading

import time

from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Set

DEV_ROOT_DIR = "pythonnative_dev"

"""Name of the writable on-device directory that shadows bundled app code."""

RELOAD_MANIFEST = "reload.json"

"""Manifest filename written by the host and polled by native templates."""

def configure_dev_environment(writable_root: str) -> str:

"""Create and prioritize the writable hot-reload source overlay.

The returned directory is inserted at the front of `sys.path`, so a

pushed `app/main.py` shadows the copy bundled into the native

application. Templates call this before importing user code.

Args:

writable_root: Platform data directory that the app can write to

(Android `filesDir`, iOS `Documents`, or a test directory).

Returns:

Absolute path to the hot-reload overlay root.

"""

dev_root = os.path.abspath(os.path.join(writable_root, DEV_ROOT_DIR))

os.makedirs(os.path.join(dev_root, "app"), exist_ok=True)

if dev_root in sys.path:

sys.path.remove(dev_root)

sys.path.insert(0, dev_root)

os.environ["PYTHONNATIVE_HOT_RELOAD_ROOT"] = dev_root

return dev_root

def manifest_path_for(dev_root: str) -> str:

"""Return the reload-manifest path inside a hot-reload overlay."""

return os.path.join(dev_root, RELOAD_MANIFEST)

def _overlay_module_path(module_name: str) -> Optional[str]:

dev_root = os.environ.get("PYTHONNATIVE_HOT_RELOAD_ROOT")

if not dev_root:

return None

rel_parts = module_name.split(".")

module_path = os.path.join(dev_root, *rel_parts) + ".py"

if os.path.exists(module_path):

return module_path

package_path = os.path.join(dev_root, *rel_parts, "__init__.py")

if os.path.exists(package_path):

return package_path

return None

# ======================================================================

# Host-side file watcher

# ======================================================================

class FileWatcher:

"""Watch a directory tree for `.py` file changes.

Uses simple `os.path.getmtime` polling rather than a native

inotify/FSEvents binding so the watcher works on every platform

where Python runs without extra dependencies.

Args:

watch_dir: Directory to watch (recursively).

on_change: Called with a list of changed file paths when

modifications are detected.

interval: Polling interval, in seconds.

Attributes:

watch_dir: Directory being watched.

on_change: Change callback.

interval: Polling interval.

"""

def __init__(self, watch_dir: str, on_change: Callable[[List[str]], None], interval: float = 1.0) -> None:

self.watch_dir = watch_dir

self.on_change = on_change

self.interval = interval

self._running = False

self._thread: Optional[threading.Thread] = None

self._mtimes: Dict[str, float] = {}

def start(self) -> None:

"""Begin watching in a background daemon thread.

Performs an initial scan to seed mtimes so the first

notification reflects subsequent edits, not pre-existing files.

"""

self._running = True

self._scan()

self._thread = threading.Thread(target=self._loop, daemon=True)

self._thread.start()

def stop(self) -> None:

"""Stop the watcher and join the background thread."""

self._running = False

if self._thread is not None:

self._thread.join(timeout=self.interval * 2)

self._thread = None

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

changed: List[str] = []

current_files: set = set()

for root, _dirs, files in os.walk(self.watch_dir):

for fname in files:

if not fname.endswith(".py"):

continue

fpath = os.path.join(root, fname)

current_files.add(fpath)

try:

mtime = os.path.getmtime(fpath)

except OSError:

continue

if fpath in self._mtimes:

if mtime > self._mtimes[fpath]:

changed.append(fpath)

self._mtimes[fpath] = mtime

for old in list(self._mtimes):

if old not in current_files:

del self._mtimes[old]

return changed

def _loop(self) -> None:

while self._running:

time.sleep(self.interval)

changed = self._scan()

if changed:

try:

self.on_change(changed)

except Exception:

pass

# ======================================================================

# Device-side module reloader

# ======================================================================

class ModuleReloader:

"""Reload changed Python modules on device and trigger a re-render.

Designed to be invoked from device-side glue when a hot-reload

push completes. All public methods are static; the class holds a

single piece of process-wide state (the manifest version that

has most recently been applied to ``sys.modules``) so that

multiple screen hosts polling the same manifest do not each

re-execute the user-app modules. The first host to see a new

version pays the ``reload_modules`` cost; subsequent hosts on the

same version refresh only their own reconciler tree against the

already-fresh modules.

"""

_last_reloaded_version: Optional[str] = None

_reload_lock = threading.Lock()

@staticmethod

def reload_module(module_name: str) -> bool:

"""Reload a single module by its dotted name.

Args:

module_name: Dotted module name (e.g., `"app.main"`).

Returns:

`True` if the module imported successfully from the current

`sys.path`; `False` otherwise.

"""

previous = sys.modules.get(module_name)

try:

importlib.invalidate_caches()

overlay_path = _overlay_module_path(module_name)

if overlay_path is not None:

spec = importlib.util.spec_from_file_location(module_name, overlay_path)

if spec is None or spec.loader is None:

return False

module = importlib.util.module_from_spec(spec)

sys.modules[module_name] = module

spec.loader.exec_module(module)

else:

sys.modules.pop(module_name, None)

importlib.import_module(module_name)

return True

except Exception:

if previous is not None:

sys.modules[module_name] = previous

else:

sys.modules.pop(module_name, None)

return False

@staticmethod

def reload_modules(module_names: Sequence[str]) -> List[str]:

"""Reload the modules that are already imported.

Args:

module_names: Dotted module names to reload.

Returns:

Names that were successfully reloaded.

"""

importlib.invalidate_caches()

reloaded: List[str] = []

seen: set[str] = set()

for module_name in module_names:

if not module_name or module_name in seen:

continue

seen.add(module_name)

if ModuleReloader.reload_module(module_name):

reloaded.append(module_name)

return reloaded

@staticmethod

def reload_modules_for_version(

module_names: Sequence[str],

version: Optional[str],

) -> List[str]:

"""Reload ``module_names`` for ``version``, deduping across hosts.

Each native screen host on iOS / Android runs its own poll

loop and would otherwise call

[`reload_modules`][pythonnative.hot_reload.ModuleReloader.reload_modules]

independently for the same manifest version. That re-executes

every user-app module N times (once per host) per file change,

producing N different generations of the same function objects

in ``sys.modules`` and leaving each host's reconciler tree

pointing at a different generation. Beyond the wasted work,

the inconsistent state has been observed to crash UIKit on iOS

with ``CALayerInvalidGeometry`` (NaN values fed into ``setFrame_:``

during the interleaved renders).

This helper serializes on

[`_reload_lock`][pythonnative.hot_reload.ModuleReloader] and uses

[`_last_reloaded_version`][pythonnative.hot_reload.ModuleReloader]

to ensure only the *first* host to see a given ``version``

actually re-executes the modules. Subsequent hosts on the same

version get back the already-fresh entries from ``sys.modules``

so their own

[`refresh_in_place`][pythonnative.hot_reload.ModuleReloader.refresh_in_place]

pass can still rewrite their tree against the same generation.

Args:

module_names: Dotted module names to reload.

version: Manifest version this reload is processing. When

``None`` (e.g. tests calling reload directly) the call

falls back to the unconditional

[`reload_modules`][pythonnative.hot_reload.ModuleReloader.reload_modules]

behavior.

Returns:

The list of module names that are currently fresh in

``sys.modules``, either freshly reloaded by this call, or

already reloaded by an earlier host for the same version.

"""

with ModuleReloader._reload_lock:

if version is not None and version == ModuleReloader._last_reloaded_version:

return [name for name in module_names if name in sys.modules]

reloaded = ModuleReloader.reload_modules(module_names)

if reloaded and version is not None:

ModuleReloader._last_reloaded_version = version

return reloaded

@staticmethod

def expand_reload_targets(changed_modules: Sequence[str], component_path: str) -> List[str]:

"""Expand a manifest of changed modules into the full reload order.

When a user edits ``app/screens/home.py``, only that file is in

the manifest. But the entry-point module ``app.main`` has

bindings like ``from app.screens.home import HomeScreen`` that

need to be re-evaluated against the freshly-loaded

``app.screens.home``; likewise other user-app modules may carry

transitive bindings (e.g. through a shared ``app/theme.py``)

that go stale if only the changed file is reloaded.

This helper computes the full ordered reload list:

1. Explicitly changed modules first (in the order given), so

their fresh source replaces the cached version in

``sys.modules`` before any dependent modules re-execute.

2. All other currently-imported modules under the entry-point's

top-level package, deepest first. The depth heuristic biases

toward leaves so re-executing a screen file picks up the

newest shared utilities before the file that imports it does.

3. The entry-point module itself, last, so its

``from ... import`` bindings rebind against everything that

was refreshed in steps 1 and 2.

Modules outside the entry-point's top-level package

(``pythonnative.*``, stdlib, third-party) are never included;

framework code is not reloaded.

Args:

changed_modules: Modules reported as changed by the host

file-watcher (already in dotted form).

component_path: The host's entry-point identifier, either a

module path (``"app.main"``) or a dotted attribute path

(``"app.main.RootScreen"``).

Returns:

The ordered list of modules to feed to

[`reload_modules`][pythonnative.hot_reload.ModuleReloader.reload_modules].

"""

entry_module: Optional[str] = None

if component_path in sys.modules:

entry_module = component_path

elif "." in component_path:

parent = component_path.rsplit(".", 1)[0]

if parent in sys.modules:

entry_module = parent

app_prefix: Optional[str] = None

if entry_module:

app_prefix = entry_module.split(".")[0]

else:

for m in changed_modules:

if m:

app_prefix = m.split(".")[0]

break

app_modules: Set[str] = set()

if app_prefix:

for name in list(sys.modules):

if name == app_prefix or name.startswith(app_prefix + "."):

app_modules.add(name)

ordered: List[str] = []

seen: Set[str] = set()

for m in changed_modules:

if m and m not in seen:

ordered.append(m)

seen.add(m)

others = [m for m in app_modules if m not in seen and m != entry_module]

others.sort(key=lambda m: (-m.count("."), m))

for m in others:

ordered.append(m)

seen.add(m)

if entry_module:

if entry_module in seen:

ordered.remove(entry_module)

ordered.append(entry_module)

return ordered

@staticmethod

def file_to_module(file_path: str, base_dir: str = "") -> Optional[str]:

"""Convert a file path to a dotted module name.

Args:

file_path: Path to a `.py` file (absolute or relative).

base_dir: Base directory that names should be relative to.

If empty, `file_path` is treated as already relative.

Returns:

The dotted module name (e.g., `"app.screens.home"`), or

`None` for an empty path.

"""

rel = os.path.relpath(file_path, base_dir) if base_dir else file_path

rel = rel.replace("\\", os.sep).replace("/", os.sep).lstrip(os.sep)

if rel.endswith(".py"):

rel = rel[:-3]

parts = rel.replace(os.sep, ".").split(".")

if parts[-1] == "__init__":

parts = parts[:-1]

return ".".join(parts) if parts else None

@staticmethod

def modules_from_files(file_paths: Sequence[str], base_dir: str = "") -> List[str]:

"""Convert Python source paths to importable module names."""

modules: List[str] = []

for file_path in file_paths:

module = ModuleReloader.file_to_module(file_path, base_dir=base_dir)

if module is not None:

modules.append(module)

return modules

@staticmethod

def reload_screen(screen_instance: Any, module_names: Optional[Sequence[str]] = None) -> None:

"""Force a screen re-render after a module reload.

Args:

screen_instance: A `_ScreenHost` instance (or duck-typed

equivalent) that exposes a `_reconciler` attribute.

module_names: Optional modules that changed. Reload-aware

screen hosts use this to refresh imports before re-render.

"""

reload_fn = getattr(screen_instance, "reload", None)

if callable(reload_fn):

reload_fn(list(module_names or []))

return

from .screen import _request_render

if hasattr(screen_instance, "_reconciler") and screen_instance._reconciler is not None:

_request_render(screen_instance)

@staticmethod

def find_replacement_function(old_fn: Any) -> Optional[Any]:

"""Locate a function's post-reload counterpart by qualname.

Functions decorated with [`component`][pythonnative.component]

store the user's original function on the wrapper's

``__wrapped__`` attribute and forward ``__module__`` /

``__qualname__`` so that the reconciler's stored

``element.type`` (the unwrapped function) still has the

information needed to re-resolve after a module reload.

Args:

old_fn: The function captured in an

[`Element`][pythonnative.Element]'s ``type`` slot.

Returns:

The reloaded module's matching function, ``None`` if no

replacement was found, or the original function itself

when the module has not been reloaded (so callers can

skip the swap).

"""

module_name = getattr(old_fn, "__module__", None)

qualname = getattr(old_fn, "__qualname__", None) or getattr(old_fn, "__name__", None)

if not module_name or not qualname:

return None

if "<locals>" in qualname:

return None # nested functions are not addressable from the module surface

module = sys.modules.get(module_name)

if module is None:

return None

obj: Any = module

for part in qualname.split("."):

obj = getattr(obj, part, None)

if obj is None:

return None

if getattr(obj, "_pn_component", False):

obj = getattr(obj, "__wrapped__", obj)

if obj is old_fn:

return None

return obj

@staticmethod

def build_replacement_map(reconciler: Any, reloaded_modules: Iterable[str]) -> Dict[Any, Any]:

"""Compute ``{old_function: new_function}`` for one tree.

The reconciler's stored tree references the *pre-reload*

component functions through ``VNode.element.type``. This

method walks the tree, collects every callable type whose

``__module__`` was just reloaded, and asks

[`find_replacement_function`][pythonnative.hot_reload.ModuleReloader.find_replacement_function]

for its successor.

Args:

reconciler: The reconciler whose

``_tree`` should be inspected.

reloaded_modules: Set of module names that were just

reloaded (only callables from these modules are

considered).

Returns:

A mapping suitable for passing to

[`swap_components_in_tree`][pythonnative.hot_reload.ModuleReloader.swap_components_in_tree].

"""

modules: Set[str] = {m for m in reloaded_modules if m}

if not modules or reconciler is None or getattr(reconciler, "_tree", None) is None:

return {}

seen: Set[int] = set()

mapping: Dict[Any, Any] = {}

def visit(vnode: Any) -> None:

if vnode is None:

return

elem = getattr(vnode, "element", None)

if elem is not None and callable(elem.type):

fn = elem.type

fn_id = id(fn)

if fn_id not in seen:

seen.add(fn_id)

if getattr(fn, "__module__", None) in modules:

replacement = ModuleReloader.find_replacement_function(fn)

if replacement is not None and replacement is not fn:

mapping[fn] = replacement

for child in getattr(vnode, "children", []) or []:

visit(child)

visit(reconciler._tree)

return mapping

@staticmethod

def swap_components_in_tree(reconciler: Any, replacement_map: Dict[Any, Any]) -> int:

"""Apply a ``{old: new}`` map to every node in the reconciler tree.

Mutates ``vnode.element.type`` directly so the NEXT diff sees

identical types and reuses VNodes (preserving hook state).

Pending ``Element`` trees stored on ``vnode._rendered`` are

rewritten too because the reconciler reads from them when

comparing keys across renders.

Returns:

The number of element type references that were rewritten.

"""

if not replacement_map or reconciler is None or getattr(reconciler, "_tree", None) is None:

return 0

rewrites = 0

def rewrite_element_tree(element: Any) -> None:

nonlocal rewrites

if element is None:

return

new_type = replacement_map.get(element.type)

if new_type is not None:

element.type = new_type

rewrites += 1

for child in element.children or []:

rewrite_element_tree(child)

def visit(vnode: Any) -> None:

if vnode is None:

return

if getattr(vnode, "element", None) is not None:

rewrite_element_tree(vnode.element)

rendered = getattr(vnode, "_rendered", None)

if isinstance(rendered, (list, tuple)):

for rendered_el in rendered:

rewrite_element_tree(rendered_el)

elif rendered is not None:

rewrite_element_tree(rendered)

for child in getattr(vnode, "children", []) or []:

visit(child)

visit(reconciler._tree)

return rewrites

@staticmethod

def refresh_in_place(reconciler: Any, reloaded_modules: Iterable[str]) -> bool:

"""Try a state-preserving Fast Refresh for one reconciler.

Returns:

``True`` if any component function was replaced (callers

should then trigger a re-render). ``False`` means the

tree already references the latest functions (or has no

nodes from the reloaded modules at all).

"""

replacement_map = ModuleReloader.build_replacement_map(reconciler, reloaded_modules)

if not replacement_map:

return False

rewrites = ModuleReloader.swap_components_in_tree(reconciler, replacement_map)

if rewrites > 0 and hasattr(reconciler, "reset_hook_signatures"):

# New component bodies may legitimately call a different

# hook sequence; forget the recorded signatures so the

# dev-mode order guard doesn't flag the refresh itself.

reconciler.reset_hook_signatures()

return rewrites > 0

@staticmethod

def reload_from_manifest(

screen_instance: Any,

manifest_path: str,

last_version: Optional[str] = None,

) -> Optional[str]:

"""Apply a reload manifest if it is newer than `last_version`.

Args:

screen_instance: Screen host to refresh.

manifest_path: JSON manifest written by the CLI.

last_version: Version already applied by this screen host.

Returns:

The manifest version after applying, or `last_version` when

no new manifest is available.

"""

if not os.path.exists(manifest_path):

return last_version

with open(manifest_path, encoding="utf-8") as f:

manifest = json.load(f)

version = str(manifest.get("version", ""))

if not version or version == last_version:

return last_version

modules = manifest.get("modules")

if not isinstance(modules, list):

files = manifest.get("files", [])

modules = ModuleReloader.modules_from_files(files if isinstance(files, list) else [])

# Stash the version on the host so `_reload_host` can dedupe

# `reload_modules` across multiple hosts polling the same

# manifest. See `reload_modules_for_version`.

previous_pending = getattr(screen_instance, "_hot_reload_pending_version", None)

try:

screen_instance._hot_reload_pending_version = version

ModuleReloader.reload_screen(screen_instance, [str(module) for module in modules])

finally:

screen_instance._hot_reload_pending_version = previous_pending

return version

pythonnative/src/pythonnative/hot_reload.py at main · andevsr/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

hot_reload.py

Latest commit

History

Breadcrumbs

hot_reload.py

File metadata and controls