python - 同时使用 GUI 和 CLI 与我的应用程序交互

Question

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

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

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

运行 Qt 桌面应用程序的仪器和计算机位于远程站点，互联网连接(非常)缓慢且昂贵，因此类似 TeamViewer 的解决方案是 Not Acceptable ；但是我可以通过 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_())