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

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 

21 

22class Tee: 

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

24 

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

26 

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

28 it for duplication. 

29 """ 

30 # Inspired by: 

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

32 

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

34 """Construct a new Tee object. 

35 

36 Parameters 

37 ---------- 

38 file_or_name : filename or open filehandle (writable) 

39 File that will be duplicated 

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

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

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

43 """ 

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

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

46 

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

48 self.file = file_or_name 

49 else: 

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

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

52 self.channel = channel 

53 self.ostream = getattr(sys, channel) 

54 setattr(sys, channel, self) 

55 self._closed = False 

56 

57 def close(self): 

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

59 self.flush() 

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

61 self.file.close() 

62 self._closed = True 

63 

64 def write(self, data): 

65 """Write data to both channels.""" 

66 self.file.write(data) 

67 self.ostream.write(data) 

68 self.ostream.flush() 

69 

70 def flush(self): 

71 """Flush both channels.""" 

72 self.file.flush() 

73 self.ostream.flush() 

74 

75 def __del__(self): 

76 if not self._closed: 

77 self.close() 

78 

79 def isatty(self): 

80 return False 

81 

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

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

84 

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

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

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

88 given. 

89 

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

91 exception is raised to prevent infinite loops. 

92 

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

94 

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

96 ans = None 

97 while ans not in answers.keys(): 

98 try: 

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

100 if not ans: # response was an empty string 

101 ans = default 

102 except KeyboardInterrupt: 

103 if interrupt: 

104 ans = interrupt 

105 print("\r") 

106 except EOFError: 

107 if default in answers.keys(): 

108 ans = default 

109 print() 

110 else: 

111 raise 

112 

113 return answers[ans] 

114 

115 

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

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

118 

119 Parameters 

120 ---------- 

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

122 Source code to be written to the file. 

123 ext : optional, string 

124 Extension for the generated file. 

125 

126 Returns 

127 ------- 

128 (filename, open filehandle) 

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

130 """ 

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

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

133 f.write(src) 

134 f.flush() 

135 return fname