bridge.py

import json
import threading
import queue

class FrontendBridge:
    def __init__(self):
        self.window = None
        self._window_ready = threading.Event()
        self._frontend_ready = threading.Event()
        self._dispatch_queue: queue.Queue[tuple[str, tuple[object, ...]]] = queue.Queue()
        self._feed_lock = threading.Lock()
        self._feed_buffer: list[str] = []
        self._feed_pending = False
        self._dispatch_thread = threading.Thread(
            target=self._dispatch_loop,
            name="project-center-ui-dispatch",
            daemon=True,
        )
        self._dispatch_thread.start()

    def attach_window(self, window):
        self.window = window
        self._window_ready.set()

    def mark_frontend_ready(self):
        self._frontend_ready.set()

    def call(self, name, *args):
        self._dispatch_queue.put((name, args))

    def feed(self, chunk):
        with self._feed_lock:
            self._feed_buffer.append(chunk)
            if self._feed_pending:
                return
            self._feed_pending = True
        self._dispatch_queue.put(("term_feed", ()))

    def _take_feed_payload(self):
        with self._feed_lock:
            payload = "".join(self._feed_buffer)
            self._feed_buffer.clear()
            self._feed_pending = False
        return payload

    def _dispatch_loop(self):
        while True:
            name, args = self._dispatch_queue.get()
            self._window_ready.wait()
            self._frontend_ready.wait()

            if self.window is None:
                continue

            if name == "term_feed":
                payload = (self._take_feed_payload(),)
                if not payload[0]:
                    continue
            else:
                payload = args

            script = (
                "window.ProjectCenterBridge && "
                f"window.ProjectCenterBridge.invoke({json.dumps(name)}, {json.dumps(payload)})"
            )

            try:
                self.window.evaluate_js(script)
            except Exception as exc:
                print(f"Error dispatching UI event '{name}': {exc}")

frontend = FrontendBridge()