131 lines
4.0 KiB
Python
131 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Textual Development Server - Live reload and debugging for Textual apps
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
from pathlib import Path
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import FileSystemEventHandler
|
|
import threading
|
|
import signal
|
|
import os
|
|
|
|
class TextualAppHandler(FileSystemEventHandler):
|
|
def __init__(self, app_path, restart_callback):
|
|
self.app_path = app_path
|
|
self.restart_callback = restart_callback
|
|
self.last_restart = 0
|
|
|
|
def on_modified(self, event):
|
|
if event.is_directory:
|
|
return
|
|
|
|
# Only restart for Python files
|
|
if not event.src_path.endswith('.py'):
|
|
return
|
|
|
|
# Throttle restarts
|
|
now = time.time()
|
|
if now - self.last_restart < 2.0:
|
|
return
|
|
|
|
print(f"📝 File changed: {event.src_path}")
|
|
print("🔄 Restarting Textual app...")
|
|
self.last_restart = now
|
|
self.restart_callback()
|
|
|
|
class TextualDevServer:
|
|
def __init__(self, app_path, watch_dirs=None):
|
|
self.app_path = Path(app_path)
|
|
self.watch_dirs = watch_dirs or [self.app_path.parent]
|
|
self.current_process = None
|
|
self.observer = None
|
|
|
|
def start_app(self):
|
|
"""Start the Textual app"""
|
|
if self.current_process:
|
|
self.current_process.terminate()
|
|
self.current_process.wait()
|
|
|
|
print(f"🚀 Starting {self.app_path}")
|
|
self.current_process = subprocess.Popen([
|
|
sys.executable, str(self.app_path)
|
|
], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
|
|
# Start threads to handle output
|
|
threading.Thread(target=self._handle_output, daemon=True).start()
|
|
|
|
def _handle_output(self):
|
|
"""Handle app output in real-time"""
|
|
if not self.current_process:
|
|
return
|
|
|
|
while self.current_process.poll() is None:
|
|
line = self.current_process.stdout.readline()
|
|
if line:
|
|
print(f"📱 APP: {line.strip()}")
|
|
|
|
# Print any remaining output
|
|
stdout, stderr = self.current_process.communicate()
|
|
if stdout:
|
|
print(f"📱 APP STDOUT: {stdout}")
|
|
if stderr:
|
|
print(f"❌ APP STDERR: {stderr}")
|
|
|
|
def start_watching(self):
|
|
"""Start file watching"""
|
|
self.observer = Observer()
|
|
handler = TextualAppHandler(self.app_path, self.start_app)
|
|
|
|
for watch_dir in self.watch_dirs:
|
|
print(f"👀 Watching: {watch_dir}")
|
|
self.observer.schedule(handler, str(watch_dir), recursive=True)
|
|
|
|
self.observer.start()
|
|
|
|
def run(self):
|
|
"""Run the development server"""
|
|
print("🔧 Textual Development Server")
|
|
print("=" * 50)
|
|
|
|
# Start initial app
|
|
self.start_app()
|
|
|
|
# Start file watching
|
|
self.start_watching()
|
|
|
|
try:
|
|
print("✅ Development server running. Press Ctrl+C to stop.")
|
|
print("💡 Edit Python files to see live reload!")
|
|
|
|
while True:
|
|
time.sleep(1)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n🛑 Stopping development server...")
|
|
|
|
finally:
|
|
if self.observer:
|
|
self.observer.stop()
|
|
self.observer.join()
|
|
|
|
if self.current_process:
|
|
self.current_process.terminate()
|
|
self.current_process.wait()
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python textual_dev_server.py <app_path> [watch_dir1] [watch_dir2]...")
|
|
sys.exit(1)
|
|
|
|
app_path = sys.argv[1]
|
|
watch_dirs = sys.argv[2:] if len(sys.argv) > 2 else None
|
|
|
|
server = TextualDevServer(app_path, watch_dirs)
|
|
server.run()
|
|
|
|
if __name__ == "__main__":
|
|
main() |