【问题标题】:Integrating two Python scripts into one?将两个 Python 脚本集成为一个?
【发布时间】:2021-02-13 07:24:46
【问题描述】:

我有两个 Python 脚本需要相互通信。第一个是在 PySide2 中制作的 GUI。 GUI 由用于控制蓝牙音频设备(播放、暂停、下一个、上一个等)的简单控件组成。这些命令与我发现的第二个 python 脚本一起运行。第二个脚本是一个循环,等待输入这些命令并在这些命令执行后做出响应。我对编程很陌生,我猜这本质上是将前端与后端连接起来,但这是我以前从未做过的事情。

我编写了一个简化版的 GUI,只显示我需要的控件。 “后端”也在下面,但最初可以在这里找到:https://scribles.net/controlling-bluetooth-audio-on-raspberry-pi/

我之前问过一个类似的问题,@eyllanesc 在这里给出了一个可靠且有效的答案:Execute command to a Python script from separate Python script? 但是,使用 QProcess 方法我无法弄清楚如何从后面获取print 输出- end 进入前端脚本。但是,错误消息打印正确。我曾尝试在后端使用sys.stdoutprocess.read 的变体和 QByteArrays,但似乎没有任何进展。

我遇到的另一个问题是,只有在启动脚本之前连接了蓝牙设备时,该脚本才能工作。如果我在它运行时断开连接并尝试重新连接,它将不再接受命令。如果还有一种方法可以监控设备是否正在播放/暂停,以便播放/暂停按钮可以根据设备状态进行更新,那也很有用,但在此阶段并不重要。

有多种方法可以完成,但我觉得最终将两个脚本集成到一个中对我来说会更好,但是我愿意接受任何可行的解决方案。如果有人有任何建议或可以让我开始,我将非常感激!

前端:

import sys
from PySide2.QtWidgets import *

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

        self.playbtn = QPushButton("Play")
        self.nextbtn = QPushButton("Next")
        self.prevbtn = QPushButton("Prev")

        layout = QVBoxLayout()
        layout.addWidget(self.playbtn)
        layout.addWidget(self.nextbtn)
        layout.addWidget(self.prevbtn)

        self.setLayout(layout)

        self.playbtn.released.connect(self.btnplay)
        self.nextbtn.released.connect(self.btnnext)
        self.prevbtn.released.connect(self.btnprev)

    def btnplay(self): #play button turns into pause button upon being pressed
        status = self.playbtn.text()
        if status == "Play":
            self.playbtn.setText("Pause")
            print("Play Pressed")
        elif status == "Pause":
            self.playbtn.setText("Play")
            print("Pause pressed")

    def btnnext(self):
        print("Next pressed")

    def btnprev(self):
        print("Prev pressed")


app = QApplication(sys.argv)

window = MainWindow()
window.show()
app.exec_()

后端:

import dbus, dbus.mainloop.glib, sys
from gi.repository import GLib


def on_property_changed(interface, changed, invalidated):
    if interface != 'org.bluez.MediaPlayer1':
        return
    for prop, value in changed.items():
        if prop == 'Status':
            print('Playback Status: {}'.format(value))
        elif prop == 'Track':
            print('Music Info:')
            for key in ('Title', 'Artist', 'Album'):
                print('   {}: {}'.format(key, value.get(key, '')))


def on_playback_control(fd, condition):
    str = fd.readline()
    if str.startswith('play'):
        player_iface.Play()
    elif str.startswith('pause'):
        player_iface.Pause()
    elif str.startswith('next'):
        player_iface.Next()
    elif str.startswith('prev'):
        player_iface.Previous()
    elif str.startswith('vol'):
        vol = int(str.split()[1])
        if vol not in range(0, 128):
            print('Possible Values: 0-127')
            return True
        transport_prop_iface.Set(
            'org.bluez.MediaTransport1',
            'Volume',
            dbus.UInt16(vol))
    return True


if __name__ == '__main__':
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    obj = bus.get_object('org.bluez', "/")
    mgr = dbus.Interface(obj, 'org.freedesktop.DBus.ObjectManager')
    player_iface = None
    transport_prop_iface = None
    for path, ifaces in mgr.GetManagedObjects().items():
        if 'org.bluez.MediaPlayer1' in ifaces:
            player_iface = dbus.Interface(
                bus.get_object('org.bluez', path),
                'org.bluez.MediaPlayer1')
        elif 'org.bluez.MediaTransport1' in ifaces:
            transport_prop_iface = dbus.Interface(
                bus.get_object('org.bluez', path),
                'org.freedesktop.DBus.Properties')
    if not player_iface:
        sys.exit('Error: Media Player not found.')
    if not transport_prop_iface:
        sys.exit('Error: DBus.Properties iface not found.')

    bus.add_signal_receiver(
        on_property_changed,
        bus_name='org.bluez',
        signal_name='PropertiesChanged',
        dbus_interface='org.freedesktop.DBus.Properties')
    GLib.io_add_watch(sys.stdin, GLib.IO_IN, on_playback_control)
    GLib.MainLoop().run()

2020 年 10 月 31 日更新: 我一直在玩上面链接的早期问题中建议的QProcess 类。通过在按钮按下功能上使用它并在命令执行后添加sys.exit,它消除了始终连接设备的需要,但我仍然找不到从后面接收print 输出的方法-结束脚本。这也感觉像是一种非常肮脏的工作方式。它还保留了播放/暂停状态不会自动更新的问题。如果有人有任何建议,我将非常感激!

import sys
import os.path
from PySide2.QtCore import *
from PySide2.QtWidgets import *

CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))

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

        self.playbtn = QPushButton("Play")
        self.nextbtn = QPushButton("Next")
        self.prevbtn = QPushButton("Prev")

        layout = QVBoxLayout()
        layout.addWidget(self.playbtn)
        layout.addWidget(self.nextbtn)
        layout.addWidget(self.prevbtn)

        self.setLayout(layout)

        self.playbtn.released.connect(self.btnplay)
        self.nextbtn.released.connect(self.btnnext)
        self.prevbtn.released.connect(self.btnprev)

    def btnplay(self):
        self.process = QProcess()
        self.process.readyReadStandardError.connect(self.handle_readyReadStandardError)
        self.process.readyReadStandardOutput.connect(self.handle_readyReadStandardOutput)
        self.process.setProgram(sys.executable)
        script_path = os.path.join(CURRENT_DIR, "test2.py")
        self.process.setArguments([script_path])
        self.process.start()

        status = self.playbtn.text()
        if status == "Play":
            command = "play"
            self.playbtn.setText("Pause")
            print("play pressed")
        elif status == "Pause":
            command = "pause"
            self.playbtn.setText("Play")
            print("pause pressed")

        msg = "{}\n".format(command)
        self.process.write(msg.encode())

    def btnnext(self):
        self.process = QProcess()
        self.process.readyReadStandardError.connect(self.handle_readyReadStandardError)
        self.process.readyReadStandardOutput.connect(self.handle_readyReadStandardOutput)
        self.process.setProgram(sys.executable)
        script_path = os.path.join(CURRENT_DIR, "test2.py")
        self.process.setArguments([script_path])
        self.process.start()

        command = "next"
        msg = "{}\n".format(command)
        self.process.write(msg.encode())
        print("next pressed")

    def btnprev(self):
        self.process = QProcess()
        self.process.readyReadStandardError.connect(self.handle_readyReadStandardError)
        self.process.readyReadStandardOutput.connect(self.handle_readyReadStandardOutput)
        self.process.setProgram(sys.executable)
        script_path = os.path.join(CURRENT_DIR, "test2.py")
        self.process.setArguments([script_path])
        self.process.start()

        command = "prev"
        msg = "{}\n".format(command)
        self.process.write(msg.encode())
        print("prev pressed")

    def handle_readyReadStandardError(self):
        print(self.process.readAllStandardError().data().decode())

    def handle_readyReadStandardOutput(self):
        print(self.process.readAllStandardOutput().data().decode())


app = QApplication(sys.argv)

window = MainWindow()
window.show()
app.exec_()

【问题讨论】:

    标签: python python-3.x glib pyside2 dbus


    【解决方案1】:

    恕我直言,OP 有一个 XY 问题,这给应用程序增加了不必要的复杂性,因为 dbus 事件循环可以与 Qt 共存,如下例所示:

    import sys
    
    import dbus
    import dbus.mainloop.glib
    
    from PyQt5 import QtCore, QtWidgets
    
    
    class AudioManager(QtCore.QObject):
        statusChanged = QtCore.pyqtSignal(str)
        infoChanged = QtCore.pyqtSignal(dict)
    
        def __init__(self, parent=None):
            super().__init__(parent)
            self._player_iface = None
            self._transport_prop_iface = None
    
        def initialize(self):
            bus = dbus.SystemBus()
            obj = bus.get_object("org.bluez", "/")
            mgr = dbus.Interface(obj, "org.freedesktop.DBus.ObjectManager")
            player_iface = None
            transport_prop_iface = None
            for path, ifaces in mgr.GetManagedObjects().items():
                if "org.bluez.MediaPlayer1" in ifaces:
                    player_iface = dbus.Interface(
                        bus.get_object("org.bluez", path), "org.bluez.MediaPlayer1"
                    )
                elif "org.bluez.MediaTransport1" in ifaces:
                    transport_prop_iface = dbus.Interface(
                        bus.get_object("org.bluez", path), "org.freedesktop.DBus.Properties"
                    )
            if not player_iface:
                raise Exception("Error: Media Player not found.")
            if not transport_prop_iface:
                raise Exception("Error: DBus.Properties iface not found.")
    
            self._player_iface = player_iface
            self._transport_prop_iface = transport_prop_iface
    
            bus.add_signal_receiver(
                self.handle_property_changed,
                bus_name="org.bluez",
                signal_name="PropertiesChanged",
                dbus_interface="org.freedesktop.DBus.Properties",
            )
    
        def play(self):
            self._player_iface.Play()
    
        def pause(self):
            self._player_iface.Pause()
    
        def next(self):
            self._player_iface.Next()
    
        def previous(self):
            self._player_iface.Previous()
    
        def set_volume(self, Volume):
            if Volume not in range(0, 128):
                raise ValueError("Possible Values: 0-127")
            self._transport_prop_iface.Set(
                "org.bluez.MediaTransport1", "Volume", dbus.UInt16(vol)
            )
    
        def handle_property_changed(self, interface, changed, invalidated):
            if interface != "org.bluez.MediaPlayer1":
                return
            for prop, value in changed.items():
                if prop == "Status":
                    self.statusChanged.emit(value)
                elif prop == "Track":
                    info = dict()
                    for key in ("Title", "Artist", "Album"):
                        info[key] = str(value.get(key, ""))
                    self.infoChanged.emit(info)
    
    
    class MainWindow(QtWidgets.QWidget):
        def __init__(self, parent=None):
            super().__init__(parent)
    
            self._manager = AudioManager()
            self._manager.infoChanged.connect(self.handle_info_changed)
            self._manager.initialize()
    
            self.playbtn = QtWidgets.QPushButton("Play")
            self.nextbtn = QtWidgets.QPushButton("Next")
            self.prevbtn = QtWidgets.QPushButton("Prev")
    
            layout = QtWidgets.QVBoxLayout(self)
            layout.addWidget(self.playbtn)
            layout.addWidget(self.nextbtn)
            layout.addWidget(self.prevbtn)
    
            self.playbtn.released.connect(self._manager.play)
            self.nextbtn.released.connect(self._manager.next)
            self.prevbtn.released.connect(self._manager.previous)
    
        def handle_info_changed(self, info):
            print(info)
    
    
    if __name__ == "__main__":
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    
        app = QtWidgets.QApplication(sys.argv)
    
        w = MainWindow()
        w.show()
    
        app.exec_()
    

    【讨论】:

    • 我同意避免 stdin / stdout 几乎总是更简单的答案,尽管当您使用子进程的二进制文件而不是另一个 python 脚本时,有时这是不可避免的。
    • 我可以为此吻你!刚刚学习dbus,所以这是非常好的帮助。与我的完整脚本完美配合,并将向您捐赠几杯咖啡!我读过一些关于有一个特定的 QT 事件循环 - dbus.mainloop.qt.DBusQtMainLoop - 是有原因没有使用它还是它并不重要?您还可以推荐任何有关学习更多信息的指南/教程吗?我发现 dbus 真的很有趣。
    【解决方案2】:

    我不熟悉你在后端使用的东西,但它应该像这样简单:

    # in your frontend
    import NAME_OF_BACKEND_PY_FILE_HERE as backend
    def btnplay(self): #play button turns into pause button upon being pressed
            status = self.playbtn.text()
            if status == "Play":
                self.playbtn.setText("Pause")
                print("Play Pressed")
                backend.on_playback_control("play")
            elif status == "Pause":
                self.playbtn.setText("Play")
                print("Pause pressed")
                backend.on_playback_control("pause")
    
        def btnnext(self):
            print("Next pressed")
            backend.on_playback_control("next")
    
        def btnprev(self):
            print("Prev pressed")
            backend.on_playback_control("prev")
    

    在您的后端,您还应该删除此行:if __name__ == '__main__': 并取消缩进它下面的所有代码。我不确定应该如何正常调用该函数或第二个变量“条件”的用途。但这是我能想到的

    【讨论】:

    • 不要认为这有什么好处,因为后端在自己的循环中运行。一旦它到达 import 语句,它就会停留在循环中,直到您手动退出。
    • 不确定它的循环是否仍然有效,您可以尝试做线程以同时运行 2 件事情。这是我能想到的唯一方法
    猜你喜欢
    • 1970-01-01
    • 2016-10-19
    • 2016-09-03
    • 1970-01-01
    • 2014-10-02
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2014-08-17
    相关资源
    最近更新 更多