How do I render the output from a subprocess in a Rich panel?

[Hubro]

I'm making a script that automates some C builds. Instead of the build output flushing the screen with garbage, I want to display the build output in a 15-20 line height panel (while also piping it to a log file) just to show that progress is happening.

Are there any tools currently available in Rich to achieve this?

Basically I need to:

• Create a panel
• Start a subprocess
• Somehow pipe the subprocess stdout and stderr inside the panel

First of all, how do I even write text into a panel? From the docs it seems like they can only display a static string, which doesn't seem very useful.

[willmcgugan]

Rich is designed for static output. If you want something more dynamic, consider textual

[Hubro]

Textual looks amazing, but I think it might be a bit overkill for my use case. It looks to me like Rich is already capable of everything I'm looking for, I'm just not sure how to wire everything together.

Essentially, all I need is to be able to update the text inside a panel several times per second, so I can show a window into the status of a build instead of flooding the terminal.

Is this something that can't be done with Rich?

[willmcgugan]

Of course it can. Have a look at rich.Live.

[Hubro]

Hey, awesome! That works.

def run_subprocess(
    title: str | None,
    subtitle: str | None,
    log_path: Path,
    command: list[str],
) -> None:
    """Run a subprocess while displaying the output in a temporary box with Rich

    Also logs the subprocess output to a file for later examination.
    """

    box_height = min([console.height // 4, 12])
    box_inner_height = box_height - 2  # The panel has 1 row of padding on each side
    box_width = console.width - 4  # The panel has 2 cols of padding on each side

    buffer = deque(maxlen=box_inner_height)

    def process_panel() -> rich.panel.Panel:
        return rich.panel.Panel(
            "\n".join(buffer), height=box_height, title=title, subtitle=subtitle
        )

    with log_path.open("w") as log_file, rich.live.Live(
        get_renderable=process_panel, refresh_per_second=30, transient=True
    ):
        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            encoding="UTF-8",
        )

        assert process.stdout

        try:
            for line in process.stdout:
                log_file.write(line)
                log_file.flush()
                buffer.append(line[:box_width].rstrip())
        except KeyboardInterrupt:
            process.terminate()
            raise click.Abort()
        finally:
            # Always wait for the process to terminate, or we might fail later
            # cleanup steps
            process.wait()

recording.mp4

My only issue now is that when I press Ctrl+C to interrupt, this inserts a line break into the terminal, which causes the top line of the live view to remain after the context manager exits (I'm using transient=True).

Is there any way to chomp that line away?

---

EDIT: Seems this also happens in the official live demo:

Anything I can do about this? 🤔

[trentonyo]

Hey, awesome! That works.

def run_subprocess(
    title: str | None,
    subtitle: str | None,
    log_path: Path,
    command: list[str],
) -> None:
    """Run a subprocess while displaying the output in a temporary box with Rich

    Also logs the subprocess output to a file for later examination.
    """

    box_height = min([console.height // 4, 12])
    box_inner_height = box_height - 2  # The panel has 1 row of padding on each side
    box_width = console.width - 4  # The panel has 2 cols of padding on each side

    buffer = deque(maxlen=box_inner_height)

    def process_panel() -> rich.panel.Panel:
        return rich.panel.Panel(
            "\n".join(buffer), height=box_height, title=title, subtitle=subtitle
        )

    with log_path.open("w") as log_file, rich.live.Live(
        get_renderable=process_panel, refresh_per_second=30, transient=True
    ):
        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            encoding="UTF-8",
        )

        assert process.stdout

        try:
            for line in process.stdout:
                log_file.write(line)
                log_file.flush()
                buffer.append(line[:box_width].rstrip())
        except KeyboardInterrupt:
            process.terminate()
            raise click.Abort()
        finally:
            # Always wait for the process to terminate, or we might fail later
            # cleanup steps
            process.wait()

Hey I just wanted to say this is a super awesome snippet, I'd like to ask if I can use a modified version in my code?

[Hubro]

Help yourself 🙂