diff --git a/README.md b/README.md index ee48cb5..405bf86 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ Rich print files — markdown, JSON, code, and more — beautifully in your term - **JSON formatting** with syntax highlighting - **Pager integration** with mouse scrolling support - **Stdin support** for piping content +- **Watch mode** for live file redraws ## Installation @@ -65,6 +66,15 @@ rp --no-pager file.py # Disable pager The pager prefers `less -R --mouse` for mouse scrolling when available, otherwise falls back to the system pager. +### Watch a file + +```bash +rp --watch README.md +rp --watch --type json data.txt +``` + +`--watch` is file-only. It does not support stdin or `-`, disables the pager, redraws the screen when the file changes, and keeps retrying if the file becomes missing or temporarily unreadable. + ### Show version ```bash diff --git a/src/rp/cli.py b/src/rp/cli.py index 4501c47..434d105 100644 --- a/src/rp/cli.py +++ b/src/rp/cli.py @@ -2,7 +2,9 @@ from __future__ import annotations +from collections.abc import Callable import sys +import time from pathlib import Path from typing import Annotated @@ -10,10 +12,12 @@ import typer from rich.console import Console from rp import __author__, __license__, __version__ -from rp.debug import env_debug_enabled, log_debug, make_debug_logger +from rp.debug import DebugLogger, env_debug_enabled, log_debug, make_debug_logger from rp.detect import detect_type from rp.render import RenderOptions, render +_WATCH_POLL_INTERVAL = 1.0 + app = typer.Typer( name="rp", help="Rich print files — markdown, JSON, code, and more — beautifully in your terminal.", @@ -36,6 +40,103 @@ def version_callback(value: bool) -> None: raise typer.Exit() +def _read_file_content(file: Path) -> str: + """Read a UTF-8 text file for rendering.""" + if not file.exists(): + raise FileNotFoundError(f"File not found: {file}") + if file.is_dir(): + raise IsADirectoryError(f"Is a directory: {file}") + + try: + return file.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"Cannot read binary file: {file}") from exc + + +def _render_content( + *, + console: Console, + debug: DebugLogger | None, + content: str, + file_path: str | None, + file_type: str | None, + guess_content: bool, + theme: str, + line_numbers: bool | None, + pager: bool | None, +) -> None: + """Detect type and render content with the standard pipeline.""" + detected_type = detect_type( + path=file_path, + content=content, + explicit_type=file_type, + guess_content=guess_content, + debug=debug, + ) + + options = RenderOptions( + theme=theme, + line_numbers=line_numbers, + pager=pager, + debug=debug, + ) + render(content, detected_type, console, options) + + +def _watch_file( + *, + file: Path, + console: Console, + debug: DebugLogger | None, + file_type: str | None, + guess_content: bool, + theme: str, + line_numbers: bool | None, + sleep: Callable[[float], None] | None = None, +) -> None: + """Poll a file and redraw when its visible state changes.""" + if sleep is None: + sleep = time.sleep + + previous_state: tuple[str, str] | None = None + + while True: + try: + content = _read_file_content(file) + except ( + FileNotFoundError, + IsADirectoryError, + PermissionError, + OSError, + ValueError, + ) as exc: + state = ("error", str(exc)) + if state != previous_state: + log_debug(debug, f"watch: detected error state {state[1]!r}") + console.clear(home=True) + console.print(f"[red]Error:[/red] {state[1]}") + previous_state = state + else: + state = ("content", content) + if state != previous_state: + log_debug(debug, f"watch: rendering update for {str(file)!r}") + console.clear(home=True) + _render_content( + console=console, + debug=debug, + content=content, + file_path=str(file), + file_type=file_type, + guess_content=guess_content, + theme=theme, + line_numbers=line_numbers, + pager=False, + ) + previous_state = state + + sleep(_WATCH_POLL_INTERVAL) + + @app.command() def main( file: Annotated[ @@ -66,6 +167,13 @@ def main( help="Use pager for output. Default: auto-detect.", ), ] = None, + watch: Annotated[ + bool, + typer.Option( + "--watch/--no-watch", + help="Watch a file for changes and redraw when it updates.", + ), + ] = False, guess_content: Annotated[ bool, typer.Option( @@ -92,12 +200,17 @@ def main( theme: Pygments color theme. line_numbers: Show line numbers. Default: auto. pager: Use pager for output. Default: auto-detect. + watch: Watch a file for changes and redraw when it updates. guess_content: Use heuristic content-based type detection as a fallback. version: Show version and exit. """ console = Console() debug = make_debug_logger(Console(stderr=True), env_debug_enabled()) + if watch and (file is None or str(file) == "-"): + console.print("[red]Error:[/red] --watch only supports file input.") + raise typer.Exit(code=1) + if file is None or str(file) == "-": log_debug(debug, "input: reading from stdin") if sys.stdin.isatty(): @@ -106,18 +219,28 @@ def main( file_path = None else: log_debug(debug, f"input: reading file {str(file)!r}") - if not file.exists(): - console.print(f"[red]Error:[/red] File not found: {file}") - raise typer.Exit(code=1) - if file.is_dir(): - console.print(f"[red]Error:[/red] Is a directory: {file}") - raise typer.Exit(code=1) + if watch: + log_debug(debug, f"watch: polling {str(file)!r}") + _watch_file( + file=file, + console=console, + debug=debug, + file_type=file_type, + guess_content=guess_content, + theme=theme, + line_numbers=line_numbers, + ) + raise typer.Exit() + try: - content = file.read_text(encoding="utf-8") - except UnicodeDecodeError: - console.print(f"[red]Error:[/red] Cannot read binary file: {file}") - raise typer.Exit(code=1) - except (PermissionError, OSError) as e: + content = _read_file_content(file) + except ( + FileNotFoundError, + IsADirectoryError, + PermissionError, + OSError, + ValueError, + ) as e: console.print(f"[red]Error:[/red] {e}") raise typer.Exit(code=1) file_path = str(file) @@ -126,21 +249,17 @@ def main( console.print("[dim]Empty input, nothing to display.[/dim]") raise typer.Exit() - file_type = detect_type( - path=file_path, - content=content, - explicit_type=file_type, - guess_content=guess_content, + _render_content( + console=console, debug=debug, - ) - - options = RenderOptions( + content=content, + file_path=file_path, + file_type=file_type, + guess_content=guess_content, theme=theme, line_numbers=line_numbers, pager=pager, - debug=debug, ) - render(content, file_type, console, options) if __name__ == "__main__": diff --git a/tests/test_cli.py b/tests/test_cli.py index ea94f46..7ee4d12 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5,6 +5,7 @@ from __future__ import annotations from pathlib import Path from unittest.mock import patch +import pytest from typer.testing import CliRunner from rp import __version__ @@ -171,6 +172,18 @@ class TestStdinInput: assert result.exit_code == 0 + def test_watch_rejects_stdin(self) -> None: + result = runner.invoke(app, ["--watch"], input="hello\n") + + assert result.exit_code == 1 + assert "--watch only supports file input" in result.output + + def test_watch_rejects_dash_input(self) -> None: + result = runner.invoke(app, ["-", "--watch"], input="hello\n") + + assert result.exit_code == 1 + assert "--watch only supports file input" in result.output + def test_stdin_passes_none_path(self) -> None: """When reading from stdin, detect_type should receive path=None.""" with patch("rp.cli.detect_type") as mock_detect_type: @@ -311,3 +324,96 @@ class TestLineNumbersOption: result = runner.invoke(app, [str(file), "--no-line-numbers", "--no-pager"]) assert result.exit_code == 0 + + +class TestWatchMode: + """Tests for the watch mode loop.""" + + def test_watch_disables_pager(self, tmp_file) -> None: + file = tmp_file("x = 1\n", suffix=".py") + + with patch("rp.cli.detect_type", return_value="python"): + with patch("rp.cli.time.sleep", side_effect=KeyboardInterrupt): + with patch("rp.cli.render") as mock_render: + with pytest.raises(KeyboardInterrupt): + main(file=file, pager=True, watch=True) + + call_args = mock_render.call_args + assert call_args is not None + options = call_args[0][3] + assert options.pager is False + + def test_watch_rerenders_on_file_change(self, tmp_file) -> None: + file = tmp_file("x = 1\n", suffix=".py") + sleep_calls = 0 + + def fake_sleep(_: float) -> None: + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 1: + file.write_text("x = 2\n", encoding="utf-8") + return + raise KeyboardInterrupt + + with patch("rp.cli.detect_type", return_value="python"): + with patch("rp.cli.time.sleep", side_effect=fake_sleep): + with patch("rp.cli.render") as mock_render: + with pytest.raises(KeyboardInterrupt): + main(file=file, watch=True) + + assert [call.args[0] for call in mock_render.call_args_list] == [ + "x = 1\n", + "x = 2\n", + ] + + def test_watch_retries_after_missing_file_and_recovers(self, tmp_file) -> None: + file = tmp_file("x = 1\n", suffix=".py") + sleep_calls = 0 + + def fake_sleep(_: float) -> None: + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 1: + file.unlink() + return + if sleep_calls == 2: + file.write_text("x = 3\n", encoding="utf-8") + return + raise KeyboardInterrupt + + with patch("rp.cli.detect_type", return_value="python"): + with patch("rp.cli.time.sleep", side_effect=fake_sleep): + with patch("rp.cli.Console.print") as mock_print: + with patch("rp.cli.render") as mock_render: + with pytest.raises(KeyboardInterrupt): + main(file=file, watch=True) + + assert [call.args[0] for call in mock_render.call_args_list] == [ + "x = 1\n", + "x = 3\n", + ] + mock_print.assert_any_call(f"[red]Error:[/red] File not found: {file}") + + def test_watch_recovers_when_file_is_missing_at_startup( + self, tmp_path: Path + ) -> None: + file = tmp_path / "startup.py" + sleep_calls = 0 + + def fake_sleep(_: float) -> None: + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls == 1: + file.write_text("x = 4\n", encoding="utf-8") + return + raise KeyboardInterrupt + + with patch("rp.cli.detect_type", return_value="python"): + with patch("rp.cli.time.sleep", side_effect=fake_sleep): + with patch("rp.cli.Console.print") as mock_print: + with patch("rp.cli.render") as mock_render: + with pytest.raises(KeyboardInterrupt): + main(file=file, watch=True) + + assert [call.args[0] for call in mock_render.call_args_list] == ["x = 4\n"] + mock_print.assert_any_call(f"[red]Error:[/red] File not found: {file}")