Merge branch 'experimaestro:master' into master

Author: VictorMorand

src/experimaestro/tests/test_log_viewer.py

@@ -0,0 +1,85 @@
+"""Tests for the TUI log viewer's carriage-return handling"""
+
+from experimaestro.tui.log_viewer import LogFile, _apply_carriage_returns
+
+
+def test_apply_carriage_returns_keeps_last_segment():
+    assert _apply_carriage_returns(" 10%\r 20%\r 30%") == " 30%"
+
+
+def test_apply_carriage_returns_no_cr():
+    assert _apply_carriage_returns("hello") == "hello"
+
+
+def test_log_file_handles_tqdm_progress(tmp_path):
+    """A typical tqdm sequence should collapse into a single completed line."""
+    log = tmp_path / "task.out"
+    log.write_text("\rStep 0%\rStep 50%\rStep 100%\nDone\n")
+
+    reader = LogFile(str(log))
+    complete, partial = reader.read_tail()
+    assert complete == ["Step 100%", "Done"]
+    assert partial == ""
+
+
+def test_log_file_partial_line_buffering(tmp_path):
+    """Partial last line (no \\n) should be returned as partial, not as a full line."""
+    log = tmp_path / "task.out"
+    log.write_text("first\n\rprog 10%")
+
+    reader = LogFile(str(log))
+    complete, partial = reader.read_tail()
+    assert complete == ["first"]
+    assert partial == "prog 10%"
+
+
+def test_log_file_progress_across_reads(tmp_path):
+    """Progress updates split across reads should overwrite, not stack."""
+    log = tmp_path / "task.out"
+    log.write_text("start\n\rprog 10%")
+
+    reader = LogFile(str(log))
+    complete, partial = reader.read_tail()
+    assert complete == ["start"]
+    assert partial == "prog 10%"
+
+    # Append more progress updates and a final newline + next line
+    with open(log, "a") as f:
+        f.write("\rprog 50%\rprog 100%\nnext line\n")
+
+    complete, partial = reader.read_new_content()
+    assert complete == ["prog 100%", "next line"]
+    assert partial == ""
+
+
+def test_log_file_truncation_resets_partial(tmp_path):
+    log = tmp_path / "task.out"
+    log.write_text("start\n\rprog 10%")
+
+    reader = LogFile(str(log))
+    reader.read_tail()
+    assert reader._partial_line == "prog 10%"
+
+    # Truncate
+    log.write_text("")
+    complete, partial = reader.read_new_content()
+    assert complete == []
+    assert partial == ""
+    assert reader._partial_line == ""
+
+
+def test_log_file_partial_within_chunk(tmp_path):
+    """Multiple \\r updates inside one read chunk collapse to the last."""
+    log = tmp_path / "task.out"
+    log.write_text("")
+
+    reader = LogFile(str(log))
+    reader.read_tail()
+
+    # Write a chunk containing multiple \r updates and no trailing \n
+    with open(log, "a") as f:
+        f.write("\rA\rB\rC")
+
+    complete, partial = reader.read_new_content()
+    assert complete == []
+    assert partial == "C"

src/experimaestro/tools/documentation.py

@@ -14,12 +14,18 @@
 from docutils.nodes import document
 from docutils.parsers.rst import Directive, Parser, directives
 from docutils.utils import new_document
+from sphinx.directives.other import TocTree
+from sphinx.domains.python import PyCurrentModule
+from termcolor import cprint
+
+from experimaestro import Config
+from experimaestro.sphinx import PyObject
 
 
-# ``docutils.frontend.get_default_settings`` is the modern API (docutils >=
-# 0.19); older versions exposed the same through ``OptionParser``. Support
-# both so users aren't forced to a docutils floor just to import this
-# module.
+# ``docutils.frontend.get_default_settings`` is the modern API (docutils
+# >= 0.19); older versions exposed the same through ``OptionParser``.
+# Support both so users aren't forced to a docutils floor just to import
+# this module.
 try:
     from docutils.frontend import get_default_settings as _get_default_settings
 except ImportError:  # pragma: no cover - exercised only on old docutils
@@ -29,14 +35,6 @@ def _get_default_settings(*components):
         return _OptionParser(components=components).get_default_values()
 
 
-from sphinx.directives.other import TocTree
-from sphinx.domains.python import PyCurrentModule
-from termcolor import cprint
-
-from experimaestro import Config
-from experimaestro.sphinx import PyObject
-
-
 def documented_from_objects(objects_inv: Path) -> Set[str]:
     inv = sphobjinv.Inventory(objects_inv)
     return set(

src/experimaestro/tui/log_viewer.py

@@ -18,13 +18,27 @@
 INITIAL_TAIL_SIZE = 256 * 1024  # 256KB
 
 
+def _apply_carriage_returns(line: str) -> str:
+    """Collapse a line as a terminal would: keep only content after the last \\r.
+
+    tqdm-like progress bars rewrite the same line via \\r; treat that as
+    overwrite rather than as a new line.
+    """
+    if "\r" in line:
+        return line.rsplit("\r", 1)[-1]
+    return line
+
+
 class LogFile:
     """Efficient log file reader that tracks position and watches for changes"""
 
     def __init__(self, path: str):
         self.path = Path(path)
         self.position = 0
         self.size = 0
+        # Buffer for the last incomplete line (no trailing \n yet).
+        # Stored already collapsed for any \r updates seen so far.
+        self._partial_line = ""
         self._update_size()
 
     def _update_size(self) -> None:
@@ -34,17 +48,35 @@ def _update_size(self) -> None:
         except OSError:
             self.size = 0
 
-    def read_tail(self, max_bytes: int = INITIAL_TAIL_SIZE) -> str:
-        """Read the last N bytes of the file"""
+    def _process_content(self, content: str) -> tuple[list[str], str]:
+        """Split a chunk into completed lines plus a trailing partial line.
+
+        Carriage returns are handled like a terminal would: within each
+        \\n-terminated line, only the substring after the last \\r is kept.
+        """
+        combined = self._partial_line + content
+        parts = combined.split("\n")
+        # parts[-1] is the partial trailing line (empty if combined ended in \n)
+        complete = [_apply_carriage_returns(line) for line in parts[:-1]]
+        partial = _apply_carriage_returns(parts[-1])
+        self._partial_line = partial
+        return complete, partial
+
+    def read_tail(self, max_bytes: int = INITIAL_TAIL_SIZE) -> tuple[list[str], str]:
+        """Read the last N bytes of the file.
+
+        Returns ``(complete_lines, partial_line)``. ``partial_line`` is the
+        in-progress last line (no trailing \\n yet) and may be empty.
+        """
         if not self.path.exists():
-            return ""
+            return [], ""
 
         self._update_size()
         if self.size == 0:
-            return ""
+            return [], ""
 
         try:
-            with open(self.path, "r", errors="replace") as f:
+            with open(self.path, "r", errors="replace", newline="") as f:
                 # Start from max_bytes before end, or beginning
                 start_pos = max(0, self.size - max_bytes)
                 f.seek(start_pos)
@@ -55,32 +87,40 @@ def read_tail(self, max_bytes: int = INITIAL_TAIL_SIZE) -> str:
 
                 content = f.read()
                 self.position = f.tell()
-                return content
+                # Reset partial buffer since we're reading from a known boundary
+                self._partial_line = ""
+                return self._process_content(content)
         except Exception:
-            return ""
+            return [], ""
+
+    def read_new_content(self) -> tuple[list[str], str]:
+        """Read any new content since last read.
 
-    def read_new_content(self) -> str:
-        """Read any new content since last read"""
+        Returns ``(new_complete_lines, partial_line)``. ``partial_line`` is the
+        current in-progress last line (which may have changed even when no new
+        complete lines were produced).
+        """
         if not self.path.exists():
-            return ""
+            return [], self._partial_line
 
         self._update_size()
 
         # File was truncated or rotated
         if self.size < self.position:
             self.position = 0
+            self._partial_line = ""
 
         if self.position >= self.size:
-            return ""
+            return [], self._partial_line
 
         try:
-            with open(self.path, "r", errors="replace") as f:
+            with open(self.path, "r", errors="replace", newline="") as f:
                 f.seek(self.position)
                 content = f.read()
                 self.position = f.tell()
-                return content
+                return self._process_content(content)
         except Exception:
-            return ""
+            return [], self._partial_line
 
     def has_new_content(self) -> bool:
         """Check if there's new content without reading it"""
@@ -96,29 +136,39 @@ def __init__(self, file_path: str, widget_id: str):
         self.file_path = file_path
         self.log_file = LogFile(file_path)
         self.following = True
+        self._last_partial = ""
 
     def compose(self) -> ComposeResult:
         yield Static(f"📄 {self.file_path}", classes="log-file-path")
         yield RichLog(id=f"{self.id}-content", wrap=True, highlight=True, markup=False)
+        yield Static("", id=f"{self.id}-partial", classes="log-partial")
 
-    def on_mount(self) -> None:
-        """Load initial content from tail of file"""
-        content = self.log_file.read_tail()
-        if content:
+    def _apply_update(self, complete_lines: list[str], partial: str) -> None:
+        if complete_lines:
             log_widget = self.query_one(f"#{self.id}-content", RichLog)
-            for line in content.splitlines():
+            for line in complete_lines:
                 log_widget.write(line)
+        if partial != self._last_partial:
+            partial_widget = self.query_one(f"#{self.id}-partial", Static)
+            partial_widget.update(partial)
+            partial_widget.display = bool(partial)
+            self._last_partial = partial
+
+    def on_mount(self) -> None:
+        """Load initial content from tail of file"""
+        complete_lines, partial = self.log_file.read_tail()
+        # Hide partial widget when there is nothing in progress
+        partial_widget = self.query_one(f"#{self.id}-partial", Static)
+        partial_widget.display = bool(partial)
+        self._apply_update(complete_lines, partial)
 
     def refresh_content(self) -> None:
         """Check for and append new content"""
         if not self.following:
             return
 
-        new_content = self.log_file.read_new_content()
-        if new_content:
-            log_widget = self.query_one(f"#{self.id}-content", RichLog)
-            for line in new_content.splitlines():
-                log_widget.write(line)
+        complete_lines, partial = self.log_file.read_new_content()
+        self._apply_update(complete_lines, partial)
 
     def scroll_to_end(self) -> None:
         """Scroll to end of log"""
@@ -175,6 +225,14 @@ class LogViewerScreen(Screen, inherit_bindings=False):
         height: 1fr;
         border: solid $primary;
     }
+
+    .log-partial {
+        background: $boost;
+        padding: 0 1;
+        height: auto;
+        color: $text;
+        text-style: italic;
+    }
     """
 
     BINDINGS = [