Abstraction layer for Python Qt networks.
Project description
QtPyNetwork is a small abstraction layer for sending and receiving messages using TCP sockets.
Check out the complete documentation.
Features
every data write and read call is executed inside thread
signals for each event - connected, disconnected, error, etc.
There are two servers available:
QBalancedServer
QThreadedServer
The first one has constant amount of threads. When new client connects, server checks which worker has the least amount of active sockets and passes socket descriptor to that thread. QThreadedServer creates new thread for each connected device.
QThreadedClient is socket client, that keeps socket in separate thread.
Usage
See examples directory for client and server code samples. You can use both composition and inheritance to create client and server.
Client
from qtpy.QtWidgets import QApplication
from qtpy.QtCore import QObject, Slot, QCoreApplication
import sys
import logging
from QtPyNetwork.client import QThreadedClient
IP, PORT = "127.0.0.1", 12500
class Main(QObject):
def __init__(self):
super(Main, self).__init__(None)
def setup(self):
self.logger = logging.getLogger(self.__class__.__name__)
self.cln = QThreadedClient()
self.cln.message.connect(self.on_message)
self.cln.failed_to_connect.connect(self.close)
self.cln.disconnected.connect(self.close)
self.cln.start(IP, PORT)
@Slot(bytes)
def on_message(self, data: bytes):
self.logger.info(data)
@Slot()
def close(self):
self.cln.close()
while self.cln.is_running():
self.cln.wait()
QApplication.instance().quit()
if __name__ == '__main__':
logging.basicConfig(level=logging.NOTSET)
app = QCoreApplication(sys.argv)
main = Main()
main.setup()
sys.exit(app.exec_())
Server
from qtpy.QtCore import QObject, Slot, QCoreApplication
import sys
import logging
from QtPyNetwork.server import QBalancedServer
from QtPyNetwork.models import Device
IP, PORT = "127.0.0.1", 12500
class Main(QBalancedServer):
def __init__(self):
super(Main, self).__init__(None)
self.logger = logging.getLogger(self.__class__.__name__)
@Slot(Device)
def on_connected(device: Device):
self.logger.info("New device connected: {}".format(device.id()))
@Slot(Device, bytes)
def on_message(self, device: Device, message: bytes):
self.logger.info("Received from {}: {}".format(device.id(), message))
@Slot(Device)
def on_disconnected(self, device: Device):
self.logger.info("Device {} disconnected".format(device.id()))
self.close()
if __name__ == '__main__':
logging.basicConfig(level=logging.NOTSET)
app = QCoreApplication(sys.argv)
main = Main()
main.start(IP, PORT)
sys.exit(app.exec_())
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for QtPyNetwork-0.6.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c305b21f118a41aa2c233911c1c1fc5c838c0ea4a253fe8cc415914f994e8340 |
|
MD5 | 8269224786640851b8c019db6d2a9b74 |
|
BLAKE2b-256 | ac7c50ddf3a95905cfa916ca37e1961202427c38e19b48d59b2f0edfaaf37aff |