Files
p5_bounce/dev_server.py
2026-06-30 22:28:37 +03:00

116 lines
3.4 KiB
Python

#!/usr/bin/env python3
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from io import BytesIO
from pathlib import Path
import argparse
import errno
import os
import time
ROOT = Path(__file__).resolve().parent
WATCH_EXTENSIONS = {".html", ".css", ".js"}
RELOAD_SNIPPET = """
<script>
(() => {
const events = new EventSource("/__hotreload");
events.addEventListener("reload", () => window.location.reload());
})();
</script>
"""
class HotReloadHandler(SimpleHTTPRequestHandler):
def end_headers(self):
self.send_header("Cache-Control", "no-store")
super().end_headers()
def do_GET(self):
if self.path == "/__hotreload":
self.handle_hotreload()
return
super().do_GET()
def translate_path(self, path):
clean_path = path.split("?", 1)[0].split("#", 1)[0].lstrip("/")
return str(ROOT / clean_path)
def send_head(self):
path = Path(self.translate_path(self.path))
if path.is_dir():
path = path / "index.html"
if path.suffix == ".html" and path.exists():
html = path.read_text(encoding="utf-8")
html = html.replace("</body>", f"{RELOAD_SNIPPET}</body>")
body = html.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
return BytesIO(body)
return super().send_head()
def handle_hotreload(self):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "keep-alive")
self.end_headers()
last_seen = latest_mtime()
while True:
time.sleep(0.5)
current = latest_mtime()
if current > last_seen:
last_seen = current
try:
self.wfile.write(b"event: reload\ndata: changed\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
break
def latest_mtime():
latest = 0.0
for directory, names, files in os.walk(ROOT):
if ".git" in names:
names.remove(".git")
for filename in files:
path = Path(directory) / filename
if path.suffix in WATCH_EXTENSIONS:
latest = max(latest, path.stat().st_mtime)
return latest
def main():
parser = argparse.ArgumentParser(description="Run a tiny p5 dev server with hot reload.")
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()
server, port = create_server(args.host, args.port)
print(f"Serving http://{args.host}:{port}", flush=True)
print("Watching .html, .css, and .js files for changes", flush=True)
server.serve_forever()
def create_server(host, preferred_port):
for port in range(preferred_port, preferred_port + 20):
try:
return ThreadingHTTPServer((host, port), HotReloadHandler), port
except OSError as error:
if error.errno != errno.EADDRINUSE:
raise
raise OSError(f"No available port found from {preferred_port} to {preferred_port + 19}")
if __name__ == "__main__":
main()