Python kivy asyncio

Question

Пишу клиент для сервера с помощью киви. При запуске выдает ошибку AttributeError: 'MyApp' object has no attribute 'parent'. Скорее всего ошибка в последней части кода. Вот код

import asyncio
from asyncio import transports
from PySide2.QtWidgets import QMainWindow, QApplication
from asyncqt import QEventLoop
from interface import Ui_MainWindow


class ClientProtocol(asyncio.Protocol):
    transport: transports.Transport
    window: 'Chat'

    def __init__(self, chat):
        self.window = chat

    def data_received(self, data: bytes):
        print(data)
        decoded = data.decode()
        self.window.plainTextEdit.appendPlainText(decoded)

    def connection_made(self, transport: transports.Transport):
        self.window.plainTextEdit.appendPlainText("Успешно подключились, введите логин")
        self.transport = transport

    def connection_lost(self, exception):
        self.window.plainTextEdit.appendPlainText("Вы отключены от сервера")


class Chat(QMainWindow, Ui_MainWindow):
    protocol: ClientProtocol

    def __init__(self):
        super().__init__()
        self.setupUi(self)
        self.pushButton.clicked.connect(self.send_message)

    def send_message(self):
        message = self.lineEdit.text()
        self.lineEdit.clear()
        self.protocol.transport.write(message.encode())

    def create_protocol(self):
        self.protocol = ClientProtocol(self)
        return self.protocol

    async def start(self):
        self.show()

        loop = asyncio.get_running_loop()

        coroutine = loop.create_connection(
            self.create_protocol,
            "127.0.0.1",
            8888
        )

        await asyncio.wait_for(coroutine, 1000)


app = QApplication()
loop = QEventLoop(app)

asyncio.set_event_loop(loop)

window = Chat()

loop.create_task(window.start())
loop.run_forever()

Answer 1

Для стыковки PySide2 и asyncio я рекомендую asyncqt: pip3 install asyncqt

from asyncqt import QEventLoop, asyncSlot, asyncClose
import sys
import asyncio

from PySide2.QtWidgets import (
    QApplication, QWidget, QLabel, QLineEdit, QTextEdit, QPushButton,
    QVBoxLayout)

from PySide2.QtCore import Signal


class ClientProtocol(asyncio.Protocol):
    def __init__(self, chat):
        self.window = chat

    def data_received(self, data: bytes):
        decoded = data.decode()
        self.window.speak.emit(decoded)

    def connection_made(self, transport):
        self.transport = transport

    def connection_lost(self, exception):
        pass


class MainWindow(QWidget):

    speak = Signal(str)

    __transport = None
    __protocol = None

    def __init__(self):
        super().__init__()
        self.setLayout(QVBoxLayout())  

        self.editResponse = QTextEdit('', self)
        self.layout().addWidget(self.editResponse)

        self.editSend = QLineEdit('', self)
        self.layout().addWidget(self.editSend)
        self.editSend.returnPressed.connect(self.on_editSend)

        self.speak.connect(self.editResponse.append)


    @asyncSlot()
    async def on_editSend(self):
        text = self.editSend.text()
        if self.__transport:
            self.__transport.write(text.encode())
            self.editSend.setText('')
            self.editResponse.append('>'+text)

    def set_transport(self, task):
        self.__transport, self.__protocol = task.result()


if __name__ == '__main__':
    app = QApplication(sys.argv)
    loop = QEventLoop(app)
    asyncio.set_event_loop(loop)
    mainWindow = MainWindow()

    coro = loop.create_connection(
               lambda : ClientProtocol(mainWindow),
               "127.0.0.1",
               8888
        )
    task = loop.create_task(coro)

    task.add_done_callback(mainWindow.set_transport)

    mainWindow.show()

    with loop:
        sys.exit(loop.run_forever())

Код работает - проверил на серевере netcat -l 8888

asyncSlot тут не сильно нужен, но пригодится как пример

БЛОГ НА HUSL

Python kivy asyncio

Ответы (1 шт):