同时使用GUI和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，必须将“开”或“关”置于“开”或“关”位置以更改“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_())