##// END OF EJS Templates
Fix bug when network map is disconnected (#13925)...
Fix bug when network map is disconnected (#13925) Hi there :) I encounter a little bug with the ipython logger. When using the `%logstart` magic and writing the log to a remote path, everything works fine until the network map gets disconnected. After the network drive is disconnected you basically powerless, and you have to reopen the console. This can be easily reproduce by sharing a windows folder ![image](https://user-images.githubusercontent.com/23289491/216761685-1bbcc92a-13e4-409c-badf-513f610fef46.png) Mounting it using ![image](https://user-images.githubusercontent.com/23289491/216761695-8b413c89-b4f8-4423-b9c0-d68c2997af38.png) Start an IPython session, start a log with `%logstart Z:\log` and disconnect. Anything you will try to do from now on will not work, and you can't use the Session, and will result `OSError: [Errno 22] Invalid argument` I couldn't reproduce it in Linux environment, i guess it really depends on the `OS` here. Also weirdly this happens when the client close the connection. that's mean if the Server close the connection to the mount everything still working perfectly I added a `try except` block to allow you to stop the logger and continue using your ipython normally I Search for a way to test it but I couldn't find an easy way to reproduce it in test.

File last commit:

r23969:d73e5f74
r28082:defefe64 merge
Show More
capture.py
170 lines | 5.0 KiB | text/x-python | PythonLexer
# encoding: utf-8
"""IO capturing utilities."""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import sys
from io import StringIO
#-----------------------------------------------------------------------------
# Classes and functions
#-----------------------------------------------------------------------------
class RichOutput(object):
def __init__(self, data=None, metadata=None, transient=None, update=False):
self.data = data or {}
self.metadata = metadata or {}
self.transient = transient or {}
self.update = update
def display(self):
from IPython.display import publish_display_data
publish_display_data(data=self.data, metadata=self.metadata,
transient=self.transient, update=self.update)
def _repr_mime_(self, mime):
if mime not in self.data:
return
data = self.data[mime]
if mime in self.metadata:
return data, self.metadata[mime]
else:
return data
def _repr_mimebundle_(self, include=None, exclude=None):
return self.data, self.metadata
def _repr_html_(self):
return self._repr_mime_("text/html")
def _repr_latex_(self):
return self._repr_mime_("text/latex")
def _repr_json_(self):
return self._repr_mime_("application/json")
def _repr_javascript_(self):
return self._repr_mime_("application/javascript")
def _repr_png_(self):
return self._repr_mime_("image/png")
def _repr_jpeg_(self):
return self._repr_mime_("image/jpeg")
def _repr_svg_(self):
return self._repr_mime_("image/svg+xml")
class CapturedIO(object):
"""Simple object for containing captured stdout/err and rich display StringIO objects
Each instance `c` has three attributes:
- ``c.stdout`` : standard output as a string
- ``c.stderr`` : standard error as a string
- ``c.outputs``: a list of rich display outputs
Additionally, there's a ``c.show()`` method which will print all of the
above in the same order, and can be invoked simply via ``c()``.
"""
def __init__(self, stdout, stderr, outputs=None):
self._stdout = stdout
self._stderr = stderr
if outputs is None:
outputs = []
self._outputs = outputs
def __str__(self):
return self.stdout
@property
def stdout(self):
"Captured standard output"
if not self._stdout:
return ''
return self._stdout.getvalue()
@property
def stderr(self):
"Captured standard error"
if not self._stderr:
return ''
return self._stderr.getvalue()
@property
def outputs(self):
"""A list of the captured rich display outputs, if any.
If you have a CapturedIO object ``c``, these can be displayed in IPython
using::
from IPython.display import display
for o in c.outputs:
display(o)
"""
return [ RichOutput(**kargs) for kargs in self._outputs ]
def show(self):
"""write my output to sys.stdout/err as appropriate"""
sys.stdout.write(self.stdout)
sys.stderr.write(self.stderr)
sys.stdout.flush()
sys.stderr.flush()
for kargs in self._outputs:
RichOutput(**kargs).display()
__call__ = show
class capture_output(object):
"""context manager for capturing stdout/err"""
stdout = True
stderr = True
display = True
def __init__(self, stdout=True, stderr=True, display=True):
self.stdout = stdout
self.stderr = stderr
self.display = display
self.shell = None
def __enter__(self):
from IPython.core.getipython import get_ipython
from IPython.core.displaypub import CapturingDisplayPublisher
from IPython.core.displayhook import CapturingDisplayHook
self.sys_stdout = sys.stdout
self.sys_stderr = sys.stderr
if self.display:
self.shell = get_ipython()
if self.shell is None:
self.save_display_pub = None
self.display = False
stdout = stderr = outputs = None
if self.stdout:
stdout = sys.stdout = StringIO()
if self.stderr:
stderr = sys.stderr = StringIO()
if self.display:
self.save_display_pub = self.shell.display_pub
self.shell.display_pub = CapturingDisplayPublisher()
outputs = self.shell.display_pub.outputs
self.save_display_hook = sys.displayhook
sys.displayhook = CapturingDisplayHook(shell=self.shell,
outputs=outputs)
return CapturedIO(stdout, stderr, outputs)
def __exit__(self, exc_type, exc_value, traceback):
sys.stdout = self.sys_stdout
sys.stderr = self.sys_stderr
if self.display and self.shell:
self.shell.display_pub = self.save_display_pub
sys.displayhook = self.save_display_hook