-import os
-
from threading import Lock
-from .utils import compat_os_name, get_windows_version
+from .utils import supports_terminal_sequences, TERMINAL_SEQUENCES
+
+class MultilinePrinterBase:
+ def __init__(self, stream=None, lines=1):
+ self.stream = stream
+ self.maximum = lines - 1
-class MultilinePrinterBase():
def __enter__(self):
return self
def end(self):
pass
+ def _add_line_number(self, text, line):
+ if self.maximum:
+ return f'{line + 1}: {text}'
+ return text
-class MultilinePrinter(MultilinePrinterBase):
- def __init__(self, stream, lines):
- """
- @param stream stream to write to
- @lines number of lines to be written
- """
- self.stream = stream
+class QuietMultilinePrinter(MultilinePrinterBase):
+ pass
- is_win10 = compat_os_name == 'nt' and get_windows_version() >= (10, )
- self.CARRIAGE_RETURN = '\r'
- if os.getenv('TERM') and self._isatty() or is_win10:
- # reason not to use curses https://github.com/yt-dlp/yt-dlp/pull/1036#discussion_r713851492
- # escape sequences for Win10 https://docs.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences
- self.UP = '\x1b[A'
- self.DOWN = '\n'
- self.ERASE_LINE = '\x1b[K'
- self._HAVE_FULLCAP = self._isatty() or is_win10
- else:
- self.UP = self.DOWN = self.ERASE_LINE = None
- self._HAVE_FULLCAP = False
- # lines are numbered from top to bottom, counting from 0 to self.maximum
- self.maximum = lines - 1
- self.lastline = 0
- self.lastlength = 0
+class MultilineLogger(MultilinePrinterBase):
+ def print_at_line(self, text, pos):
+ # stream is the logger object, not an actual stream
+ self.stream.debug(self._add_line_number(text, pos))
- self.movelock = Lock()
- @property
- def have_fullcap(self):
- """
- True if the TTY is allowing to control cursor,
- so that multiline progress works
- """
- return self._HAVE_FULLCAP
+class BreaklineStatusPrinter(MultilinePrinterBase):
+ def print_at_line(self, text, pos):
+ self.stream.write(self._add_line_number(text, pos) + '\n')
- def _isatty(self):
- try:
- return self.stream.isatty()
- except BaseException:
- return False
+
+class MultilinePrinter(MultilinePrinterBase):
+ def __init__(self, stream=None, lines=1, preserve_output=True):
+ super().__init__(stream, lines)
+ self.preserve_output = preserve_output
+ self._lastline = self._lastlength = 0
+ self._movelock = Lock()
+ self._HAVE_FULLCAP = supports_terminal_sequences(self.stream)
+
+ def lock(func):
+ def wrapper(self, *args, **kwargs):
+ with self._movelock:
+ return func(self, *args, **kwargs)
+ return wrapper
def _move_cursor(self, dest):
- current = min(self.lastline, self.maximum)
- self.stream.write(self.CARRIAGE_RETURN)
- if current == dest:
- # current and dest are at same position, no need to move cursor
+ current = min(self._lastline, self.maximum)
+ self.stream.write('\r')
+ distance = dest - current
+ if distance < 0:
+ self.stream.write(TERMINAL_SEQUENCES['UP'] * -distance)
+ elif distance > 0:
+ self.stream.write(TERMINAL_SEQUENCES['DOWN'] * distance)
+ self._lastline = dest
+
+ @lock
+ def print_at_line(self, text, pos):
+ if self._HAVE_FULLCAP:
+ self._move_cursor(pos)
+ self.stream.write(TERMINAL_SEQUENCES['ERASE_LINE'])
+ self.stream.write(text)
return
- elif current > dest:
- # when maximum == 2,
- # 0. dest
- # 1.
- # 2. current
- self.stream.write(self.UP * (current - dest))
- elif current < dest:
- # when maximum == 2,
- # 0. current
- # 1.
- # 2. dest
- self.stream.write(self.DOWN * (dest - current))
- self.lastline = dest
- def print_at_line(self, text, pos):
- with self.movelock:
- if self.have_fullcap:
- self._move_cursor(pos)
- self.stream.write(self.ERASE_LINE)
- self.stream.write(text)
- else:
- if self.maximum != 0:
- # let user know about which line is updating the status
- text = f'{pos + 1}: {text}'
- textlen = len(text)
- if self.lastline == pos:
- # move cursor at the start of progress when writing to same line
- self.stream.write(self.CARRIAGE_RETURN)
- if self.lastlength > textlen:
- text += ' ' * (self.lastlength - textlen)
- self.lastlength = textlen
- else:
- # otherwise, break the line
- self.stream.write('\n')
- self.lastlength = 0
- self.stream.write(text)
- self.lastline = pos
+ text = self._add_line_number(text, pos)
+ textlen = len(text)
+ if self._lastline == pos:
+ # move cursor at the start of progress when writing to same line
+ self.stream.write('\r')
+ if self._lastlength > textlen:
+ text += ' ' * (self._lastlength - textlen)
+ self._lastlength = textlen
+ else:
+ # otherwise, break the line
+ self.stream.write('\n')
+ self._lastlength = textlen
+ self.stream.write(text)
+ self._lastline = pos
+ @lock
def end(self):
- with self.movelock:
- # move cursor to the end of the last line, and write line break
- # so that other to_screen calls can precede
+ # move cursor to the end of the last line, and write line break
+ # so that other to_screen calls can precede
+ if self._HAVE_FULLCAP:
self._move_cursor(self.maximum)
+ if self.preserve_output:
self.stream.write('\n')
+ return
-
-class QuietMultilinePrinter(MultilinePrinterBase):
- def __init__(self):
- self.have_fullcap = True
-
-
-class BreaklineStatusPrinter(MultilinePrinterBase):
-
- def __init__(self, stream, lines):
- """
- @param stream stream to write to
- """
- self.stream = stream
- self.maximum = lines
- self.have_fullcap = True
-
- def print_at_line(self, text, pos):
- if self.maximum != 0:
- # let user know about which line is updating the status
- text = f'{pos + 1}: {text}'
- self.stream.write(text + '\n')
+ if self._HAVE_FULLCAP:
+ self.stream.write(
+ TERMINAL_SEQUENCES['ERASE_LINE']
+ + f'{TERMINAL_SEQUENCES["UP"]}{TERMINAL_SEQUENCES["ERASE_LINE"]}' * self.maximum)
+ else:
+ self.stream.write(' ' * self._lastlength)
projects / yt-dlp.git / blobdiff