[core] Support emitting ConEmu progress codes

yt_dlp/YoutubeDL.py

@@ -639,20 +639,19 @@ def __init__(self, params=None, auto_init=True):
         self.cache = Cache(self)
         self.__header_cookies = []
 
+        try:
+            windows_enable_vt_mode()
+        except Exception as e:
+            self.write_debug(f'Failed to enable VT mode: {e}')
+
         stdout = sys.stderr if self.params.get('logtostderr') else sys.stdout
         self._out_files = Namespace(
             out=stdout,
             error=sys.stderr,
             screen=sys.stderr if self.params.get('quiet') else stdout,
-            console=None if compat_os_name == 'nt' else next(
+            console=next(filter(supports_terminal_sequences, (sys.stderr, sys.stdout)), None),
-                filter(supports_terminal_sequences, (sys.stderr, sys.stdout)), None),
         )
 
-        try:
-            windows_enable_vt_mode()
-        except Exception as e:
-            self.write_debug(f'Failed to enable VT mode: {e}')
-
         if self.params.get('no_color'):
             if self.params.get('color') is not None:
                 self.params.setdefault('_warnings', []).append(
@@ -953,21 +952,18 @@ def to_stderr(self, message, only_once=False):
             self._write_string(f'{self._bidi_workaround(message)}\n', self._out_files.error, only_once=only_once)
 
     def _send_console_code(self, code):
-        if compat_os_name == 'nt' or not self._out_files.console:
+        if not supports_terminal_sequences(self._out_files.console):
-            return
+            return False
         self._write_string(code, self._out_files.console)
+        return True
 
     def to_console_title(self, message):
-        if not self.params.get('consoletitle', False):
+        if not self.params.get('consoletitle'):
             return
         message = remove_terminal_sequences(message)
-        if compat_os_name == 'nt':
+        if not self._send_console_code(f'\033]0;{message}\007'):
-            if ctypes.windll.kernel32.GetConsoleWindow():
+            if os.name == 'nt' and ctypes.windll.kernel32.GetConsoleWindow():
-                # c_wchar_p() might not be necessary if `message` is
+                ctypes.windll.kernel32.SetConsoleTitleW(message)
-                # already of type unicode()
-                ctypes.windll.kernel32.SetConsoleTitleW(ctypes.c_wchar_p(message))
-        else:
-            self._send_console_code(f'\033]0;{message}\007')
 
     def save_console_title(self):
         if not self.params.get('consoletitle') or self.params.get('simulate'):
@@ -981,6 +977,9 @@ def restore_console_title(self):
 
     def __enter__(self):
         self.save_console_title()
+        if self.params.get('consoletitle'):
+            # Set progress bar to "indeterminate"
+            self._send_console_code('\033]9;4;3\007')
         return self
 
     def save_cookies(self):
@@ -989,6 +988,9 @@ def save_cookies(self):
 
     def __exit__(self, *args):
         self.restore_console_title()
+        if self.params.get('consoletitle'):
+            # Set progress bar to "disabled"
+            self._send_console_code('\033]9;4;0\007')
         self.close()
 
     def close(self):

yt_dlp/downloader/common.py

@@ -337,6 +337,15 @@ def _report_progress_status(self, s, default_template):
             progress_template.get('download-title') or 'yt-dlp %(progress._default_template)s',
             progress_dict))
 
+        percent = s.get('_percent')
+        if s['status'] not in ('downloading', 'error', 'finished') or percent is None:
+            return
+        if s['status'] == 'finished':
+            self.ydl._send_console_code('\033]9;4;3\007')
+            return
+        state = 1 if s['status'] == 'downloading' else 2
+        self.ydl._send_console_code(f'\033]9;4;{state};{int(percent)}\007')
+
     def _format_progress(self, *args, **kwargs):
         return self.ydl._format_text(
             self._multiline.stream, self._multiline.allow_colors, *args, **kwargs)
@@ -377,13 +386,15 @@ def with_fields(*tups, default=''):
                     return
                 self._progress_delta_time += update_delta
 
+        progress = try_call(
+            lambda: 100 * s['downloaded_bytes'] / s['total_bytes'],
+            lambda: 100 * s['downloaded_bytes'] / s['total_bytes_estimate'],
+            lambda: s['downloaded_bytes'] == 0 and 0)
         s.update({
             '_eta_str': self.format_eta(s.get('eta')).strip(),
             '_speed_str': self.format_speed(s.get('speed')),
-            '_percent_str': self.format_percent(try_call(
+            '_percent': progress,
-                lambda: 100 * s['downloaded_bytes'] / s['total_bytes'],
+            '_percent_str': self.format_percent(progress),
-                lambda: 100 * s['downloaded_bytes'] / s['total_bytes_estimate'],
-                lambda: s['downloaded_bytes'] == 0 and 0)),
             '_total_bytes_str': _format_bytes('total_bytes'),
             '_total_bytes_estimate_str': _format_bytes('total_bytes_estimate'),
             '_downloaded_bytes_str': _format_bytes('downloaded_bytes'),

yt_dlp/postprocessor/common.py

@@ -192,6 +192,15 @@ def report_progress(self, s):
             progress_template.get('postprocess-title') or 'yt-dlp %(progress._default_template)s',
             progress_dict))
 
+        percent = s.get('_percent')
+        if s['status'] not in ('downloading', 'error', 'finished') or percent is None:
+            return
+        if s['status'] == 'finished':
+            self._downloader._send_console_code('\033]9;4;3\007')
+            return
+        state = 1 if s['status'] == 'downloading' else 2
+        self._downloader._send_console_code(f'\033]9;4;{state};{int(percent)}\007')
+
     def _retry_download(self, err, count, retries):
         # While this is not an extractor, it behaves similar to one and
         # so obey extractor_retries and "--retry-sleep extractor"