1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
# encoding: utf-8
"""
IO related utilities.
"""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import atexit
import os
import sys
import tempfile
import warnings
from warnings import warn
from IPython.utils.decorators import undoc
from .capture import CapturedIO, capture_output
@undoc
class IOStream:
def __init__(self, stream, fallback=None):
warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
DeprecationWarning, stacklevel=2)
if not hasattr(stream,'write') or not hasattr(stream,'flush'):
if fallback is not None:
stream = fallback
else:
raise ValueError("fallback required, but not specified")
self.stream = stream
self._swrite = stream.write
# clone all methods not overridden:
def clone(meth):
return not hasattr(self, meth) and not meth.startswith('_')
for meth in filter(clone, dir(stream)):
try:
val = getattr(stream, meth)
except AttributeError:
pass
else:
setattr(self, meth, val)
def __repr__(self):
cls = self.__class__
tpl = '{mod}.{cls}({args})'
return tpl.format(mod=cls.__module__, cls=cls.__name__, args=self.stream)
def write(self,data):
warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
DeprecationWarning, stacklevel=2)
try:
self._swrite(data)
except:
try:
# print handles some unicode issues which may trip a plain
# write() call. Emulate write() by using an empty end
# argument.
print(data, end='', file=self.stream)
except:
# if we get here, something is seriously broken.
print('ERROR - failed to write data to stream:', self.stream,
file=sys.stderr)
def writelines(self, lines):
warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
DeprecationWarning, stacklevel=2)
if isinstance(lines, str):
lines = [lines]
for line in lines:
self.write(line)
# This class used to have a writeln method, but regular files and streams
# in Python don't have this method. We need to keep this completely
# compatible so we removed it.
@property
def closed(self):
return self.stream.closed
def close(self):
pass
# setup stdin/stdout/stderr to sys.stdin/sys.stdout/sys.stderr
devnull = open(os.devnull, 'w')
atexit.register(devnull.close)
# io.std* are deprecated, but don't show our own deprecation warnings
# during initialization of the deprecated API.
with warnings.catch_warnings():
warnings.simplefilter('ignore', DeprecationWarning)
stdin = IOStream(sys.stdin, fallback=devnull)
stdout = IOStream(sys.stdout, fallback=devnull)
stderr = IOStream(sys.stderr, fallback=devnull)
class Tee(object):
"""A class to duplicate an output stream to stdout/err.
This works in a manner very similar to the Unix 'tee' command.
When the object is closed or deleted, it closes the original file given to
it for duplication.
"""
# Inspired by:
# http://mail.python.org/pipermail/python-list/2007-May/442737.html
def __init__(self, file_or_name, mode="w", channel='stdout'):
"""Construct a new Tee object.
Parameters
----------
file_or_name : filename or open filehandle (writable)
File that will be duplicated
mode : optional, valid mode for open().
If a filename was give, open with this mode.
channel : str, one of ['stdout', 'stderr']
"""
if channel not in ['stdout', 'stderr']:
raise ValueError('Invalid channel spec %s' % channel)
if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'):
self.file = file_or_name
else:
self.file = open(file_or_name, mode)
self.channel = channel
self.ostream = getattr(sys, channel)
setattr(sys, channel, self)
self._closed = False
def close(self):
"""Close the file and restore the channel."""
self.flush()
setattr(sys, self.channel, self.ostream)
self.file.close()
self._closed = True
def write(self, data):
"""Write data to both channels."""
self.file.write(data)
self.ostream.write(data)
self.ostream.flush()
def flush(self):
"""Flush both channels."""
self.file.flush()
self.ostream.flush()
def __del__(self):
if not self._closed:
self.close()
def ask_yes_no(prompt, default=None, interrupt=None):
"""Asks a question and returns a boolean (y/n) answer.
If default is given (one of 'y','n'), it is used if the user input is
empty. If interrupt is given (one of 'y','n'), it is used if the user
presses Ctrl-C. Otherwise the question is repeated until an answer is
given.
An EOF is treated as the default answer. If there is no default, an
exception is raised to prevent infinite loops.
Valid answers are: y/yes/n/no (match is not case sensitive)."""
answers = {'y':True,'n':False,'yes':True,'no':False}
ans = None
while ans not in answers.keys():
try:
ans = input(prompt+' ').lower()
if not ans: # response was an empty string
ans = default
except KeyboardInterrupt:
if interrupt:
ans = interrupt
print("\r")
except EOFError:
if default in answers.keys():
ans = default
print()
else:
raise
return answers[ans]
def temp_pyfile(src, ext='.py'):
"""Make a temporary python file, return filename and filehandle.
Parameters
----------
src : string or list of strings (no need for ending newlines if list)
Source code to be written to the file.
ext : optional, string
Extension for the generated file.
Returns
-------
(filename, open filehandle)
It is the caller's responsibility to close the open file and unlink it.
"""
fname = tempfile.mkstemp(ext)[1]
with open(fname,'w') as f:
f.write(src)
f.flush()
return fname
@undoc
def atomic_writing(*args, **kwargs):
"""DEPRECATED: moved to notebook.services.contents.fileio"""
warn("IPython.utils.io.atomic_writing has moved to notebook.services.contents.fileio since IPython 4.0", DeprecationWarning, stacklevel=2)
from notebook.services.contents.fileio import atomic_writing
return atomic_writing(*args, **kwargs)
@undoc
def raw_print(*args, **kw):
"""DEPRECATED: Raw print to sys.__stdout__, otherwise identical interface to print()."""
warn("IPython.utils.io.raw_print has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2)
print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
file=sys.__stdout__)
sys.__stdout__.flush()
@undoc
def raw_print_err(*args, **kw):
"""DEPRECATED: Raw print to sys.__stderr__, otherwise identical interface to print()."""
warn("IPython.utils.io.raw_print_err has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2)
print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
file=sys.__stderr__)
sys.__stderr__.flush()
# used by IPykernel <- 4.9. Removed during IPython 7-dev period and re-added
# Keep for a version or two then should remove
rprint = raw_print
rprinte = raw_print_err
@undoc
def unicode_std_stream(stream='stdout'):
"""DEPRECATED, moved to nbconvert.utils.io"""
warn("IPython.utils.io.unicode_std_stream has moved to nbconvert.utils.io since IPython 4.0", DeprecationWarning, stacklevel=2)
from nbconvert.utils.io import unicode_std_stream
return unicode_std_stream(stream)
|
