Created
June 16, 2022 18:58
-
-
Save jrnold/bc252d3477a1c2493c1521c4c4a620da to your computer and use it in GitHub Desktop.
Context manager to temporarily set callbacks in rpy2
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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