同时使用 GUI _and_ CLI 与我的应用程序交互

Question

我有一个带有交互式 GUI 的 PyQt 桌面应用程序。

基本上它是一个仪表控制器：点击按钮可能会启动泵、触发声音警报、一些线程不断地从流量计中读取数据等。

我希望能够远程控制应用程序/仪器。

仪器和运行 Qt 桌面应用程序的计算机位于远程站点，互联网连接（非常）缓慢且成本高昂，因此无法接受类似 TeamViewer 的解决方案；但是我可以通过 SSH 连接到 linux 计算机。

这就是为什么我认为使用命令行界面与我的应用程序交互可能是一个不错的解决方案。

我知道可以开发一个作为桌面 GUI 或 CLI 启动的应用程序，但是是否可以在已经运行的桌面 GUI 上与 CLI 交互？

我希望桌面一直在运行，它不应该停止。 我希望当我使用 CLI 启动泵时，它同时在 GUI 中显示为“已启动”。

换句话说，我希望能够编写一些 CLI 的命令，就像我虚拟地单击 GUI 的按钮一样。

Answer 1

在这些情况下，最好有一个始终运行的服务，并且 GUI 和 CLI 是客户端：

      ┌-----------------  GUI     
      ↓
┌----------┐
| Service  | ←----------  CLI
└----------┘
      ↑
      └-----------------  Another Client

客户端之间的通信必须通过一些进程间通信 (IPC) 来完成，例如 DBus、ZeroMQ、Mqtt 等。在 Qt 的情况下，您可以使用 Qt Remote Objects (QtRO)。

因此，GUI、CLI 或任何客户端都会向服务发出请求以修改任何输出（启动泵、触发声音警报等）并接收通知。

---

在以下示例中，它说明了我在前几行中指出的内容。

为此，首先启动 service.py，然后启动 cli.py 或 gui.py，对于带有按钮的 GUI，使用 QLabel 显示的“bumb”的状态会发生变化。在 cli 的情况下，您必须设置“on”或“off”来更改“bumb”的状态。

service.py

import sys

from PyQt5 import QtCore, QtRemoteObjects


class Bumb(QtCore.QObject):
    def __init__(self, parent=None):
        super().__init__(parent)
        self._current_state = False

    @QtCore.pyqtSlot()
    def pushCurrentState(self, state):
        pass

    def _get_current_state(self):
        return self._current_state

    def _set_current_state(self, state):
        if self._current_state != state:
            self._current_state = state
            print("current state:", state)
            self.currentStateChanged.emit(state)

    currentStateChanged = QtCore.pyqtSignal(bool)

    currentState = QtCore.pyqtProperty(
        bool,
        fget=_get_current_state,
        fset=_set_current_state,
        notify=currentStateChanged,
    )


if __name__ == "__main__":
    app = QtCore.QCoreApplication(sys.argv)

    bumb = Bumb()

    register_node = QtRemoteObjects.QRemoteObjectRegistryHost(
        QtCore.QUrl("local:registry")
    )
    source_node = QtRemoteObjects.QRemoteObjectHost(
        QtCore.QUrl("local:replica"), QtCore.QUrl("local:registry")
    )
    source_node.enableRemoting(bumb, "bumb")

    sys.exit(app.exec_())

gui.py

import sys

from PyQt5 import QtCore, QtGui, QtWidgets, QtRemoteObjects


class Widget(QtWidgets.QWidget):
    def __init__(self, bumb, parent=None):
        super().__init__(parent)

        self._bumb = bumb

        button = QtWidgets.QPushButton("Change State")
        self.label = QtWidgets.QLabel(text="Bumb", alignment=QtCore.Qt.AlignCenter)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(button)
        lay.addWidget(self.label)
        self.resize(640, 480)

        button.clicked.connect(self.onClicked)

    @property
    def bumb(self):
        return self._bumb

    @QtCore.pyqtSlot()
    def onClicked(self):
        self.bumb.setProperty("currentState", not self.bumb.property("currentState"))

    @QtCore.pyqtSlot()
    def initConnection(self):
        self.bumb.currentStateChanged.connect(self.onCurrentStateChanged)

    @QtCore.pyqtSlot(bool)
    def onCurrentStateChanged(self, state):
        color = QtGui.QColor("red") if state else QtGui.QColor("green")
        self.label.setStyleSheet("background-color: {}".format(color.name()))


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    replica_node = QtRemoteObjects.QRemoteObjectNode(QtCore.QUrl("local:registry"))
    replica_bumb = replica_node.acquireDynamic("bumb")

    w = Widget(replica_bumb)
    w.show()

    replica_bumb.initialized.connect(w.initConnection)

    sys.exit(app.exec_())

cli.py

import platform
import sys

from PyQt5 import QtCore, QtRemoteObjects


class NativeMessenger(QtCore.QObject):
    messageChanged = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)

        self.m_qin = QtCore.QFile()

        self.m_qin.open(
            sys.stdin.fileno(), QtCore.QIODevice.ReadOnly | QtCore.QIODevice.Unbuffered
        )

        if platform.system() == "Windows":
            import win32api

            if sys.platform == "win32":
                import os
                import msvcrt

                if platform.python_implementation() == "PyPy":
                    os.fdopen(fh.fileno(), "wb", 0)
                else:
                    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)

            self.m_notifier = QtCore.QWinEventNotifier(
                win32api.GetStdHandle(win32api.STD_INPUT_HANDLE)
            )

        else:
            self.m_notifier = QtCore.QSocketNotifier(
                sys.stdin.fileno(), QtCore.QSocketNotifier.Read, self
            )

        self.m_notifier.activated.connect(self.readyRead)

    @QtCore.pyqtSlot()
    def readyRead(self):
        line = self.m_qin.readLine().data().decode().strip()
        self.messageChanged.emit(line)


class Manager(QtCore.QObject):
    def __init__(self, bumb, parent=None):
        super().__init__(parent)
        self._bumb = bumb

    @property
    def bumb(self):
        return self._bumb

    def execute(self, command):
        commands = {"on": True, "off": False}
        state = commands.get(command)
        if state is not None:
            self.bumb.setProperty("currentState", state)

    @QtCore.pyqtSlot()
    def initConnection(self):
        self.bumb.currentStateChanged.connect(self.onCurrentStateChanged)

    @QtCore.pyqtSlot(bool)
    def onCurrentStateChanged(self, state):
        print("LOG:", state)


if __name__ == "__main__":
    app = QtCore.QCoreApplication(sys.argv)

    replica_node = QtRemoteObjects.QRemoteObjectNode(QtCore.QUrl("local:registry"))
    replica_bumb = replica_node.acquireDynamic("bumb")

    manager = Manager(replica_bumb)

    replica_bumb.initialized.connect(manager.initConnection)

    messenger = NativeMessenger()
    messenger.messageChanged.connect(manager.execute)

    sys.exit(app.exec_())