lock stdin

lock_stdin

lock_stdin() is a context manager on the driver that gives you exclusive ownership of stdin for the duration of the block.

What it does

On entry it:

1. Pauses the driver's stdin reader thread (cooperative pause via threading.Condition, ≤ ~100 ms to acknowledge).
2. Disables all terminal event reporting that Textual enables (like mouse tracking, focus tracking, kitty key protocol, and bracketed paste) so no weird escape sequences arrive while you hold the lock.
3. Waits 50 ms for any already-in-transit events to land in the OS buffer before yielding, so a drain at the top of the block reliably clears them.

On exit, event reporting is re-enabled and the input thread resumes. Nesting is supported, though not recommended; the disable/re-enable only happens on the outermost entry and exit.

Notes

• lock_stdin() waits up to 0.5 s for the input thread to acknowledge the pause. If called before the input thread starts (or after it stops) it proceeds immediately.
• Plain key events cannot be disabled via escape sequences. Drain stdin at the start of the block to discard any buffered keypresses, the terminal response arrives in < 1 ms, before the user can type another character.
• lock_stdin() pauses Textual's stdin reader and silences terminal events, but does not restore the terminal to cooked/canonical mode. If your subprocess needs line-buffered input (e.g. input() or a shell), call the app.suspend() context manager.

Example — querying the terminal

import os
import select
import sys
from textual.app import ComposeResult
from textual.widgets import Button
from textual_drivers import DrivenApp

class MyApp(DrivenApp):
    def compose(self) -> ComposeResult:
        yield Button("Query terminal")

    def on_button_pressed(self) -> None:
        self.run_worker(self._query, thread=True)

    def _query(self) -> None:
        fd = sys.stdin.fileno()
        with self.app._driver.lock_stdin():
            # Drain buffered keypresses, then read the terminal's response directly.
            while select.select([fd], [], [], 0)[0]:
                os.read(fd, 4096)
            self.app._driver.write("\x1b[c")   # Primary Device Attributes query
            self.app._driver.flush()
            # Read until the response terminator 'c' arrives (< 1 ms round trip).
            buf = b""
            while b"c" not in buf:
                buf += os.read(fd, 4096)