Fix line ending handling in reverse_readfile/readline across OS, and not skipping empty lines

.github/workflows/test.yml

@@ -6,6 +6,7 @@ jobs:
   build:
     strategy:
       max-parallel: 20
+      fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-14, windows-latest]
         python-version: ["3.9", "3.x"]
@@ -29,6 +30,6 @@ jobs:
         run: pytest --cov=monty --cov-report html:coverage_reports tests
 
       - name: Upload coverage reports to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         env:
           CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

src/monty/io.py

@@ -9,20 +9,21 @@
 import errno
 import gzip
 import io
-
-try:
-    import lzma
-except ImportError:
-    lzma = None  # type: ignore[assignment]
 import mmap
 import os
 import subprocess
 import time
+import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+try:
+    import lzma
+except ImportError:
+    lzma = None  # type: ignore[assignment]
+
 if TYPE_CHECKING:
-    from typing import IO, Generator, Union
+    from typing import IO, Iterator, Literal, Union
 
 
 def zopen(filename: Union[str, Path], *args, **kwargs) -> IO:
@@ -54,7 +55,53 @@ def zopen(filename: Union[str, Path], *args, **kwargs) -> IO:
     return open(filename, *args, **kwargs)
 
 
-def reverse_readfile(filename: Union[str, Path]) -> Generator[str, str, None]:
+def _get_line_ending(
+    file: str | Path | io.TextIOWrapper,
+) -> Literal["\r\n", "\n", "\r"]:
+    """Helper function to get line ending of a file.
+
+    This function assumes the file has a single consistent line ending.
+
+    Returns:
+        "\n": Unix line ending.
+        "\r\n": Windows line ending.
+        "\r": Classic MacOS line ending.
+
+    Raises:
+        ValueError: If line ending is unknown.
+
+    Warnings:
+        If file is empty, "\n" would be used as default.
+    """
+    if isinstance(file, (str, Path)):
+        with open(file, "rb") as f:
+            first_line = f.readline()
+    elif isinstance(file, io.TextIOWrapper):
+        first_line = file.buffer.readline()
+    elif isinstance(file, (gzip.GzipFile, bz2.BZ2File)):
+        first_line = file.readline()
+        file.seek(0)  # reset pointer
+    else:
+        raise TypeError(f"Unknown file type {type(file).__name__}")
+
+    # Return Unix "\n" line ending as default if file is empty
+    if not first_line:
+        warnings.warn("File empty, use default line ending \n.", stacklevel=2)
+        return "\n"
+
+    if first_line.endswith(b"\r\n"):
+        return "\r\n"
+    if first_line.endswith(b"\n"):
+        return "\n"
+    if first_line.endswith(b"\r"):
+        return "\r"
+
+    raise ValueError(f"Unknown line ending in line {repr(first_line)}.")
+
+
+def reverse_readfile(
+    filename: Union[str, Path],
+) -> Iterator[str]:
     """
     A much faster reverse read of file by using Python's mmap to generate a
     memory-mapped file. It is slower for very small files than
@@ -67,26 +114,32 @@ def reverse_readfile(filename: Union[str, Path]) -> Generator[str, str, None]:
     Yields:
         Lines from the file in reverse order.
     """
+    # Generate line ending
+    l_end = _get_line_ending(filename)
+
     try:
         with zopen(filename, "rb") as file:
             if isinstance(file, (gzip.GzipFile, bz2.BZ2File)):
                 for line in reversed(file.readlines()):
-                    yield line.decode("utf-8").rstrip(os.linesep)
+                    yield line.decode("utf-8").rstrip(l_end)
+
             else:
                 filemap = mmap.mmap(file.fileno(), 0, access=mmap.ACCESS_READ)
                 n = len(filemap)
                 while n > 0:
-                    i = filemap.rfind(os.linesep.encode(), 0, n)
-                    yield filemap[i + 1 : n].decode("utf-8").rstrip(os.linesep)
+                    i = filemap.rfind(l_end.encode(), 0, n)
+                    yield filemap[i + 1 : n].decode("utf-8")
                     n = i
 
     except ValueError:
         return
 
 
 def reverse_readline(
-    m_file, blk_size: int = 4096, max_mem: int = 4000000
-) -> Generator[str, str, None]:
+    m_file,
+    blk_size: int = 4096,
+    max_mem: int = 4000000,
+) -> Iterator[str]:
     """
     Generator function to read a file line-by-line, but backwards.
     This allows one to efficiently get data at the end of a file.
@@ -104,68 +157,75 @@ def reverse_readline(
 
     Args:
         m_file (File): File stream to read (backwards)
-        blk_size (int): The buffer size. Defaults to 4096.
+        blk_size (int): The buffer size in bytes. Defaults to 4096.
         max_mem (int): The maximum amount of memory to involve in this
             operation. This is used to determine when to reverse a file
             in-memory versus seeking portions of a file. For bz2 files,
             this sets the maximum block size.
 
-    Returns:
-        Generator that yields lines from the file. Behave similarly to the
-        file.readline() function, except the lines are returned from the back
-        of the file.
+    Yields:
+        Lines from the file. Behave similarly to the file.readline function,
+        except the lines are returned from the back of the file.
     """
-    # Check if the file stream is a bit stream or not
+    # Generate line ending
+    l_end = _get_line_ending(m_file)
+
+    # Check if the file stream is a buffered text stream
     is_text = isinstance(m_file, io.TextIOWrapper)
 
     try:
         file_size = os.path.getsize(m_file.name)
     except AttributeError:
-        # Bz2 files do not have name attribute. Just set file_size to above
-        # max_mem for now.
+        # Bz2 files do not have "name" attribute.
+        # Just set file_size to max_mem for now.
         file_size = max_mem + 1
 
-    # If the file size is within our desired RAM use, just reverse it in memory
-    # GZip files must use this method because there is no way to negative seek
+    # If the file size is within desired RAM limit, just reverse it in memory.
+    # GZip files must use this method because there is no way to negative seek.
     # For windows, we also read the whole file.
-    if file_size < max_mem or isinstance(m_file, gzip.GzipFile) or os.name == "nt":
+    if os.name == "nt" or file_size < max_mem or isinstance(m_file, gzip.GzipFile):
         for line in reversed(m_file.readlines()):
-            yield line.rstrip()
+            yield (
+                line.rstrip(l_end)
+                if isinstance(line, str)
+                else line.decode().rstrip(l_end)
+            )
+
     else:
         if isinstance(m_file, bz2.BZ2File):
-            # for bz2 files, seeks are expensive. It is therefore in our best
+            # For bz2 files, seeks are expensive. It is therefore in our best
             # interest to maximize the blk_size within limits of desired RAM
             # use.
             blk_size = min(max_mem, file_size)
 
         buf = ""
         m_file.seek(0, 2)
-        lastchar = m_file.read(1) if is_text else m_file.read(1).decode("utf-8")
+        last_char = m_file.read(1) if is_text else m_file.read(1).decode("utf-8")
 
-        trailing_newline = lastchar == os.linesep
+        trailing_newline = last_char == l_end
 
         while True:
-            newline_pos = buf.rfind(os.linesep)
+            newline_pos = buf.rfind(l_end)
             pos = m_file.tell()
             if newline_pos != -1:
                 # Found a newline
                 line = buf[newline_pos + 1 :]
                 buf = buf[:newline_pos]
                 if pos or newline_pos or trailing_newline:
-                    line += os.linesep
+                    line += l_end
                 yield line
 
             elif pos:
                 # Need to fill buffer
-                toread = min(blk_size, pos)
-                m_file.seek(pos - toread, 0)
+                to_read = min(blk_size, pos)
+                m_file.seek(pos - to_read, 0)
                 if is_text:
-                    buf = m_file.read(toread) + buf
+                    buf = m_file.read(to_read) + buf
                 else:
-                    buf = m_file.read(toread).decode("utf-8") + buf
-                m_file.seek(pos - toread, 0)
-                if pos == toread:
-                    buf = os.linesep + buf
+                    buf = m_file.read(to_read).decode("utf-8") + buf
+                m_file.seek(pos - to_read, 0)
+                if pos == to_read:
+                    buf = l_end + buf
 
             else:
                 # Start-of-file