no_gui = 1
---- start : []
PyMOL>fragment ala
 Executive: object "ala" created.
---- fragment : []
PyMOL>print("hi")
hi
---- print : []
PyMOL>count_atoms all
 count_atoms: 10 atoms
---- count : []

sys.stdout after install : <module 'pcatch'> is pcatch: True
---- fragment ala   : ['PyMOL>fragment ala', ' Executive: object "ala" created.']
---- print via do   : ['PyMOL>print("hello-from-do")', 'hello-from-do']
---- bare print     : ['hello-from-bare-print']
---- bad command    : ['PyMOL>this_is_a_bad_command', "NameError: name 'this_is_a_bad_command' is not defined"]
---- count_atoms    : ['PyMOL>count_atoms all', ' count_atoms: 10 atoms']
---- bad load       : ['PyMOL>load /nonexistent/file.pdb', ' Error: failed to open file "/nonexistent/file.pdb"']
---- help fragment  : ['PyMOL>help fragment', ' ', 'DESCRIPTION', ' ', '    "fragment" retrieves a 3D structure from the fragment library,', ...]
---- iterate        : ['PyMOL>iterate first all, print(name, resn)', 'N ALA', ' Iterate: iterated over 1 atoms.']

---- fragment (no pcatch) : ['PyMOL>fragment ala', ' Executive: object "ala" created.']
stdout is pcatch: True
---- print (with pcatch)  : ['PyMOL>print("hi-nonsingleton")']      <-- 'hi-nonsingleton' GONE
---- bare print           : []                                       <-- GONE
---- count_atoms          : ['PyMOL>count_atoms all']                <-- ' count_atoms: 10 atoms' GONE

consume1: ['PyMOL>print("LINE-A")', 'LINE-A']
consume2: []
consume3: []
empty type: list []

### AFTER mute
C-origin queue:  ['PyMOL>fragment gly', ' Executive: object "gly" created.']   # still queued
py-origin queue: ['PyMOL>print("py-after-mute")']                              # 'py-after-mute' LOST

version ('3.2.0a', 3.0, 3000000, 1785422035, '159ed88...', 0)
atoms 10
ray 0.006s
png bytes 2645
idle() 0
getRedisplay 1
refresh() ok
scene ok (1.0, 0.0, 0.0)
reshape ok (580, 582)      <-- see below
movie ok frame 5
STOPPED OK

viewport: (800, 600)
feedback: ['PyMOL>fragment ala', ' Executive: object "ala" created.',
           'PyMOL>print("still-captured")', 'still-captured',
           'PyMOL>nope_nope', "NameError: name 'nope_nope' is not defined"]

get_feedback under -cq: []
fd-captured lines: 8
  | PyMOL>fragment ala
  |  Executive: object "ala" created.
  | PyMOL>count_atoms all
  |  count_atoms: 10 atoms
  | PyMOL>print("py-print")
  | py-print
  | PyMOL>bogus_cmd
  | NameError: name 'bogus_cmd' is not defined

"""
Headless PyMOL console-feedback capture for the web client bridge.

Verified against pymol 3.2.0a0 (git 159ed88), CPython 3.13.3, macOS arm64.

Why this shape (all empirically established, see docs/spikes/feedback.md):

  * packages/engine/layer1/Ortho.cpp:492-499 -- OrthoFeedbackIn() only queues when G->Option->pmgui.
    pmgui is derived from `not invocation.options.no_gui` (packages/engine/layer1/P.cpp:1820).
    `pymol -c` sets no_gui=1 (packages/engine/modules/pymol/invocation.py:401), so a "-c" bridge gets
    _get_feedback() == [] forever.  ==> we must NOT use -c; we set no_gui=0 and instead
    suppress the GUI by never creating a GL context / never calling _draw.

  * C-level messages (Executive/Selector/Ray/Setting/Error) reach the queue on their own.
    Python-level output (print, tracebacks, ' count_atoms: N atoms', ' Error: Unknown color.')
    only reaches it if the built-in `pcatch` module is installed as sys.stdout/sys.stderr.

  * pcatch writes via SingletonPyMOLGlobals (packages/engine/layer1/P.cpp:2667). With a NON-singleton
    pymol2.PyMOL() that pointer is null and pcatch silently DISCARDS every print().
    ==> must use pymol2.SingletonPyMOL (this is exactly what pmg_qt does:
    packages/engine/modules/pmg_qt/pymol_gl_widget.py:3,99-105).

  * cmd._get_feedback() is a DESTRUCTIVE read (packages/engine/modules/pymol/internal.py:593 loops
    _cmd.get_feedback until empty).  Exactly one consumer may exist in the process, and
    that consumer must own the scrollback.  It returns None (not []) when the API lock
    is busy -- observed once in ~100 polls during a busy render.
"""

import itertools
import re
import sys
import threading
import time
from collections import deque

__all__ = ["FeedbackCapture", "classify"]


# ---------------------------------------------------------------- classification
_RE_ERROR = re.compile(
    r"^(?:"
    r"\s*Error:"                      # " Error: Unknown color."
    r"|\s*\w+-Error:"                 # " Selector-Error:", " ScenePNG-Error:"
    r"|Traceback \(most recent call last\):"
    r"|\s*\w*(?:Error|Exception):"    # "NameError: ...", "ZeroDivisionError: ..."
    r")"
)
_RE_WARNING = re.compile(r"^\s*(?:\w+-)?Warning:")
_RE_PROMPT = re.compile(r"^PyMOL>")


def classify(line: str) -> str:
    """'prompt' | 'error' | 'warning' | 'info' -- prefixes verified against real output."""
    if _RE_PROMPT.match(line):
        return "prompt"
    if _RE_ERROR.match(line):
        return "error"
    if _RE_WARNING.match(line):
        return "warning"
    return "info"


# ---------------------------------------------------------------- capture
class FeedbackCapture:
    """Owns the single PyMOL instance + the single feedback consumer.

    Usage:
        fc = FeedbackCapture()
        fc.start()                       # configures options, starts PyMOL, installs pcatch
        cmd = fc.cmd
        cmd.do('fragment ala')
        seq, lines = fc.since(0)         # lines == [{'seq':int,'kind':str,'text':str}, ...]
        fc.stop()
    """

    def __init__(self, ring_size=20000, poll_interval=0.05, splash=False):
        self.ring_size = ring_size
        self.poll_interval = poll_interval
        self.splash = splash
        self._buf = deque(maxlen=ring_size)
        self._seq = itertools.count(1)
        self._lock = threading.Lock()
        self._cv = threading.Condition(self._lock)
        self._thread = None
        self._stop = threading.Event()
        self._installed = False
        self._real_stdout = None
        self._real_stderr = None
        self.pymol = None
        self.cmd = None
        self.lock_busy_count = 0

    # -- lifecycle ---------------------------------------------------------
    def configure_options(self):
        """MUST run before SingletonPyMOL().start()."""
        import pymol

        o = pymol.invocation.options
        o.no_gui = 0            # => G->Option->pmgui = 1 => OrthoFeedbackIn queues
        o.internal_gui = 0      # otherwise steals 220 px of viewport width
        o.internal_feedback = 0 # otherwise steals viewport height
        o.external_gui = 0
        o.show_splash = 1 if self.splash else 0
        o.read_stdin = 0
        o.sigint_handler = 0    # a server owns its own signal handling
        return o

    def start(self):
        import pymol2

        self.configure_options()
        self.pymol = pymol2.SingletonPyMOL()
        self.pymol.start()
        self.cmd = self.pymol.cmd
        self._install_pcatch()
        self._drain_once()  # absorb splash/banner
        self._thread = threading.Thread(
            target=self._poll_loop, name="pymol-feedback", daemon=True)
        self._thread.start()
        return self

    def _install_pcatch(self):
        import pcatch

        self._real_stdout, self._real_stderr = sys.stdout, sys.stderr
        pcatch._install()  # sets sys.stderr = sys.stdout = pcatch
        self._installed = True

    def _uninstall_pcatch(self):
        # CRITICAL: pcatch after p.stop() drops everything on the floor
        # (SingletonPyMOLGlobals is nulled in packages/engine/layer5/PyMOL.cpp:2075-2076).
        if self._installed:
            sys.stdout, sys.stderr = self._real_stdout, self._real_stderr
            self._installed = False

    def stop(self):
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)
        self._drain_once()
        self._uninstall_pcatch()
        if self.pymol is not None:
            self.pymol.stop()
            self.pymol = None

    # -- server-side logging that must NOT land in the PyMOL console -------
    @property
    def server_stdout(self):
        """Real process stdout, for the web server's own logs."""
        return self._real_stdout or sys.__stdout__

    # -- consumer ----------------------------------------------------------
    def _drain_once(self):
        fb = self.cmd._get_feedback()
        if fb is None:            # API lock busy; caller should retry
            self.lock_busy_count += 1
            return 0
        if not fb:
            return 0
        with self._cv:
            for text in fb:
                self._buf.append({
                    "seq": next(self._seq),
                    "kind": classify(text),
                    "text": text,
                })
            self._cv.notify_all()
        return len(fb)

    def _poll_loop(self):
        while not self._stop.is_set():
            try:
                self._drain_once()
            except Exception:  # never let the poller die
                pass
            self._stop.wait(self.poll_interval)

    def pump(self):
        """Synchronous drain, for callers that do not want the poller thread."""
        return self._drain_once()

    def since(self, seq):
        """Return (last_seq, [entries with seq > `seq`]).  Non-destructive; safe for
        reconnecting/multiple web clients because we, not PyMOL, own the scrollback."""
        with self._lock:
            out = [e for e in self._buf if e["seq"] > seq]
            last = self._buf[-1]["seq"] if self._buf else seq
        return last, out

    def wait_for(self, seq, timeout=25.0):
        """Long-poll / SSE helper: block until something newer than `seq` exists."""
        deadline = time.time() + timeout
        with self._cv:
            while True:
                if self._buf and self._buf[-1]["seq"] > seq:
                    break
                remaining = deadline - time.time()
                if remaining <= 0:
                    break
                self._cv.wait(remaining)
        return self.since(seq)

last_seq = 18  n = 18
  prompt   PyMOL>fragment ala
  info      Executive: object "ala" created.
  prompt   PyMOL>count_atoms all
  info      count_atoms: 10 atoms
  prompt   PyMOL>print("hello world")
  info     hello world
  prompt   PyMOL>color notacolor, all
  error     Error: Unknown color.
  prompt   PyMOL>nope_nope
  error    NameError: name 'nope_nope' is not defined
  prompt   PyMOL>set colored_feedback, 1
  info      Setting: colored_feedback set to on.
  warning   Setting-Warning: colored_feedback is not supported in Open-Source version of PyMOL
  prompt   PyMOL>show surface
  prompt   PyMOL>ray 200,150
  info      Ray: render time: 0.01 sec. = 496612.2 frames/hour (0.01 sec. accum.).
  prompt   PyMOL>png /root/nope.png
  error     ScenePNG-Error: error writing "/root/nope.png"! Please check directory...
--- non-destructive replay (client reconnect) ---
replay n = 18 identical: True
--- incremental ---
since(last) n = 0
wait_for got: ['PyMOL>fragment gly', ' Executive: object "gly" created.']
lock_busy_count = 0
stdout after stop is real: True
OK

this is a server log line, must NOT be swallowed
...
stdout restored, this must appear on the real terminal