Networking in CircuitPython

published August 14, 2024, last updated September 10, 2024

posted in Internet of Things - IOT/ Wireless CircuitPython Programming/ MicroPython / CircuitPython Breakout Boards/ WiFi

Save Download

New

Beginner

Project guide

Advanced Topics: Ping and UDP

Ping

Ping is a method of measuring the round trip time for messages sent from a host to a network destination and echoed back to the source on an IP network. Pinging involves sending an ICMP echo request to the target host and waiting for an ICMP echo reply.

In CircuitPython, the wifi Module's radio function provides ping functionality. ping = wifi.radio.ping(ip=ping_ip) where ping is the round trip time in seconds a request took place.

Limitations: On Espressif, calling ping() multiple times rapidly exhausts available resources after several calls. Rather than failing at that point, ping() will wait two seconds for enough resources to be freed up before proceeding.

The following shows how a ping can be sent and printed out.

Download File

Copy Code

import os
import ipaddress
import ssl
import wifi

print(f"Connecting to {os.getenv('CIRCUITPY_WIFI_SSID')}")
wifi.radio.connect(os.getenv("CIRCUITPY_WIFI_SSID"), os.getenv("CIRCUITPY_WIFI_PASSWORD"))
print(f"Connected to {os.getenv('CIRCUITPY_WIFI_SSID')}")
print(f"My IP address: {wifi.radio.ipv4_address}")

ping_ip = ipaddress.IPv4Address("8.8.8.8")  # Google.com
ping = wifi.radio.ping(ip=ping_ip)

# retry once if timed out
if ping is None:
    ping = wifi.radio.ping(ip=ping_ip)

if ping is None:
    print("Couldn't ping 'google.com' successfully")
else:
    # convert s to ms
    print(f"Pinging 'google.com' took: {ping * 1000} ms")

import os
import ipaddress
import ssl
import wifi

print(f"Connecting to {os.getenv('CIRCUITPY_WIFI_SSID')}")
wifi.radio.connect(os.getenv("CIRCUITPY_WIFI_SSID"), os.getenv("CIRCUITPY_WIFI_PASSWORD"))
print(f"Connected to {os.getenv('CIRCUITPY_WIFI_SSID')}")
print(f"My IP address: {wifi.radio.ipv4_address}")

ping_ip = ipaddress.IPv4Address("8.8.8.8")  # Google.com
ping = wifi.radio.ping(ip=ping_ip)

# retry once if timed out
if ping is None:
    ping = wifi.radio.ping(ip=ping_ip)

if ping is None:
    print("Couldn't ping 'google.com' successfully")
else:
    # convert s to ms
    print(f"Pinging 'google.com' took: {ping * 1000} ms")

UDP

TCP (Transmission Control Protocol), used for HTTP, HTTPS, and many other kinds of internet connections, is not the only method of transmitting data over a network connection. TCP is a "reliable delivery" protocol: it keeps trying until the data is delivered or gives up with an error. It sets up a persistent connection between two points and uses a sequence of protocol acknowledgments and "handshakes" to ensure reliable delivery.

By contrast, UDP (User Datagram Protocol) is a "connectionless" protocol. It simply tries to deliver packets of data from one point to another. The packets may be dropped along the way if there is congestion or other problems, and their ordering is not guaranteed. It is useful for streaming data such audio, video, or periodic data reporting, where loss of data is not fatal or corrupting.

In general, you request a socket supporting UDP with the following:

Download File

Copy Code

wifi.radio.connect(ssid=os.getenv("CIRCUITPY_WIFI_SSID"), password=os.getenv("CIRCUITPY_WIFI_PASSWORD"))
pool = socketpool.SocketPool(wifi.radio)
sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM)

wifi.radio.connect(ssid=os.getenv("CIRCUITPY_WIFI_SSID"), password=os.getenv("CIRCUITPY_WIFI_PASSWORD"))
pool = socketpool.SocketPool(wifi.radio)
sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM)

You'd then create your packet to send (a bytearray) and send it like this:

Download File

Copy Code

sock.sendto(packet, (URL, port))

sock.sendto(packet, (URL, port))

The port used for UDP for your own use should be 1024 or greater, as lower ports (0-1023) are typically reserved for specific services and protected by the operating system.

Below is an example of using UDP for getting the time from an NTP server:

Download Project Bundle

Copy Code

import wifi
import socketpool
import struct
import time

# connect to wifi
print("Connecting to Wifi")
wifi.radio.connect("mySSID", "myPASS")
pool = socketpool.SocketPool(wifi.radio)

# make socket
print("Creating socket")
sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM)

# Fill packet
packet = bytearray(48)
packet[0] = 0b00100011  # Not leap second, NTP version 4, Client mode
NTP_TO_UNIX_EPOCH = 2208988800  # 1970-01-01 00:00:00

print("Sending packet")
sock.sendto(packet, ("pool.ntp.org", 123))

size, address = sock.recvfrom_into(packet)
print("Received packet")

seconds = struct.unpack_from("!I", packet, offset=len(packet) - 8)[0]
print("Address:", address)
print("Time:", time.localtime(seconds - NTP_TO_UNIX_EPOCH))

import wifi
import socketpool
import struct
import time

# connect to wifi
print("Connecting to Wifi")
wifi.radio.connect("mySSID", "myPASS")
pool = socketpool.SocketPool(wifi.radio)

# make socket
print("Creating socket")
sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM)

# Fill packet
packet = bytearray(48)
packet[0] = 0b00100011  # Not leap second, NTP version 4, Client mode
NTP_TO_UNIX_EPOCH = 2208988800  # 1970-01-01 00:00:00

print("Sending packet")
sock.sendto(packet, ("pool.ntp.org", 123))

size, address = sock.recvfrom_into(packet)
print("Received packet")

seconds = struct.unpack_from("!I", packet, offset=len(packet) - 8)[0]
print("Address:", address)
print("Time:", time.localtime(seconds - NTP_TO_UNIX_EPOCH))

Below is an example posted by Tod Kurt (@todbot):

Download File

Copy Code

# udp_recv_code.py -- receive UDP messages from any receiver, can be another CircuitPython device
# 24 Aug 2022 - @todbot / Tod Kurt
# cribbing from code at https://github.com/adafruit/circuitpython/blob/main/tests/circuitpython-manual/socketpool/datagram/ntp.py

import time, wifi, socketpool, os
print("Connecting to WiFi...")
wifi.radio.connect(ssid=os.getenv("CIRCUITPY_WIFI_SSID"), password=os.getenv("CIRCUITPY_WIFI_PASSWORD"))
print("my IP addr:", wifi.radio.ipv4_address)
pool = socketpool.SocketPool(wifi.radio)

# I, @PaulskPt, used for udp_host erroneously: os.getenv("MULTICAST_GROUP")
udp_host = str(wifi.radio.ipv4_address) # my LAN IP as a string  
udp_port = int(os.getenv("MULTICAST_PORT"))  # a number of your choosing, should be 1024-65000
udp_buffer = bytearray(64)  # stores our incoming packet

sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM) # UDP socket
sock.bind((udp_host, udp_port))  # say we want to listen on this host,port

print("waiting for packets on",udp_host, udp_port)
while True:
    size, addr = sock.recvfrom_into(udp_buffer)
    msg = udp_buffer.decode('utf-8')  # assume a string, so convert from bytearray
    print(f"Received message from {addr[0]}:", msg)

# udp_recv_code.py -- receive UDP messages from any receiver, can be another CircuitPython device
# 24 Aug 2022 - @todbot / Tod Kurt
# cribbing from code at https://github.com/adafruit/circuitpython/blob/main/tests/circuitpython-manual/socketpool/datagram/ntp.py

import time, wifi, socketpool, os
print("Connecting to WiFi...")
wifi.radio.connect(ssid=os.getenv("CIRCUITPY_WIFI_SSID"), password=os.getenv("CIRCUITPY_WIFI_PASSWORD"))
print("my IP addr:", wifi.radio.ipv4_address)
pool = socketpool.SocketPool(wifi.radio)

# I, @PaulskPt, used for udp_host erroneously: os.getenv("MULTICAST_GROUP")
udp_host = str(wifi.radio.ipv4_address) # my LAN IP as a string  
udp_port = int(os.getenv("MULTICAST_PORT"))  # a number of your choosing, should be 1024-65000
udp_buffer = bytearray(64)  # stores our incoming packet

sock = pool.socket(pool.AF_INET, pool.SOCK_DGRAM) # UDP socket
sock.bind((udp_host, udp_port))  # say we want to listen on this host,port

print("waiting for packets on",udp_host, udp_port)
while True:
    size, addr = sock.recvfrom_into(udp_buffer)
    msg = udp_buffer.decode('utf-8')  # assume a string, so convert from bytearray
    print(f"Received message from {addr[0]}:", msg)