Wednesday, June 30, 2021

UDP communication between Raspberry Pi/Python and Arduino Nano RP2040 Connect.

This exercise run on Raspberry Pi/Python and Arduino Nano RP2040 Connect, communicate in UDP.

WiFiNINA is needed, install it in Arduino IDE's Library Manager. 

Exercise 1: Simple UDP example.

With WiFiNINA installed and board of Arduino Nano RP2040 Connect selected, you can load WiFiNINA example WiFiUdpSendReceiveString. Remember to update ssid and password in arduino_secrets.h to match your WiFi network. Upload it to Nano RP2040, open Serial Monitor, it will connect to the WiFi network. Check the ip show in Serial Monitor.

In Raspberry Pi side, run the Python code with matched UDP_IP; simple send message to nano RP2040 via UDP.

pyUDP_client_20210630.py
# ref:
# https://wiki.python.org/moin/UdpCommunication
import socket

#UDP_IP = "127.0.0.1"
#UDP_PORT = 5005
UDP_IP = "192.168.197.39"
UDP_PORT = 2390
MESSAGE = b"Hello, World!"

print("UDP target IP: %s" % UDP_IP)
print("UDP target port: %s" % UDP_PORT)
print("message: %s" % MESSAGE)

sock = socket.socket(socket.AF_INET, # Internet
                     socket.SOCK_DGRAM) # UDP
sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))

Exercise 2: With GUI using PyQt5

Keep WiFiUdpSendReceiveString run on Nano RP2040.

In Raspberry Pi side, run following code with GUI.

pyQt5_UDP_client_20210630.py
import sys
from pkg_resources import require

from PyQt5.QtWidgets import (QApplication, QWidget, QLabel,
                             QTextEdit, QPushButton,
                             QVBoxLayout, QMessageBox)
from PyQt5.QtGui import QFont

from PyQt5.QtNetwork import QUdpSocket, QHostAddress

UDP_IP = "192.168.197.39"
UDP_PORT = 2390

print("Python version")
print(sys.version)
print()

class AppWindow(QWidget):
    
    def __init__(self):
        super().__init__()
        
        self.sock = QUdpSocket(self)
        self.sock.bind(QHostAddress(UDP_PORT), UDP_PORT)
        self.sock.readyRead.connect(self.sock_readyRead_slot)
        
        lbAppTitle = QLabel('Python UDP Client to send msg')
        lbAppTitle.setFont(QFont('Anton', 15, QFont.Bold))
        lbSysInfo = QLabel('Python:\n' + sys.version)
        vboxInfo = QVBoxLayout()
        vboxInfo.addWidget(lbAppTitle)
        vboxInfo.addWidget(lbSysInfo)
        
        self.edMsg = QTextEdit()
        btnSend = QPushButton("Send")
        btnSend.clicked.connect(self.btnSend_Clicked)
        vboxMsg = QVBoxLayout()
        vboxMsg.addWidget(self.edMsg)
        vboxMsg.addWidget(btnSend)
        
        vboxMain = QVBoxLayout()
        vboxMain.addLayout(vboxInfo)
        vboxMain.addLayout(vboxMsg)
        vboxMain.addStretch()
        self.setLayout(vboxMain)
        
        self.setGeometry(100, 100, 500,400)
        self.show()
        
    def sock_readyRead_slot(self):
        while self.sock.hasPendingDatagrams():
            datagram, host, port = self.sock.readDatagram(
                self.sock.pendingDatagramSize()
            )
            
            print("rx:")
            message = '{}\nHost: {}\nPort: {}\n\n'.format(datagram.decode(),
                                                          host.toString(),
                                                          port)

            print(message)
            print()
        
    def btnSend_Clicked(self):
        msgToSend = self.edMsg.toPlainText()
        print("tx:")
        print(msgToSend)
        print()
        
        datagram = msgToSend.encode()
        self.sock.writeDatagram(datagram, QHostAddress(UDP_IP), UDP_PORT)
        
    def closeEvent(self, event):
        close = QMessageBox.question(
            self,
            "QUIT",
            "Close Application?",
            QMessageBox.Yes | QMessageBox.No)
        if close == QMessageBox.Yes:
            print("Close")
            event.accept()
        else:
            event.ignore()

if __name__ == '__main__':
    print('run __main__')
    app = QApplication(sys.argv)
    window = AppWindow()
    sys.exit(app.exec_())

print("- bye -")

Exercise 3: Raspberry Pi/Python remote control Nano RP2040 onboard RGB LED via UDP.

In Nano RP2040, modify to control onboard RGB LED base on incoming command: start with "#RGB", follow with three bytes for R, G and B.
WiFiUdp_RGB__20210630.ino
/*
  WiFi UDP to control Nano RP2040 Connect onboard RGB
 */
#include <SPI.h>
#include <WiFiNINA.h>
#include <WiFiUdp.h>

int status = WL_IDLE_STATUS;
#include "arduino_secrets.h" 
///////please enter your sensitive data in the Secret tab/arduino_secrets.h
char ssid[] = SECRET_SSID;  // your network SSID (name)
char pass[] = SECRET_PASS;  // your network password (use for WPA, or use as key for WEP)
int keyIndex = 0;           // your network key index number (needed only for WEP)

unsigned int localPort = 2390;      // local port to listen on

char packetBuffer[256]; //buffer to hold incoming packet
char  ReplyBuffer[] = "acknowledged";       // a string to send back

WiFiUDP Udp;

void setup() {
  //Initialize serial and wait for port to open:
  Serial.begin(9600);

  pinMode(LEDR, OUTPUT);
  pinMode(LEDG, OUTPUT);
  pinMode(LEDB, OUTPUT);

  analogWrite(LEDR, 0);
  analogWrite(LEDG, 0);
  analogWrite(LEDB, 0);
  
  while (!Serial) {
    ; // wait for serial port to connect. Needed for native USB port only
  }
  analogWrite(LEDR, 255);
  analogWrite(LEDG, 255);
  analogWrite(LEDB, 255);

  // check for the WiFi module:
  if (WiFi.status() == WL_NO_MODULE) {
    Serial.println("Communication with WiFi module failed!");
    // don't continue
    while (true);
  }

  String fv = WiFi.firmwareVersion();
  if (fv < WIFI_FIRMWARE_LATEST_VERSION) {
    Serial.println("Please upgrade the firmware");
  }

  // attempt to connect to WiFi network:
  while (status != WL_CONNECTED) {
    Serial.print("Attempting to connect to SSID: ");
    Serial.println(ssid);
    // Connect to WPA/WPA2 network. Change this line if using open or WEP network:
    status = WiFi.begin(ssid, pass);

    // wait 10 seconds for connection:
    delay(10000);
  }
  Serial.println("Connected to WiFi");
  printWifiStatus();

  Serial.println("\nStarting connection to server...");
  // if you get a connection, report back via serial:
  Udp.begin(localPort);

  analogWrite(LEDR, 0);
  analogWrite(LEDG, 0);
  analogWrite(LEDB, 0);
}

void loop() {

  // if there's data available, read a packet
  int packetSize = Udp.parsePacket();
  if (packetSize) {
    Serial.print("Received packet of size ");
    Serial.println(packetSize);
    Serial.print("From ");
    IPAddress remoteIp = Udp.remoteIP();
    Serial.print(remoteIp);
    Serial.print(", port ");
    Serial.println(Udp.remotePort());

    // read the packet into packetBufffer
    int len = Udp.read(packetBuffer, 255);
    if (len > 0) {
      packetBuffer[len] = 0;
    }
    Serial.println("Contents:");
    Serial.println(packetBuffer);

    // send a reply, to the IP address and port that sent us the packet we received
    Udp.beginPacket(Udp.remoteIP(), Udp.remotePort());
    Udp.write(ReplyBuffer);
    Udp.endPacket();

    String cmd = String(packetBuffer);
    if(cmd.startsWith("#RGB")){
      Serial.println("CMD: #RGB");
      int valR = (int)(packetBuffer[4]);
      int valG = (int)(packetBuffer[5]);
      int valB = (int)(packetBuffer[6]);
      Serial.println("R: " + String(valR));
      Serial.println("G: " + String(valG));
      Serial.println("B: " + String(valB));

      analogWrite(LEDR, 255-valR);
      analogWrite(LEDG, 255-valG);
      analogWrite(LEDB, 255-valB);
    }else{
      Serial.println("NOT MATCH");
    }

  }
}


void printWifiStatus() {
  // print the SSID of the network you're attached to:
  Serial.print("SSID: ");
  Serial.println(WiFi.SSID());

  // print your board's IP address:
  IPAddress ip = WiFi.localIP();
  Serial.print("IP Address: ");
  Serial.println(ip);

  // print the received signal strength:
  long rssi = WiFi.RSSI();
  Serial.print("signal strength (RSSI):");
  Serial.print(rssi);
  Serial.println(" dBm");
}

In Raspberry Pi side:

pyQt5_UDP_client_RGB_20210630.py
import sys
from pkg_resources import require

from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import (QApplication, QWidget, QLabel,
                             QSlider, QPushButton,
                             QVBoxLayout, QHBoxLayout, QMessageBox)
from PyQt5.QtGui import QFont

from PyQt5.QtNetwork import QUdpSocket, QHostAddress

UDP_IP = "192.168.197.39"
UDP_PORT = 2390

print("Python version")
print(sys.version)
print()

class AppWindow(QWidget):
    
    def __init__(self):
        super().__init__()
        
        self.sock = QUdpSocket(self)
        self.sock.bind(QHostAddress(UDP_PORT), UDP_PORT)
        self.sock.readyRead.connect(self.sock_readyRead_slot)
        
        lbAppTitle = QLabel('Python UDP Client to control RGB')
        lbAppTitle.setFont(QFont('Anton', 15, QFont.Bold))
        lbSysInfo = QLabel('Python:\n' + sys.version)
        vboxInfo = QVBoxLayout()
        vboxInfo.addWidget(lbAppTitle)
        vboxInfo.addWidget(lbSysInfo)

        
        lbR = QLabel(' R: ')
        self.sliderR = QSlider(Qt.Horizontal, self)
        self.sliderR.setRange(0, 255)
        self.sliderR.valueChanged.connect(self.sliderRGB_valueChanged_slot)
        boxR = QHBoxLayout()
        boxR.addWidget(lbR)
        boxR.addWidget(self.sliderR)
        
        lbG = QLabel(' G: ')
        self.sliderG = QSlider(Qt.Horizontal, self)
        self.sliderG.setRange(0, 255)
        self.sliderG.valueChanged.connect(self.sliderRGB_valueChanged_slot)
        boxG = QHBoxLayout()
        boxG.addWidget(lbG)
        boxG.addWidget(self.sliderG)
        
        lbB = QLabel(' B: ')
        self.sliderB = QSlider(Qt.Horizontal, self)
        self.sliderB.setRange(0, 255)
        self.sliderB.valueChanged.connect(self.sliderRGB_valueChanged_slot)
        boxB = QHBoxLayout()
        boxB.addWidget(lbB)
        boxB.addWidget(self.sliderB)
        
        boxRGB = QVBoxLayout()
        boxRGB.addLayout(boxR)
        boxRGB.addLayout(boxG)
        boxRGB.addLayout(boxB)
        
        btnSend = QPushButton("Update")
        btnSend.clicked.connect(self.btnSend_Clicked)
        vboxMsg = QVBoxLayout()
        vboxMsg.addLayout(boxRGB)
        vboxMsg.addWidget(btnSend)
        
        vboxMain = QVBoxLayout()
        vboxMain.addLayout(vboxInfo)
        vboxMain.addLayout(vboxMsg)
        vboxMain.addStretch()
        self.setLayout(vboxMain)
        
        self.setGeometry(100, 100, 500,400)
        self.show()
        
    def sock_readyRead_slot(self):
        while self.sock.hasPendingDatagrams():
            datagram, host, port = self.sock.readDatagram(
                self.sock.pendingDatagramSize()
            )
            
            print("rx:")
            message = '{}\nHost: {}\nPort: {}\n\n'.format(datagram.decode(),
                                                          host.toString(),
                                                          port)

            print(message)
            print()
        
    def btnSend_Clicked(self):
        print("tx:")
        
        valueR = self.sliderR.value()
        valueG = self.sliderG.value()
        valueB = self.sliderB.value()
        
        CMD_RGB = "#RGB"
        bCMD_RGB = str.encode(CMD_RGB) + bytes([valueR, valueG, valueB])
        print(type(bCMD_RGB))
        print("CMD: ", bCMD_RGB)
        
        self.sock.writeDatagram(bCMD_RGB, QHostAddress(UDP_IP), UDP_PORT)
        
        """
        msgToSend = self.edMsg.toPlainText()
        print("tx:")
        print(msgToSend)
        print()
        
        datagram = msgToSend.encode()
        self.sock.writeDatagram(datagram, QHostAddress(UDP_IP), UDP_PORT)
        """
        
    def sliderRGB_valueChanged_slot(self):
        print("sliderRGB_valueChanged_slot")
        valueR = self.sliderR.value()
        valueG = self.sliderG.value()
        valueB = self.sliderB.value()
        print(" R: ", valueR, " G: ", valueG, " B: ", valueB)
        
        CMD_RGB = "#RGB"
        bCMD_RGB = str.encode(CMD_RGB) + bytes([valueR, valueG, valueB])
        print(type(bCMD_RGB))
        print(bCMD_RGB)
        
    def closeEvent(self, event):
        close = QMessageBox.question(
            self,
            "QUIT",
            "Close Application?",
            QMessageBox.Yes | QMessageBox.No)
        if close == QMessageBox.Yes:
            print("Close")
            event.accept()
        else:
            event.ignore()

if __name__ == '__main__':
    print('run __main__')
    app = QApplication(sys.argv)
    window = AppWindow()
    sys.exit(app.exec_())

print("- bye -")


~ More exercise of Arduino Nano RP2040 Connect.

No comments:

Post a Comment