NodeMCU and light detector – part 3: PyQt5 to the rescue

Our previous computer app was a failure. Happens from time to time:) Choosing wxPython was a huge mistake. Seems wx bindings are outdated.
After a long walk to cool of my head and think a little (but mostly catching pokemons) I recovered my strength and began second approach. This time I’m using PyQt5. It has Python3 bindings and lots of nice features, like ability to emit signals to other objects.

Spoiler: this time a full success 🙂

Get source from GitHub

Planing again

After some digging into PyQt5 I’m sure that this is a good choice. It’s system tray implementation covers Windows, Gnome, KDE, Unity and some more.

Communication between view and server will be taken care by Qt Signals.

This time I split application into four files, view, message, main and signals. Lets look at them



import message
import signals
import view
import socket
import json

__author__ = 'Bartosz Kościow'

HOST = ''
PORT = 5053
NODE = "light-wc"

def get_current_state():
    address = ('', PORT)
    packet = {
        "protocol": "iot:1",
        "node": "computer",
        "event": "state",
        "targets": [

    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    msg = json.dumps(packet)
    s.sendto(msg.encode(), address)
    (data, ip) = s.recvfrom(1024)
    state = "unknown"
        msg = json.loads(data.decode())
        if msg['protocol'] == "iot:1":
            state = msg['response']
    except ValueError:

    return state

I still do not know what to do with get_current_state function so I left it here. And I still have to set IP manually. Automatically wrong interface is selected.

def main():
    communication = {
        'state': signals.StateChange(),
        'close': signals.ExitApplication()
    state = get_current_state()
    message.run_server(HOST, PORT, NODE, communication)
    view.run_gui(state, communication)

if __name__ == '__main__':

This our entry point. We have two signals. First one is from server to gui and takes event name as argument. Second one is from gui to server – it is emitted when exit is clicked. I named them state and close. For easiness they are in dictionary.
We get current sensor state, start a server and start a gui. State from sensor is passed to gui so it will know what icon it must display.



from PyQt5.QtCore import pyqtSignal, QObject

__author__ = 'Bartosz Kościow'

class StateChange(QObject):
    state = pyqtSignal(str)

class ExitApplication(QObject):
    close = pyqtSignal()

Yes, thread communication is such simple. We create a class and one parameter in it. From now on we may emit or connect to it.
State takes event name as argument while close is without any arguments.



import socketserver
import threading
import json

__author__ = 'Bartosz Kościow'

class MessageHandler(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.request[0].strip()
        event = None
            message = json.loads(data.decode())
            if 'protocol' in message \
                    and message['protocol'] == "iot:1" \
                    and 'node' in message \
                    and message['node'] == self.server.node_name:
                event = message['event']
        except ValueError:

        if event:

It is almost same as previous version. One big change is to call emit_signal function when receiving proper message.

class ThreadedUDPServer(socketserver.ThreadingMixIn, socketserver.UDPServer):
    def __init__(self, address, handler, node_name, communication):
        super().__init__(address, handler)
        self.node_name = node_name
        self.communication = communication

    def emit_signal(self, event):

    def close_socket(self):

First lets look at emit_signal implementation. This function with one line send event to listeners – in this case our gui.
In constructor we hook function close_socket to close signal. This allows us to close socket while exiting application.

def run_server(host, port, node_name, communication):
    udp_server = ThreadedUDPServer((host, port), MessageHandler, node_name, communication)
    udp_thread = threading.Thread(target=udp_server.serve_forever)

Remember how dirty this function was? Now looks nice and clean.



from PyQt5 import QtGui, QtCore
from PyQt5.QtWidgets import QApplication, QMenu, QSystemTrayIcon, QAction
import sys

__author__ = 'Bartosz Kościów'

class SystemTrayIcon(QSystemTrayIcon):
    def __init__(self, state, communication, parent=None):
        QSystemTrayIcon.__init__(self, parent)
        self.icons = {
            'unknown': QtGui.QIcon('img/unknown.png'),
            'detect.light': QtGui.QIcon('img/busy.png'),
            'detect.dark': QtGui.QIcon('img/empty.png')
        self.communication = communication

    def create_menu(self, parent):
        menu = QMenu(parent)
        exit_action = menu.addAction("Exit")

    def exit(self):

    def state_change(self, event):
        if event in self.icons:

We hook function state_change to state signal. It will receive event name as argument. See that exit function emits close signal. What else? We have a menu with exit option and function to change icon in taskbar.

See that green bulb displays when light is off. This means that WC is free and one may go 🙂


This time multiplatform application is a success. It works under Windows, Linux and maybe even MacOS.
With it WC-Bulb project is completed.


One comment

Leave a Reply

Fill in your details below or click an icon to log in: Logo

You are commenting using your account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s