Running external commands & processes

28. Running external commands &

processes

So far we’ve looked at how to run things in separate threads, including external programs using Python subprocess. But in PyQt6 we can also make use of a Qt- based system for running external programs, QProcess. Creating and executing a job with QProcess is relatively straightforward.

The simplest possible example is shown below — we create a QProcess object and then call .start passing in the command to execute and a list of string arguments. In this case we’re running our custom demo script, with Python python dummy_script.py.

p = QProcess()

p.start("python", ["dummy_script.py"])

ë

Depending on your environment, you may need to specify

python3 instead of python.

q

You need to keep a reference to the created QProcess instance,

either on self or elsewhere, while it is running.

The simple example is enough if you just want to run a program and don’t care what happens to it. However, if you want to know more about what a program is doing, QProcess provides a number of signals which can be used to track the progress and state of processes.

The most useful are the .readyReadStandardOutput and .readyReadStandardError which fire whenever there is standard output and standard error ready to be read from the process. All running processes have two streams of output — standard out and standard error. The standard output returns the actual result of the execution (if any) while standard error returns any error or

logging information.

p = QProcess()

p.readyReadStandardOutput.connect(self.handle_stdout)

p.readyReadStandardError.connect(self.handle_stderr)

p.stateChanged.connect(self.handle_state)

p.finished.connect(self.cleanup)

p.start("python", ["dummy_script.py"])

Additionally, there is a .finished signal which is fired when the process completes, and a .stateChanged signal which fires when the process status changes. Valid values — defined in the QProcess.ProcessState enum — are shown below.

Constant Value Description

QProcess.NotRunning 0 The process is not running.

QProcess.Starting 1 The process is starting, but the program

has not yet been invoked.

QProcess.Running 2 The process is running and is ready for

reading and writing.

In the following example we extend this basic QProcess setup to add handlers for the standard out and standard err. The signals notifying of available data connect to these handlers and trigger a request of the data from the process, using .readAllStandardError() and .readAllStandardOutput().

Z The methods output raw bytes, so you need to decode it first.

In this example, our demo script dummy_script.py return a series of strings, which are parsed to provide progress information and structured data. The state of the process is also displayed on the statusbar.

The full code is shown below —

Listing 217. concurrent/qprocess.py

import re

import sys

from PyQt6.QtCore import QProcess

from PyQt6.QtWidgets import (

QApplication,

QMainWindow,

QPlainTextEdit,

QProgressBar,

QPushButton,

QVBoxLayout,

QWidget,

)

STATES = {

QProcess.ProcessState.NotRunning: "Not running",

QProcess.ProcessState.Starting: "Starting...",

QProcess.ProcessState.Running: "Running...",

}

progress_re = re.compile("Total complete: (\d+)%")

def simple_percent_parser (output):

"""

Matches lines using the progress_re regex,

returning a single integer for the % progress.

"""

m = progress_re.search(output)

if m:

pc_complete = m.group( 1 )

return int(pc_complete)

def extract_vars (l):

"""

Extracts variables from lines, looking for lines

containing an equals, and splitting into key=value.

"""

data = {}

for s in l.splitlines():

if "=" in s: name, value = s.split("=") data[name] = value return data

class MainWindow (QMainWindow): def init (self): super().init()

Hold process reference.

self.p = None

layout = QVBoxLayout()

self.text = QPlainTextEdit() layout.addWidget(self.text)

self.progress = QProgressBar() layout.addWidget(self.progress)

btn_run = QPushButton("Execute") btn_run.clicked.connect(self.start)

layout.addWidget(btn_run)

w = QWidget() w.setLayout(layout) self.setCentralWidget(w)

self.show()

def start (self): if self.p is not None: return

self.p = QProcess() self.p.readyReadStandardOutput.connect(self.handle_stdout) self.p.readyReadStandardError.connect(self.handle_stderr) self.p.stateChanged.connect(self.handle_state) self.p.finished.connect(self.cleanup) self.p.start("python", ["dummy_script.py"])

def handle_stderr (self):

result = bytes(self.p.readAllStandardError()).decode("utf8")

progress = simple_percent_parser(result)

self.progress.setValue(progress)

def handle_stdout (self):

result = bytes(self.p.readAllStandardOutput()).decode("utf8")

data = extract_vars(result)

self.text.appendPlainText(str(data))

def handle_state (self, state):

self.statusBar().showMessage(STATES[state])

def cleanup (self):

self.p = None

app = QApplication(sys.argv)

w = MainWindow()

app. exec ()

In this example we store a reference to the process in self.p, meaning we can only run a single process at once. But you are free to run as many processes as you like alongside your application. If you don’t need to track information from them, you can simply store references to the processes in a list.

However, if you want to track progress and parse output from workers individually, you may want to consider creating a manager class to handle and track all your processes. There is an example of this in the source files with the book, named qprocess_manager.py.

The full source code for the example is available in the source code for the book, but below we’ll look at the JobManager class itself.

Listing 218. concurrent/qprocess_manager.py

class JobManager (QAbstractListModel):

"""

Manager to handle active jobs and stdout, stderr and progress parsers. Also functions as a Qt data model for a view displaying progress for each process. """

_jobs = {} _state = {} _parsers = {}

status = pyqtSignal(str) result = pyqtSignal(str, object) progress = pyqtSignal(str, int)

def init (self): super().init()

self.status_timer = QTimer() self.status_timer.setInterval( 100 ) self.status_timer.timeout.connect(self.notify_status) self.status_timer.start()

Internal signal, to trigger update of progress via parser.

self.progress.connect(self.handle_progress)

def notify_status (self): n_jobs = len(self._jobs) self.status.emit("{} jobs".format(n_jobs))

def execute (self, command, arguments, parsers=None): """ Execute a command by starting a new process. """

job_id = uuid.uuid4().hex

By default, the signals do not have access to any

information about

the process that sent it. So we use this constructor to

annotate

each signal with a job_id.

def fwd_signal (target): return lambda *args: target(job_id, *args)

self._parsers[job_id] = parsers or []

Set default status to waiting, 0 progress.

self._state[job_id] = DEFAULT_STATE.copy()

p = QProcess() p.readyReadStandardOutput.connect( fwd_signal(self.handle_output) ) p.readyReadStandardError.connect(fwd_signal(self. handle_output)) p.stateChanged.connect(fwd_signal(self.handle_state)) p.finished.connect(fwd_signal(self.done))

self._jobs[job_id] = p

p.start(command, arguments)

self.layoutChanged.emit()

def handle_output (self, job_id): p = self._jobs[job_id] stderr = bytes(p.readAllStandardError()).decode("utf8") stdout = bytes(p.readAllStandardOutput()).decode("utf8") output = stderr + stdout

parsers = self._parsers.get(job_id) for parser, signal_name in parsers:

Parse the data using each parser in turn.

result = parser(output) if result:

Look up the signal by name (using signal_name), and

emit the parsed result.

signal = getattr(self, signal_name) signal.emit(job_id, result)

def handle_progress (self, job_id, progress): self._state[job_id]["progress"] = progress self.layoutChanged.emit()

def handle_state (self, job_id, state):

self._state[job_id]["status"] = state

self.layoutChanged.emit()

def done (self, job_id, exit_code, exit_status):

"""

Task/worker complete. Remove it from the active workers

dictionary. We leave it in worker_state, as this is used to

to display past/complete workers too.

"""

del self._jobs[job_id]

self.layoutChanged.emit()

def cleanup (self):

"""

Remove any complete/failed workers from worker_state.

"""

for job_id, s in list(self._state.items()):

if s["status"] == QProcess.ProcessState.NotRunning:

del self._state[job_id]

self.layoutChanged.emit()

# Model interface

def data (self, index, role):

if role == Qt.ItemDataRole.DisplayRole:

# See below for the data structure.

job_ids = list(self._state.keys())

job_id = job_ids[index.row()]

return job_id, self._state[job_id]

def rowCount (self, index):

return len(self._state)

This class provides a model view interface allowing it to be used as the basis for a QListView. The custom delegate ProgressBarDelegate delegate draws a progress bar for each item, along with the job identifier. The color of the progress bar is determined by the status of the process — dark green if active, or light green if complete.

Parsing of progress information from workers is tricky in this setup, because the

.readyReadStandardError and .readyReadStandardOutput signals do not pass the data, or information about the job that is ready. To work around this we define our custom job_id and intercept the signals to add this data to them.

Parsers for the jobs are passed in when executing the command and stored in _parsers. Output received from each job is passed through the respective parser and used to emit the data, or update the job’s progress. We define two simple parsers: one for extracting the current progress and one for getting the output data.

Listing 219. concurrent/qprocess_manager.py

progress_re = re.compile("Total complete: (\d+)%", re.M)

def simple_percent_parser (output):

"""

Matches lines using the progress_re regex,

returning a single integer for the % progress.

"""

m = progress_re.search(output)

if m:

pc_complete = m.group( 1 )

return int(pc_complete)

def extract_vars (l):

"""

Extracts variables from lines, looking for lines

containing an equals, and splitting into key=value.

"""

data = {}

for s in l.splitlines():

if "=" in s:

name, value = s.split("=")

data[name] = value

return data

The parsers are passed in as a simple list of tuple containing the function to be

used as the parser and the name of the signal to emit. The signal is looked up by name using getattr on the JobManager. In the example we’ve only defined 2 signals, one for the data/result output and one for progress. But you can add as many signals and parsers as you like. Using this approach you can opt to omit certain parsers for certain tasks if you wish (for example, where no progress information is available).

Run the example code and experiment running tasks in another process. You can start multiple jobs, and watch them complete, updating their current progress as they go. Experiment with adding additional commands and parsers for your own jobs.

Figure 215. The process manager, showing active processes and progress.

Plotting

One of the major strengths of Python is in data science and visualization, using tools such as Pandas , numpy and sklearn for data analysis. Building GUI applications with PyQt6 gives you access to all these Python tools directly from within your app, allowing you to build complex data-driven apps and interactive dashboards. We’ve already covered the model views, which allow us to show data in lists and tables. In this chapter we’ll look at the final piece of that puzzle — plotting data.

When building apps with PyQt6 you have two main choices — matplotlib (which also gives access to Pandas plots) and PyQtGraph, which creates plots with Qt- native graphics. In this chapter we’ll look at how you can use these libraries to visualize data in your applications.