Adding a GUI to a CLI built with rich

[Spectre5]

First, I obviously realize that rich is intended to be used in a terminal, not a GUI. That said, I have a CLI application that is built using rich and works great. Now, in a separate project, I'm trying to use the main function of that CLI (which uses a rich progress bar) in a GUI. Essentially I'm adding a GUI wrapper around the CLI. What I'd like to do is capture the progress bar to update the GUI. Below is a simplified example of what I'd like to do.

In the run method below I can call function_from_another_module just fine, which, of course, outputs to the terminal and doesn't update the GUI progress bar. I need a way to get the progress from this function --- is there a way to add a callback or something to the rich progress bar? I didn't see a way in the documentation.

I'd be great to also get all information from the progress bar, not only the % complete. For example, description, estimated time remaining, etc. That way I could add that stuff to the GUI as well.

import os
import sys
import threading
import time

from PySide6.QtGui import QIcon, Qt
from PySide6.QtWidgets import (
    QApplication,
    QLabel,
    QProgressBar,
    QVBoxLayout,
    QWidget,
)
from rich.progress import track


# Ideally I would make no changes to this function --- in the actual use case,
# this function is imported from another library that uses rich to display a
# progress bar in a CLI application.  Effectively, I'm trying to write a simple
# GUI wrapper to this CLI application, hopefully without making any change to
# the CLI application, so that it doesn't need to be aware of whether it is 
# running in the CLI or a GUI.  However, I did write the CLI, so if I need to
# add an optional keyword argument to provide a callback function or something,
# I can do that.  Or switch from using track to an actual Progress object.
def function_from_another_module(n=50):
    for n in track(range(n), description='Processing...'):
        time.sleep(0.1)


class WorkerThread(threading.Thread):
    def __init__(self, n, progress):
        threading.Thread.__init__(self)
        self.n = n
        self.progress = progress

    def run(self):
        function_from_another_module()
        # for i in range(self.n):
        #     self.progress.setValue((i + 1) / self.n * 100.0)
        #     time.sleep(0.1)


class MainWindow(QWidget):
    def __init__(self):
        QWidget.__init__(self)

        vbox = QVBoxLayout()
        label = QLabel('Processing..')
        label.setAlignment(Qt.AlignCenter)
        vbox.addWidget(label)

        self.progress_bar = QProgressBar()
        self.progress_bar.setAlignment(Qt.AlignCenter)
        vbox.addWidget(self.progress_bar)

        self.setLayout(vbox)
        self.setGeometry(300, 300, 300, 50)

    def closeEvent(self, event):
        os._exit(0)


if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    WorkerThread(n=50, progress=window.progress_bar).start()
    app.exec_()

[Arashkazemii]

Since you can change the function, an optional progress callback is the cleanest bridge. Report structured progress from the work itself, and let the terminal and GUI decide how to display it:

from rich.progress import Progress

def process(n=50, on_progress=None, terminal=True):
    with Progress(disable=not terminal) as progress:
        task_id = progress.add_task("Processing...", total=n)
        for index in range(n):
            do_one_item(index)
            progress.advance(task_id)
            task = progress.tasks[0]
            if on_progress is not None:
                on_progress({
                    "completed": task.completed,
                    "total": task.total,
                    "description": task.description,
                    "remaining_seconds": task.time_remaining,
                })

For the GUI, the callback should emit a Qt signal from the worker. Connect that signal to a slot on the main-window object, using a queued connection; update QProgressBar in that slot. Do not call a widget's setters directly from threading.Thread.run().

For example, a QThread subclass can declare updated = Signal(dict) and invoke process(on_progress=self.updated.emit, terminal=False) inside run(). The window's @Slot(dict) method can set the maximum to total and the value to completed. Keep the worker alive as an attribute of the window.

time_remaining can initially be None; show an unknown ETA until Rich has enough samples. Parsing the terminal's rendered progress output would be much more fragile than this callback.