Skip to content

Instantly share code, notes, and snippets.

@jrnold
Created June 16, 2022 18:58
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jrnold/bc252d3477a1c2493c1521c4c4a620da to your computer and use it in GitHub Desktop.
Save jrnold/bc252d3477a1c2493c1521c4c4a620da to your computer and use it in GitHub Desktop.
Context manager to temporarily set callbacks in rpy2
import rpy2.rinterface_lib.callbacks
from typing import Callable, Optional, Tuple
class RCallbackContext:
"""Context manager for setting R callbacks.
See https://rpy2.github.io/doc/v3.5.x/html/callbacks.html#console-i-o
Attributes:
consoleread: Function to use for :py:func:`rpy2.rinterface_lib.callbacks.consoleread`
consolewrite_print:
consolewrite_warnerror:
showmessage:
consoleflush:
yesnocancel:
showfiles:
processevents:
busy:
cleanup:
"""
_callbacks = [
'consoleread',
'consolewrite_print',
'consolewrite_warnerror',
'showmessage',
'consoleflush',
'yesnocancel',
'showfiles',
'processevents',
'busy',
'cleanup'
]
def __init__(self,
consoleread: Optional[Callable[[str], str]] = None,
consolewrite_print: Optional[Callable[[str], None]] = None,
consolewrite_warnerror: Optional[Callable[[str], None]] = None,
showmessage: Optional[Callable[[str], None]] = None,
consoleflush: Optional[Callable[[], None]] = None,
yesnocancel: Optional[Callable[[str], None]] = None,
showfiles: Optional[Callable[[Tuple[str, ...], Tuple[str, ...], Optional[str], Optional[str]], None]] = None,
processevents: Optional[Callable[[], None]] = None,
busy: Optional[Callable[[int], None]] = None,
cleanup: Optional[Callable[..., Optional[int]]] = None
):
self._callback_cache = {}
# TODO: check validity of these
self.consoleread = consoleread
self.consolewrite_print = consolewrite_print
self.consolewrite_warnerror = consolewrite_warnerror
self.showmessage = showmessage
self.consoleflush = consoleflush
self.yesnocancel = yesnocancel
self.showfiles = showfiles
self.processevents = processevents
self.busy = busy
self.cleanup = cleanup
def __enter__(self):
# Cache current values of callbacks
for cb in self._callbacks:
self._callback_cache[cb] = getattr(rpy2.rinterface_lib.callbacks, cb)
# Set any new callbacks
for cb in self._callbacks:
setattr(rpy2.rinterface_lib.callbacks, cb, getattr(self, cb))
return self
def __exit__(self, *excdetails) -> None:
# Cache current values of callbacks
for cb in self._callbacks:
setattr(rpy2.rinterface_lib.callbacks, cb, self._callback_cache[cb])
def _console_donothing(s: str) -> None:
pass
def suppress_output(print = True, warnerror = False) -> RCallbackContext:
"""Create a context to suppress R output.
Args:
print: Suppress standard output
warnerror: Suppress warnings and errors
Returns:
An object of `RCallbackContext` to use with a ``with`` statement to
suppress R output.
"""
callbacks = {}
if print:
callbacks['consolewrite_print'] = _console_donothing
if warnerror:
callbacks['consolewrite_warnerror'] = _console_donothing
return RCallbackContext(**callbacks)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment