New braille IPC APIs: BrailleMirror, DirectBrailleWindow, and named-pipe server

source/_brailleMirror.py

@@ -0,0 +1,204 @@
+# A part of NonVisual Desktop Access (NVDA)
+# Copyright (C) 2026 NV Access Limited, Pneuma Solutions
+# This file may be used under the terms of the GNU General Public License, version 2 or later, as modified by the NVDA license.
+# For full terms and any additional permissions, see the NVDA license file: https://github.com/nvaccess/nvda/blob/master/copying.txt
+
+"""BrailleMirror and DirectBrailleWindow public APIs, and their supporting infrastructure.
+
+Kept in a separate module to avoid further growth of braille.py (see #12772).
+Imported into braille at the bottom of that module so callers continue to access
+everything as ``braille.BrailleMirror`` etc.
+"""
+
+from __future__ import annotations
+
+import inputCore
+import winUser
+import wx
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+	import braille
+
+
+class BrailleMirror:
+	"""Abstract base class for a braille mirror.
+
+	A mirror intercepts every braille display update and can optionally influence the negotiated display width.
+	Both the physical display and all registered mirrors receive the same cells simultaneously; the mirror does **not** suppress the local display.
+	Register an instance with :func:`registerMirror` and remove it with :func:`unregisterMirror`.  To inject a gesture back into NVDA (e.g. a routing key received over a remote channel) use :func:`injectGesture`.
+	"""
+
+	def display(self, cells: list[int]) -> None:
+		"""Called with the full cell array on every display update.
+
+		:param cells: The braille cells written to the display.
+		"""
+
+	def numCells(self) -> int:
+		"""Return the number of cells this mirror can show.
+
+		Return 0 (the default) to have no effect on the negotiated display
+		width.  A positive value caps the display width used by
+		:data:`braille.filter_displayDimensions` to the smallest value across all registered mirrors and the physical display.
+		"""
+		return 0
+
+
+_registeredMirrors: list[BrailleMirror] = []
+
+
+def _mirrorPreWriteCells(cells: list[int], **kwargs) -> None:
+	for mirror in _registeredMirrors:
+		mirror.display(cells)
+
+
+def _mirrorFilterDisplayDimensions(value: braille.DisplayDimensions) -> braille.DisplayDimensions:
+	sizes = [m.numCells() for m in _registeredMirrors if m.numCells() > 0]
+	if not sizes:
+		return value
+	cap = min(sizes)
+	if cap >= value.numCols:
+		return value
+	return value._replace(numCols=cap)
+
+
+def registerMirror(mirror: BrailleMirror) -> None:
+	"""Register *mirror* to receive braille display updates.
+
+	:meth:`BrailleMirror.display` will be called on the main thread for every subsequent :meth:`braille.BrailleHandler._writeCells` call.  If *mirror* returns a positive value from :meth:`BrailleMirror.numCells`, it will also participate in display-width negotiation via :data:`braille.filter_displayDimensions`.
+	"""
+	import braille
+
+	if not _registeredMirrors:
+		braille.pre_writeCells.register(_mirrorPreWriteCells)
+		braille.filter_displayDimensions.register(_mirrorFilterDisplayDimensions)
+	_registeredMirrors.append(mirror)
+	if braille.handler:
+		braille.handler._refreshEnabled(block=True)
+
+
+def unregisterMirror(mirror: BrailleMirror) -> None:
+	"""Remove a previously registered mirror.
+
+	Safe to call even if *mirror* is not currently registered.
+	"""
+	import braille
+
+	try:
+		_registeredMirrors.remove(mirror)
+	except ValueError:
+		return
+	if not _registeredMirrors:
+		braille.pre_writeCells.unregister(_mirrorPreWriteCells)
+		braille.filter_displayDimensions.unregister(_mirrorFilterDisplayDimensions)
+	if braille.handler:
+		braille.handler._refreshEnabled(block=True)
+
+
+def injectGesture(gesture: braille.BrailleDisplayGesture) -> None:
+	"""Inject *gesture* into NVDA's input pipeline.
+
+	This is a thin wrapper around :func:`inputCore.manager.executeGesture` that silently swallows :class:`inputCore.NoInputGestureAction` so callers do not need to handle the common case where no script is bound.
+
+	Thread safety: must be called on the main thread, or scheduled via ``wx.CallAfter`` from a background thread.
+	"""
+	try:
+		inputCore.manager.executeGesture(gesture)
+	except inputCore.NoInputGestureAction:
+		pass
+
+
+class DirectBrailleWindow:
+	"""Take over braille output and input while a specific window has focus.
+
+	When the window identified by *hwnd* is the foreground window, NVDA's own braille rendering is suspended.  The application drives what appears on the physical display by calling :meth:`display`, and all braille gestures are forwarded to :meth:`onGesture` instead of being processed by NVDA.
+	When the window loses focus, normal NVDA rendering resumes automatically.
+
+	:param hwnd: The HWND of the window that triggers direct braille mode.
+	:param numCells: Advertised display width; 0 means use whatever NVDA provides.  A positive value caps the negotiated display width via :data:`braille.filter_displayDimensions`.
+	"""
+
+	def __init__(self, hwnd: int, numCells: int = 0) -> None:
+		self._hwnd = hwnd
+		self._numCells = numCells
+		self._active = False
+
+	def _isForeground(self) -> bool:
+		"""Return True if our registered window is currently in the foreground."""
+		fg = winUser.getForegroundWindow()
+		return fg == self._hwnd or winUser.isDescendantWindow(self._hwnd, fg)
+
+	def display(self, cells: list[int]) -> None:
+		"""Push *cells* to the physical braille display.
+
+		Has no effect if the registered window is not currently foreground or if no braille display is connected.
+
+		Thread safety: safe to call from any thread; the actual write is dispatched to the main thread via ``wx.CallAfter``.
+		"""
+		import braille
+
+		if braille.handler and self._isForeground():
+			wx.CallAfter(braille.handler._writeCells, cells)
+
+	def onGesture(self, gesture: braille.BrailleDisplayGesture) -> None:
+		"""Called when a braille gesture arrives while this window is active.
+
+		The default implementation does nothing, so gestures are suppressed. Subclass and override to handle them.
+
+		:param gesture: The braille gesture that was intercepted.
+		"""
+
+	def _handleDecideEnabled(self) -> bool:
+		return not self._isForeground()
+
+	def _handleDecideExecuteGesture(self, gesture: inputCore.InputGesture) -> bool:
+		import braille
+
+		if not self._isForeground():
+			return True
+		if isinstance(gesture, braille.BrailleDisplayGesture):
+			self.onGesture(gesture)
+			return False
+		return True
+
+	def _handleFilterDisplayDimensions(self, value: braille.DisplayDimensions) -> braille.DisplayDimensions:
+		if self._numCells <= 0 or not self._isForeground():
+			return value
+		if self._numCells >= value.numCols:
+			return value
+		return value._replace(numCols=self._numCells)
+
+	def activate(self) -> None:
+		"""Start intercepting braille output and input for the registered window.
+
+		Safe to call even if already active (subsequent calls are no-ops).
+		"""
+		if self._active:
+			return
+		import braille
+
+		self._active = True
+		braille.decide_enabled.register(self._handleDecideEnabled)
+		inputCore.decide_executeGesture.register(self._handleDecideExecuteGesture)
+		if self._numCells > 0:
+			braille.filter_displayDimensions.register(self._handleFilterDisplayDimensions)
+		if braille.handler:
+			braille.handler._refreshEnabled(block=True)
+
+	def deactivate(self) -> None:
+		"""Stop intercepting braille, restoring NVDA's normal rendering.
+
+		Safe to call even if not currently active.
+		"""
+		if not self._active:
+			return
+		import braille
+
+		self._active = False
+		braille.decide_enabled.unregister(self._handleDecideEnabled)
+		inputCore.decide_executeGesture.unregister(self._handleDecideExecuteGesture)
+		if self._numCells > 0:
+			braille.filter_displayDimensions.unregister(self._handleFilterDisplayDimensions)
+		if braille.handler:
+			braille.handler._refreshEnabled(block=True)

source/_remoteClient/localMachine.py

@@ -27,7 +27,6 @@
 import api
 import braille
 from config.registry import RegistryKey
-import inputCore
 import nvwave
 import speech
 import tones
@@ -126,9 +125,6 @@ def __init__(self) -> None:
 
 		self.receivingBraille = False
 
-		self._cachedSizes: list[int] | None = None
-		"""Cached braille display sizes from remote machines"""
-
 		self._showingLocalUiMessage: bool = False
 		"""Whether we're currently showing a `ui.message` while showing remote braille."""
 
@@ -257,41 +253,7 @@ def brailleInput(self, **kwargs: dict[str, Any]) -> None:
 		:param kwargs: Gesture parameters passed to BrailleInputGesture
 		:note: Silently ignores gestures that have no associated action.
 		"""
-		try:
-			inputCore.manager.executeGesture(input.BrailleInputGesture(**kwargs))
-		except inputCore.NoInputGestureAction:
-			pass
-
-	def setBrailleDisplaySize(self, sizes: list[int]) -> None:
-		"""Cache remote braille display sizes for size negotiation.
-
-		:param sizes: List of display sizes (cells) from remote machines
-		"""
-		self._cachedSizes = sizes
-
-	def _handleFilterDisplayDimensions(self, value: braille.DisplayDimensions) -> braille.DisplayDimensions:
-		"""Filter the local display dimensions based on remote display dimensions.
-
-		Determines the optimal display dimensions when sharing braille output by
-		finding the smallest positive width among local and remote displays.
-
-		.. note::
-			We can currently only support a single line of braille,
-			as sending display dimensions would require changing the Remote Access protocol.
-
-		:param value: Local display dimensions
-		:return: The negotiated display dimensions to use.
-		"""
-		if not self._cachedSizes:
-			# We cannot support multiline displays without breaking the Remote Access protocol,
-			# so always force numRows to 1.
-			return value._replace(numRows=1)
-		# There is no point storing the number of rows if we are always going to set it to 1.
-		sizes = self._cachedSizes + [value.numCols]
-		try:
-			return braille.DisplayDimensions(numRows=1, numCols=min(i for i in sizes if i > 0))
-		except ValueError:
-			return value._replace(numRows=1)
+		braille.injectGesture(input.BrailleInputGesture(**kwargs))
 
 	def handleDecideEnabled(self) -> bool:
 		"""Determine if the local braille display should be enabled.

source/_remoteClient/session.py

@@ -271,6 +271,26 @@ def connectedClientsCount(self) -> int:
 		return self.connectedLeadersCount + self.connectedFollowersCount
 
 
+class _FollowerBrailleMirror(braille.BrailleMirror):
+	"""BrailleMirror that forwards display updates to connected leader machines.
+
+	Registered while at least one leader with a braille display is connected.
+	:meth:`numCells` returns the smallest positive leader display size so NVDA
+	negotiates a compatible display width.
+	"""
+
+	def __init__(self, session: "FollowerSession") -> None:
+		self._session = session
+
+	def display(self, cells: list[int]) -> None:
+		if self._session.hasBrailleLeaders():
+			self._session.transport.send(type=RemoteMessageType.DISPLAY, cells=cells)
+
+	def numCells(self) -> int:
+		sizes = [s for s in self._session.leaderDisplaySizes if s > 0]
+		return min(sizes) if sizes else 0
+
+
 class FollowerSession(RemoteSession):
 	"""Session that runs on the controlled (follower) NVDA instance.
 
@@ -303,6 +323,9 @@ def __init__(
 		self.leaders = defaultdict(dict)
 		self.leaderDisplaySizes = []
 		self.followers = set()
+		self._brailleMirror = _FollowerBrailleMirror(self)
+		# The remote protocol only supports single-row braille; force numRows to 1.
+		braille.filter_displayDimensions.register(self._filterNumRowsToOne)
 		self.transport.transportClosing.register(self.handleTransportClosing)
 		self.transport.registerInbound(
 			RemoteMessageType.CHANNEL_JOINED,
@@ -316,9 +339,6 @@ def __init__(
 			RemoteMessageType.SET_DISPLAY_SIZE,
 			self.setDisplaySize,
 		)
-		braille.filter_displayDimensions.register(
-			self.localMachine._handleFilterDisplayDimensions,
-		)
 		self.transport.registerInbound(
 			RemoteMessageType.BRAILLE_INPUT,
 			self.localMachine.brailleInput,
@@ -341,7 +361,7 @@ def registerCallbacks(self) -> None:
 		)
 		self.transport.registerOutbound(decide_playWaveFile, RemoteMessageType.WAVE)
 		self.transport.registerOutbound(post_speechPaused, RemoteMessageType.PAUSE_SPEECH)
-		braille.pre_writeCells.register(self.display)
+		braille.registerMirror(self._brailleMirror)
 		pre_speechQueued.register(self.sendSpeech)
 		self.callbacksAdded = True
 
@@ -352,7 +372,7 @@ def unregisterCallbacks(self) -> None:
 		self.transport.unregisterOutbound(RemoteMessageType.CANCEL)
 		self.transport.unregisterOutbound(RemoteMessageType.WAVE)
 		self.transport.unregisterOutbound(RemoteMessageType.PAUSE_SPEECH)
-		braille.pre_writeCells.unregister(self.display)
+		braille.unregisterMirror(self._brailleMirror)
 		pre_speechQueued.unregister(self.sendSpeech)
 		self.callbacksAdded = False
 
@@ -383,6 +403,7 @@ def handleTransportClosing(self) -> None:
 		to ensure clean shutdown of remote features.
 		"""
 		self.unregisterCallbacks()
+		braille.filter_displayDimensions.unregister(self._filterNumRowsToOne)
 
 	def handleTransportDisconnected(self) -> None:
 		"""Handle disconnection from the transport layer.
@@ -409,7 +430,14 @@ def setDisplaySize(self, sizes: list[int] | None = None) -> None:
 			sizes if sizes else [info.get("braille_numCells", 0) for info in self.leaders.values()]
 		)
 		log.debug(f"Setting follower display size to: {self.leaderDisplaySizes!r}")
-		self.localMachine.setBrailleDisplaySize(self.leaderDisplaySizes)
+
+	def _filterNumRowsToOne(self, value: braille.DisplayDimensions) -> braille.DisplayDimensions:
+		"""Force single-row braille output.
+
+		The remote protocol does not support multi-row displays,
+		so we always constrain numRows to 1.
+		"""
+		return value._replace(numRows=1)
 
 	def handleBrailleInfo(
 		self,
@@ -452,15 +480,6 @@ def pauseSpeech(self, switch: bool) -> None:
 		"""Toggle speech pause state on leader instances."""
 		self.transport.send(type=RemoteMessageType.PAUSE_SPEECH, switch=switch)
 
-	def display(self, cells: list[int]) -> None:
-		"""Forward braille display content to leader instances.
-
-		Only sends braille data if there are connected leaders with braille displays.
-		"""
-		# Only send braille data when there are controlling machines with a braille display
-		if self.hasBrailleLeaders():
-			self.transport.send(type=RemoteMessageType.DISPLAY, cells=cells)
-
 	def hasBrailleLeaders(self) -> bool:
 		"""Check if any connected leaders have braille displays.