Fix CLI and platform-specific bugs

__main__.py: pass the configured widgets to the bar (E2), track
progress in bytes in line mode (E1), suppress BrokenPipeError for
early-closing pipes (E3), keep known-size mode for empty files (E8).
posix.py: getch() falls back to a plain read when stdin is not a tty
(E6). windows.py: explicit argtypes on all kernel32 bindings, invalid
console handles are detected and skipped, getch() honors the number
of events read and survives non-ASCII keys (E4, E5). base.py: remove
the Python 2 era __cmp__/__nonzero__ dead code (E7).

Author: wolph

progressbar/__main__.py

@@ -342,8 +342,8 @@ def main(argv: list[str] | None = None) -> None:  # noqa: C901
 
         # Initialize the progress bar
         bar = progressbar.ProgressBar(
-            # widgets=widgets,
-            max_value=total_size or None,
+            widgets=widgets,
+            max_value=total_size if filesize_available else None,
             max_error=False,
         )
 
@@ -354,7 +354,7 @@ def main(argv: list[str] | None = None) -> None:  # noqa: C901
         total_transferred = 0
 
         bar.start()
-        with contextlib.suppress(KeyboardInterrupt):
+        with contextlib.suppress(KeyboardInterrupt, BrokenPipeError):
             for input_path in input_paths:
                 if isinstance(input_path, pathlib.Path):
                     input_stream = stack.enter_context(
@@ -374,7 +374,18 @@ def main(argv: list[str] | None = None) -> None:  # noqa: C901
                         break
 
                     output_stream.write(data)
-                    total_transferred += len(data)
+                    if isinstance(data, str):
+                        # The total size is measured in bytes, so progress
+                        # must be tracked in bytes as well
+                        encoding = (
+                            getattr(input_stream, 'encoding', None) or 'utf-8'
+                        )
+                        total_transferred += len(
+                            data.encode(encoding, errors='replace'),
+                        )
+                    else:
+                        total_transferred += len(data)
+
                     bar.update(total_transferred)
 
         bar.finish(dirty=True)

progressbar/base.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import typing
 from typing import IO, TextIO
 
 
@@ -9,12 +8,6 @@ class FalseMeta(type):
     def __bool__(cls) -> bool:  # pragma: no cover
         return False
 
-    @classmethod
-    def __cmp__(cls, other: typing.Any) -> int:  # pragma: no cover
-        return -1
-
-    __nonzero__ = __bool__
-
 
 class UnknownLength(metaclass=FalseMeta):
     pass

progressbar/terminal/os_specific/posix.py

@@ -4,6 +4,10 @@
 
 
 def getch() -> str:
+    if not sys.stdin.isatty():
+        # Raw mode is unavailable (and unnecessary) without a tty
+        return sys.stdin.read(1)
+
     fd = sys.stdin.fileno()
     old_settings = termios.tcgetattr(fd)  # type: ignore
     try:

progressbar/terminal/os_specific/windows.py

@@ -25,6 +25,13 @@
 
 _STD_INPUT_HANDLE = _DWORD(-10)
 _STD_OUTPUT_HANDLE = _DWORD(-11)
+# GetStdHandle returns INVALID_HANDLE_VALUE (-1) when no console is
+# attached (piped output, pythonw, services)
+_INVALID_HANDLE_VALUE = _HANDLE(-1).value
+
+
+def _valid_handle(handle) -> bool:
+    return handle is not None and handle != _INVALID_HANDLE_VALUE
 
 
 class WindowsConsoleModeFlags(enum.IntFlag):
@@ -48,25 +55,33 @@ def __str__(self) -> str:
         return f'{self.name} (0x{self.value:04X})'
 
 
+# Explicit argtypes are required: without them ctypes passes arguments
+# as 32-bit C ints, silently truncating 64-bit HANDLE values
 _GetConsoleMode = _kernel32.GetConsoleMode
+_GetConsoleMode.argtypes = (_HANDLE, ctypes.POINTER(_DWORD))
 _GetConsoleMode.restype = _BOOL
 
 _SetConsoleMode = _kernel32.SetConsoleMode
+_SetConsoleMode.argtypes = (_HANDLE, _DWORD)
 _SetConsoleMode.restype = _BOOL
 
 _GetStdHandle = _kernel32.GetStdHandle
+_GetStdHandle.argtypes = (_DWORD,)
 _GetStdHandle.restype = _HANDLE
 
-_ReadConsoleInput = _kernel32.ReadConsoleInputA
-_ReadConsoleInput.restype = _BOOL
+_SetConsoleTextAttribute = _kernel32.SetConsoleTextAttribute
+_SetConsoleTextAttribute.argtypes = (_HANDLE, _WORD)
+_SetConsoleTextAttribute.restype = _BOOL
 
 _h_console_input = _GetStdHandle(_STD_INPUT_HANDLE)
 _input_mode = _DWORD()
-_GetConsoleMode(_HANDLE(_h_console_input), ctypes.byref(_input_mode))
+if _valid_handle(_h_console_input):
+    _GetConsoleMode(_HANDLE(_h_console_input), ctypes.byref(_input_mode))
 
 _h_console_output = _GetStdHandle(_STD_OUTPUT_HANDLE)
 _output_mode = _DWORD()
-_GetConsoleMode(_HANDLE(_h_console_output), ctypes.byref(_output_mode))
+if _valid_handle(_h_console_output):
+    _GetConsoleMode(_HANDLE(_h_console_output), ctypes.byref(_output_mode))
 
 
 class _COORD(ctypes.Structure):
@@ -121,17 +136,34 @@ class _Event(ctypes.Union):
     _fields_ = (('EventType', _WORD), ('Event', _Event))
 
 
+_ReadConsoleInput = _kernel32.ReadConsoleInputA
+_ReadConsoleInput.argtypes = (
+    _HANDLE,
+    ctypes.POINTER(_INPUT_RECORD),
+    _DWORD,
+    ctypes.POINTER(_DWORD),
+)
+_ReadConsoleInput.restype = _BOOL
+
+
 def reset_console_mode() -> None:
-    _SetConsoleMode(_HANDLE(_h_console_input), _DWORD(_input_mode.value))
-    _SetConsoleMode(_HANDLE(_h_console_output), _DWORD(_output_mode.value))
+    if _valid_handle(_h_console_input):
+        _SetConsoleMode(_HANDLE(_h_console_input), _DWORD(_input_mode.value))
+
+    if _valid_handle(_h_console_output):
+        _SetConsoleMode(_HANDLE(_h_console_output), _DWORD(_output_mode.value))
 
 
 def set_console_mode() -> bool:
-    mode = (
-        _input_mode.value
-        | WindowsConsoleModeFlags.ENABLE_VIRTUAL_TERMINAL_INPUT
-    )
-    _SetConsoleMode(_HANDLE(_h_console_input), _DWORD(mode))
+    if not _valid_handle(_h_console_output):
+        return False
+
+    if _valid_handle(_h_console_input):
+        mode = (
+            _input_mode.value
+            | WindowsConsoleModeFlags.ENABLE_VIRTUAL_TERMINAL_INPUT
+        )
+        _SetConsoleMode(_HANDLE(_h_console_input), _DWORD(mode))
 
     mode = (
         _output_mode.value
@@ -146,7 +178,8 @@ def get_console_mode() -> int:
 
 
 def set_text_color(color) -> None:
-    _kernel32.SetConsoleTextAttribute(_h_console_output, color)
+    if _valid_handle(_h_console_output):
+        _SetConsoleTextAttribute(_HANDLE(_h_console_output), _WORD(color))
 
 
 def print_color(text, color) -> None:
@@ -156,19 +189,29 @@ def print_color(text, color) -> None:
 
 
 def getch():
+    if not _valid_handle(_h_console_input):
+        return None
+
     lp_buffer = (_INPUT_RECORD * 2)()
     n_length = _DWORD(2)
     lp_number_of_events_read = _DWORD()
 
-    _ReadConsoleInput(
+    if not _ReadConsoleInput(
         _HANDLE(_h_console_input),
         lp_buffer,
         n_length,
         ctypes.byref(lp_number_of_events_read),
-    )
-
-    char = lp_buffer[1].Event.KeyEvent.uChar.AsciiChar.decode('ascii')
-    if char == '\x00':
+    ):
         return None
 
-    return char
+    # Only the records that were actually read contain valid data, and
+    # non-ASCII keys must not crash the decode
+    for i in range(min(lp_number_of_events_read.value, len(lp_buffer))):
+        char = lp_buffer[i].Event.KeyEvent.uChar.AsciiChar.decode(
+            'ascii',
+            errors='replace',
+        )
+        if char != '\x00':
+            return char
+
+    return None