Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/IPython/utils/io.py: 30%

1# encoding: utf-8 

2""" 

3IO related utilities. 

4""" 

5 

6# Copyright (c) IPython Development Team. 

7# Distributed under the terms of the Modified BSD License. 

8 

9 

10 

11import atexit 

12import os 

13import sys 

14import tempfile 

15from pathlib import Path 

16from warnings import warn 

17 

18from IPython.utils.decorators import undoc 

19from .capture import CapturedIO, capture_output 

20 

21class Tee(object): 

22 """A class to duplicate an output stream to stdout/err. 

23 

24 This works in a manner very similar to the Unix 'tee' command. 

25 

26 When the object is closed or deleted, it closes the original file given to 

27 it for duplication. 

28 """ 

29 # Inspired by: 

30 # http://mail.python.org/pipermail/python-list/2007-May/442737.html 

31 

32 def __init__(self, file_or_name, mode="w", channel='stdout'): 

33 """Construct a new Tee object. 

34 

35 Parameters 

36 ---------- 

37 file_or_name : filename or open filehandle (writable) 

38 File that will be duplicated 

39 mode : optional, valid mode for open(). 

40 If a filename was give, open with this mode. 

41 channel : str, one of ['stdout', 'stderr'] 

42 """ 

43 if channel not in ['stdout', 'stderr']: 

44 raise ValueError('Invalid channel spec %s' % channel) 

45 

46 if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'): 

47 self.file = file_or_name 

48 else: 

49 encoding = None if "b" in mode else "utf-8" 

50 self.file = open(file_or_name, mode, encoding=encoding) 

51 self.channel = channel 

52 self.ostream = getattr(sys, channel) 

53 setattr(sys, channel, self) 

54 self._closed = False 

55 

56 def close(self): 

57 """Close the file and restore the channel.""" 

58 self.flush() 

59 setattr(sys, self.channel, self.ostream) 

60 self.file.close() 

61 self._closed = True 

62 

63 def write(self, data): 

64 """Write data to both channels.""" 

65 self.file.write(data) 

66 self.ostream.write(data) 

67 self.ostream.flush() 

68 

69 def flush(self): 

70 """Flush both channels.""" 

71 self.file.flush() 

72 self.ostream.flush() 

73 

74 def __del__(self): 

75 if not self._closed: 

76 self.close() 

77 

78 

79def ask_yes_no(prompt, default=None, interrupt=None): 

80 """Asks a question and returns a boolean (y/n) answer. 

81 

82 If default is given (one of 'y','n'), it is used if the user input is 

83 empty. If interrupt is given (one of 'y','n'), it is used if the user 

84 presses Ctrl-C. Otherwise the question is repeated until an answer is 

85 given. 

86 

87 An EOF is treated as the default answer. If there is no default, an 

88 exception is raised to prevent infinite loops. 

89 

90 Valid answers are: y/yes/n/no (match is not case sensitive).""" 

91 

92 answers = {'y':True,'n':False,'yes':True,'no':False} 

93 ans = None 

94 while ans not in answers.keys(): 

95 try: 

96 ans = input(prompt+' ').lower() 

97 if not ans: # response was an empty string 

98 ans = default 

99 except KeyboardInterrupt: 

100 if interrupt: 

101 ans = interrupt 

102 print("\r") 

103 except EOFError: 

104 if default in answers.keys(): 

105 ans = default 

106 print() 

107 else: 

108 raise 

109 

110 return answers[ans] 

111 

112 

113def temp_pyfile(src, ext='.py'): 

114 """Make a temporary python file, return filename and filehandle. 

115 

116 Parameters 

117 ---------- 

118 src : string or list of strings (no need for ending newlines if list) 

119 Source code to be written to the file. 

120 ext : optional, string 

121 Extension for the generated file. 

122 

123 Returns 

124 ------- 

125 (filename, open filehandle) 

126 It is the caller's responsibility to close the open file and unlink it. 

127 """ 

128 fname = tempfile.mkstemp(ext)[1] 

129 with open(Path(fname), "w", encoding="utf-8") as f: 

130 f.write(src) 

131 f.flush() 

132 return fname 

133 

134 

135@undoc 

136def raw_print(*args, **kw): 

137 """DEPRECATED: Raw print to sys.__stdout__, otherwise identical interface to print().""" 

138 warn("IPython.utils.io.raw_print has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2) 

139 

140 print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'), 

141 file=sys.__stdout__) 

142 sys.__stdout__.flush() 

143 

144@undoc 

145def raw_print_err(*args, **kw): 

146 """DEPRECATED: Raw print to sys.__stderr__, otherwise identical interface to print().""" 

147 warn("IPython.utils.io.raw_print_err has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2) 

148 

149 print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'), 

150 file=sys.__stderr__) 

151 sys.__stderr__.flush()