Table Of Contents
Integrating with other Frameworks¶
New in version 1.0.8.
Using Twisted inside Kivy¶
Note
You can use the kivy.support.install_twisted_reactor function to install a twisted reactor that will run inside the kivy event loop.
Any arguments or keyword arguments passed to this function will be passed on the threadedselect reactors interleave function. These are the arguments one would usually pass to twisted’s reactor.startRunning
Warning
Unlike the default twisted reactor, the installed reactor will not handle any signals unless you set the ‘installSignalHandlers’ keyword argument to 1 explicitly. This is done to allow kivy to handle the signals as usual, unless you specifically want the twisted reactor to handle the signals (e.g. SIGINT).
The kivy examples include a small example of a twisted server and client. The server app has a simple twisted server running and logs any messages. The client app can send messages to the server and will print its message and the response it got. The examples are based mostly on the simple Echo example from the twisted docs, which you can find here:
To try the example, run echo_server_app.py first, and then launch echo_client_app.py. The server will reply with simple echo messages to anything the client app sends when you hit enter after typing something in the textbox.
Server App¶
# install_twisted_rector must be called before importing and using the reactor
from kivy.support import install_twisted_reactor
install_twisted_reactor()
from twisted.internet import reactor
from twisted.internet import protocol
class EchoServer(protocol.Protocol):
def dataReceived(self, data):
response = self.factory.app.handle_message(data)
if response:
self.transport.write(response)
class EchoServerFactory(protocol.Factory):
protocol = EchoServer
def __init__(self, app):
self.app = app
from kivy.app import App
from kivy.uix.label import Label
class TwistedServerApp(App):
label = None
def build(self):
self.label = Label(text="server started\n")
reactor.listenTCP(8000, EchoServerFactory(self))
return self.label
def handle_message(self, msg):
msg = msg.decode('utf-8')
self.label.text = "received: {}\n".format(msg)
if msg == "ping":
msg = "Pong"
if msg == "plop":
msg = "Kivy Rocks!!!"
self.label.text += "responded: {}\n".format(msg)
return msg.encode('utf-8')
if __name__ == '__main__':
TwistedServerApp().run()
Client App¶
# install_twisted_rector must be called before importing the reactor
from __future__ import unicode_literals
from kivy.support import install_twisted_reactor
install_twisted_reactor()
# A Simple Client that send messages to the Echo Server
from twisted.internet import reactor, protocol
class EchoClient(protocol.Protocol):
def connectionMade(self):
self.factory.app.on_connection(self.transport)
def dataReceived(self, data):
self.factory.app.print_message(data.decode('utf-8'))
class EchoClientFactory(protocol.ClientFactory):
protocol = EchoClient
def __init__(self, app):
self.app = app
def startedConnecting(self, connector):
self.app.print_message('Started to connect.')
def clientConnectionLost(self, connector, reason):
self.app.print_message('Lost connection.')
def clientConnectionFailed(self, connector, reason):
self.app.print_message('Connection failed.')
from kivy.app import App
from kivy.uix.label import Label
from kivy.uix.textinput import TextInput
from kivy.uix.boxlayout import BoxLayout
# A simple kivy App, with a textbox to enter messages, and
# a large label to display all the messages received from
# the server
class TwistedClientApp(App):
connection = None
textbox = None
label = None
def build(self):
root = self.setup_gui()
self.connect_to_server()
return root
def setup_gui(self):
self.textbox = TextInput(size_hint_y=.1, multiline=False)
self.textbox.bind(on_text_validate=self.send_message)
self.label = Label(text='connecting...\n')
layout = BoxLayout(orientation='vertical')
layout.add_widget(self.label)
layout.add_widget(self.textbox)
return layout
def connect_to_server(self):
reactor.connectTCP('localhost', 8000, EchoClientFactory(self))
def on_connection(self, connection):
self.print_message("Connected successfully!")
self.connection = connection
def send_message(self, *args):
msg = self.textbox.text
if msg and self.connection:
self.connection.write(msg.encode('utf-8'))
self.textbox.text = ""
def print_message(self, msg):
self.label.text += "{}\n".format(msg)
if __name__ == '__main__':
TwistedClientApp().run()