Open Hard- & Software [ DokuWiki WebSites MediaWiki NextCloud ]

Esp32-MicroPython-Software:

Esp32UdpClientServer.py

#
# Syntax-Arguments: <headerid> <rxipport> <txipaddress> <txipport>
#
# WindowsPC[192.168.178.33] <-> # WindowsPC[192.168.178.33]:
# WinPC: python CheckUdpClientServer.py Windows 4321 192.168.178.33 4321
 
 
# WindowsPC[192.168.178.33] <-> UbuntuPC[192.168.178.48]
# WinPC: python CheckUdpClientServer.py Windows 5001 192.168.178.48 5000
# LinPC: python CheckUdpClientServer.py Ubuntu  5000 192.168.178.33 5001 
#
import sys
import time
import socket as SKT
#
import Thread as THR
import UdpClientServer as UCS
#
WLAN_SSID = 'FritzBoxSA7'
WLAN_PW = '01234567890123456789'
#
UDP_HEADERID = 'UDPHID'
#-----------------------------------------------------------
#   UDP - TxD - TransmitData
#-----------------------------------------------------------
# UDP_IPADDRESS_TX = '192.168.178.255' NO broadcasting to local
UDP_IPADDRESS_TX = '192.168.178.255'
UDP_IPPORT_TX = 5000
#
#-----------------------------------------------------------
#   UDP - RxD - ReceiveData
#-----------------------------------------------------------
UDP_IPADDRESS_LOCAL = '192.168.178.71' # dummy
UDP_IPPORT_RX = 5001
#   
#-----------------------------------------------------------
#   Callback - UdpClient
#-----------------------------------------------------------
def CBOnTxData(udpclientserver, txdata):
    Line = 'CBOnTxData[{}]<{}>'.format(udpclientserver.GetHeaderID(), txdata)
    print(Line)
    return
#
def CBOnRxData(udpclientserver, rxdata):
    Line = 'CBOnRxData[{}]<{}>'.format(udpclientserver.GetHeaderID(), rxdata)
    print(Line)
    return
# 
#-----------------------------------------------------------
#   Main
#-----------------------------------------------------------
if ('__main__' == __name__):
    #
    print('*** CheckUdpClientServer: begin')
    #
    MillisPreset = time.ticks_ms() - 100000
    #
    UdpCS = UCS.CUdpClientServer(WLAN_SSID, WLAN_PW,
                                 UDP_HEADERID, UDP_IPPORT_RX,
                                 UDP_IPADDRESS_TX, UDP_IPPORT_TX, 
                                 CBOnTxData, CBOnRxData)
    UdpCS.Open()
    #        
    print('{}: TxIP-Address[{}]-Port[{}]'.format(UdpCS.GetHeaderID(), 
                                                 UdpCS.GetIPAddressTX(), 
                                                 UdpCS.GetIPPortTX()))
    print('{}: RxIP-Address[{}]-Port[{}](local)'.format(UdpCS.GetHeaderID(), 
                                                 UdpCS.GetIPAddressRXLocal(),
                                                 UdpCS.GetIPPortRX()))
    while UdpCS.IsBusy():
        UdpCS.Execute()
        if 10000 <= (time.ticks_ms() - MillisPreset):
            MillisPreset = time.ticks_ms()
            UdpCS.Transmit('! Millis[{}]ms'.format(MillisPreset))
    UdpCS.Close()
    #
    print('*** CheckUdpClientServer: end')
    #
#

UdpClientServer.py

#
import time
import socket as SKT
import network as NTW
import Thread as THR
#
class CUdpClientServer():
    #
    def __init__(self, wlanssid, wlanpw,
                 headerid, ipportrx, \
                 ipaddresstx, ipporttx, \
                 ontxdata, onrxdata):   
        self.WlanSSID = wlanssid
        self.WlanPW = wlanpw            
        self.HeaderID = headerid
        self.IPAddressRXLocal = ''
        self.IPPortRX = ipportrx
        self.IPAddressTX = ipaddresstx
        self.IPPortTX = ipporttx
        self.OnTxData = ontxdata
        self.OnRxData = onrxdata
        self.Thread = THR.CThread(self.CBOnStart, self.CBOnBusy, \
                                  self.CBOnAbort, self.CBOnEnd)
        self.Wlan = None
        self.RxSocket = None
        self.TxSocket = None
        #
        self.DEBUG = False
        #
        return
    #
    def GetHeaderID(self):
        return self.HeaderID
    def GetIPAddressTX(self):
        return self.IPAddressTX
    def GetIPPortTX(self):
        return self.IPPortTX
    def GetIPAddressRXLocal(self):
        return self.IPAddressRXLocal
    def GetIPPortRX(self):
        return self.IPPortRX
    #
    def IsBusy(self):
        return THR.stBusy == self.Thread.State
    #
    def CBOnStart(self, thread):
        if self.DEBUG:
            print('UdpClientServer.CBOnStart')
        return    
    #
    def CBOnBusy(self, thread):
        if self.DEBUG:
            print('UdpClientServer.CBOnBusy')
        while self.IsBusy():
            RxData = ''
            try:
                Data, Address = self.RxSocket.recvfrom(1024)
                RxData = Data.decode('utf-8')
            except:
                pass 
            else:
                if self.DEBUG:
                    print('!!!!!!!!RxData[{}]'.format(Data.decode('utf-8')))
                if (None != self.OnRxData):
                    self.OnRxData(self, RxData)
            finally:
                time.sleep(0.1)            
        self.RxSocket.close()
        return    
    #
    def CBOnAbort(self, thread):
        if self.DEBUG:
            print('UdpClientServer.CBOnAbort')
        self.RxSocket.close()
        return
    #
    def CBOnEnd(self, thread):
        if self.DEBUG:
            print('UdpClientServer.CBOnEnd')
        self.RxSocket.close()
        return
    #
    def Open(self):
        self.Wlan = NTW.WLAN(NTW.STA_IF)
        self.Wlan.active(False)
        self.Wlan.active(True)
        self.Wlan.connect(self.WlanSSID, self.WlanPW)
        while not(self.Wlan.isconnected()):
            pass
        self.IPAddressRXLocal = self.Wlan.ifconfig()[0]
        if self.DEBUG:
            print('Wlan-IDs[{}]'.format(self.Wlan.ifconfig()))
            print('Wlan-LocalIPAddress[{}]'.format(self.IPAddressRXLocal))
        # Rx
        if (THR.stBusy == self.Thread.State):
            self.Thread.Abort()
        self.RxSocket = SKT.socket(SKT.AF_INET, SKT.SOCK_DGRAM)
        # FORWARD Socket.bind !!!
        self.RxSocket.setsockopt(SKT.SOL_SOCKET, SKT.SO_REUSEADDR, 1)
        self.RxSocket.bind((self.IPAddressRXLocal, self.IPPortRX))
        self.RxSocket.setblocking(0)
        self.Thread.Start()
        # Tx
        self.TxSocket = SKT.socket(SKT.AF_INET, SKT.SOCK_DGRAM)
        self.TxSocket.connect((self.IPAddressTX, self.IPPortTX))
        return   
    #
    def Close(self):
        if self.DEBUG:
            print('UdpClientServer.Close')
        self.Thread.Abort()
        self.RxSocket.close()
        self.TxSocket.close()
    #
    def Abort(self):
        if self.DEBUG:
            print('UdpClientServer.Abort')
        self.Thread.Abort()
        return
    #
    def Transmit(self, text):
        if self.DEBUG:
            print('UdpClientServer.Transmit')
        if (None != self.OnTxData):
            self.OnTxData(self, text)
        self.TxSocket.sendto(text.encode('utf-8'), \
                                         (self.IPAddressTX, self.IPPortTX))
    #
    def Execute(self):
        return
    #     
#
#-----------------------------------------------------------
#   Check UdpClientServer
#-----------------------------------------------------------
WLAN_SSID = 'FritzBoxSA7'
WLAN_PW = '01234567890123456789'
#
UDP_HEADERID = 'UDPHID'
#-----------------------------------------------------------
#   UDP - TxD - TransmitData
#-----------------------------------------------------------
# UDP_IPADDRESS_TX = '192.168.178.255' NO broadcasting to local
UDP_IPADDRESS_TX = '192.168.178.71'
UDP_IPPORT_TX = 5000
#
#-----------------------------------------------------------
#   UDP - RxD - ReceiveData
#-----------------------------------------------------------
UDP_IPADDRESS_LOCAL = '192.168.178.71' # dummy
UDP_IPPORT_RX = 5000
#   
#-----------------------------------------------------------
#   Callback - UdpClient
#-----------------------------------------------------------
def CBOnTxData(udpclientserver, txdata):
    Line = 'UdpClientServer.CBOnTxData[{}]<{}>'.format(udpclientserver.GetHeaderID(), txdata)
    print(Line)
    return
#
def CBOnRxData(udpclientserver, rxdata):
    Line = 'UdpClientServer.CBOnRxData[{}]<{}>'.format(udpclientserver.GetHeaderID(), rxdata)
    print(Line)
    return
#
if ('__main__' == __name__):
    #
    print('*** Check UdpClientServer: begin')        
    #
    UdpCS = CUdpClientServer(WLAN_SSID, WLAN_PW,
                             UDP_HEADERID, UDP_IPPORT_RX,
                             UDP_IPADDRESS_TX, UDP_IPPORT_TX, 
                             CBOnTxData, CBOnRxData)
    UdpCS.Open()    
    #
    print('{}: TxIP-Address[{}]-Port[{}]'.format(UdpCS.GetHeaderID(), 
                                                 UdpCS.GetIPAddressTX(), 
                                                 UdpCS.GetIPPortTX()))
    print('{}: RxIP-Address[{}]-Port[{}](local)'.format(UdpCS.GetHeaderID(), 
                                                 UdpCS.GetIPAddressRXLocal(),
                                                 UdpCS.GetIPPortRX()))
    #
    UdpCS.Transmit('HalloA')
    time.sleep(1.0)
    UdpCS.Transmit('HalloB')
    time.sleep(1.0)
    UdpCS.Transmit('HalloC')
    time.sleep(1.0)
    #
    UdpCS.Close() 
    #
    print('*** Check UdpClientServer: end')
    #
# 

Thread.py

import time
import _thread as THR
# States
stIdle = 0
stBusy = 1
stEnd = 2
#
class CThread():
    #
    def __init__(self, onstart, onbusy, onabort, onend):
        self.State = stIdle        
        self.Thread = None
        self.OnStart = onstart
        self.OnBusy = onbusy
        self.OnAbort = onabort
        self.OnEnd = onend
        return
    #
    def Start(self):
        self.State = stBusy
        if (None != self.OnStart):
            self.OnStart(self)
        self.ThreadID = THR.start_new_thread(self.CBOnExecute, [self])
        return    
    #
    def Abort(self):
        self.State = stEnd
        if (None != self.OnAbort):
            self.OnAbort(self)
        return
    #
    def CBOnExecute(self, thread):
        if (None != self.OnBusy):
            self.OnBusy(self)
        if (None != self.OnEnd):
            self.OnEnd(self)
        return
    #
 
#
#   
#-----------------------------------------------------------
#   Check Thread
#-----------------------------------------------------------
def CBOnStart(thread):
    print('CBOnStart')
    return
def CBOnBusy(thread):
    print('CBOnBusy')
    Counter = 0
    while (stBusy == thread.State):
        Counter += 1
        print(Counter)
        if (3 <= Counter):
            thread.Abort()
        else:
            time.sleep(1.0)
    return
def CBOnAbort(thread):
    print('CBOnAbort')
    return
def CBOnEnd(thread):
    print('CBOnEnd')
    return
#
if ('__main__' == __name__):
    #
    print('*** Check Thread: begin')        
    #
    Thread = CThread(CBOnStart, CBOnBusy, CBOnAbort, CBOnEnd)
    Thread.Start()
    time.sleep(5.0)
    Thread.Abort()
    #
    print('*** Check Thread: end')
    #
#

2112121950_Esp32UdpClientServer_02V01.zip

Open Hard- & Software [ DokuWiki WebSites MediaWiki NextCloud ]