Pi Pico W にサーバーを建ててクライアントと双方向通信

前にHTTPリクエスト用のWebサーバーを建ててみましたが、今回はクライアントから直接接続してみます。

こんな感じ。

Clients のPico W ではMicroPython、PCやRaspberry Pi 、Andoroid 端末（スマホなど）ではCPython を使います。

スマホではコンソールアプリのTermux からPython を使います。

元ネタ

Wireless Communication Between Two Raspberry Pi Pico W Boards

上記サイトの例では、クライアントが定期的にメッセージを送るとサーバーが適当なRGBのコードを返してくれるので、それを使ってクライアントは３色のLEDを点灯させるというものでした。

ここでは単純に１回メッセージを送信してみるだけにします。

MicroyPython のセットアップはココ参照

ただPico W にはコンソールの類がないので、送信のトリガーとしてプッシュボタンのイベントでメッセージを送信します。

それでは、双方向のsocket通信をやってみます。

サーバーとしてのPico W

Thonny (Python IDE)を使って有機ELディスプレイ用ssd1306ライブラリをインストール

メニュー＞ツールー＞パッケージを管理を選択

ssd1306で検索して、インストール実行

コードはこんな感じで以下の2つを使用

【secret.py】

ssid = '＜SSID＞'
password = '＜パスワード＞'

1 2	ssid = '＜SSID＞' password = '＜パスワード＞'

【main.py】

# Webserver to send RGB data
# Tony Goodhew 5 July 2022
import network
import socket
import time
from machine import Pin, ADC
from secret import ssid,password
import random
import machine
import utime

import ssd1306

#OLED
sda = machine.Pin(0)
scl = machine.Pin(1)
i2c = machine.I2C(0,sda=sda, scl=scl, freq=400000)
oled = ssd1306.SSD1306_I2C(128, 64, i2c)

#LED
led = machine.Pin("LED", machine.Pin.OUT)

def disp_text(txt):
    oled.fill(0)
    txt_list = txt.split('|')
    a = 5
    for i in range(len(txt_list)):
        oled.text(txt_list[i], 0, a)
        a += 10
    
    oled.show()
    

#
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(ssid, password)
       
# Wait for connect or fail
max_wait = 10
while max_wait > 0:
    if wlan.status() < 0 or wlan.status() >= 3:
        break
    max_wait -= 1
    print('waiting for connection...')
    time.sleep(1)

# Handle connection error
if wlan.status() != 3:
    raise RuntimeError('network connection failed')
else:
    print('connected')
    status = wlan.ifconfig()
    txt = 'ip = ' + '|' + status[0]
    print(txt)
    #
    disp_text(txt)
    
# Open socket
addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]

s = socket.socket()
s.bind(addr)
s.listen(1)

print('listening on', addr)
led.value(1)

# Listen for connections
while True:
    try:
        cl, addr = s.accept()
        print('client connected from', addr)
        request = cl.recv(1024)
        print(request)
        
        s1 = str(request,'UTF-8')
        print(s1)
        mes = "I received messa" + "|" + "ge from" + "|" + str(addr[0]) + "|" + s1
        disp_text(mes)
        
        response = "Noted!" + "|" + str(addr[0]) # This is what we send in reply

        cl.send(response)
        print("Sent:" + response)
        cl.close()

    except OSError as e:
        cl.close()
        print('connection closed')
        led.value(0)

# Webserver to send RGB data

# Tony Goodhew 5 July 2022

import network

import socket

import time

from machine import Pin, ADC

from secret import ssid,password

import random

import machine

import utime

import ssd1306

#OLED

sda = machine.Pin(0)

scl = machine.Pin(1)

i2c = machine.I2C(0,sda=sda, scl=scl, freq=400000)

oled = ssd1306.SSD1306_I2C(128, 64, i2c)

#LED

led = machine.Pin("LED", machine.Pin.OUT)

def disp_text(txt):

oled.fill(0)

txt_list = txt.split('|')

a = 5

for i in range(len(txt_list)):

oled.text(txt_list[i], 0, a)

a += 10

oled.show()

wlan = network.WLAN(network.STA_IF)

wlan.active(True)

wlan.connect(ssid, password)

# Wait for connect or fail

max_wait = 10

while max_wait > 0:

if wlan.status() < 0 or wlan.status() >= 3:

break

max_wait -= 1

print('waiting for connection...')

time.sleep(1)

# Handle connection error

if wlan.status() != 3:

raise RuntimeError('network connection failed')

else:

print('connected')

status = wlan.ifconfig()

txt = 'ip = ' + '|' + status[0]

print(txt)

disp_text(txt)

# Open socket

addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]

s = socket.socket()

s.bind(addr)

s.listen(1)

print('listening on', addr)

led.value(1)

# Listen for connections

while True:

try:

cl, addr = s.accept()

print('client connected from', addr)

request = cl.recv(1024)

print(request)

s1 = str(request,'UTF-8')

print(s1)

mes = "I received messa" + "|" + "ge from" + "|" + str(addr[0]) + "|" + s1

disp_text(mes)

response = "Noted!" + "|" + str(addr[0]) # This is what we send in reply

cl.send(response)

print("Sent:" + response)

cl.close()

except OSError as e:

cl.close()

print('connection closed')

led.value(0)

サーバーを先に起動しておきます。

ディスプレイにサーバーのIPアドレスが表示されます。控えておきます。

以下のコードではサーバーのIPアドレスは192.168.0.24 で記述しています。

各環境に合わせて変更してください。

クライアントとしてのPico W

プッシュボタンを押すとメッセージがサーバーに送信され、応答が返されます。Pico にはモニターする手段がないので、応答があればLEDを3回点滅させます。

コードはこんな感じで以下の2つを使用

【secret.py】

ssid = '＜SSID＞'
password = '＜パスワード＞'

1 2	ssid = '＜SSID＞' password = '＜パスワード＞'

【main.py】

import network
import time
from secret import ssid, password
import socket
 
import machine
 
# LED
led = machine.Pin("LED", machine.Pin.OUT)

#プッシュボタン
button = machine.Pin(16, machine.Pin.IN, machine.Pin.PULL_UP)
#ボタンが１回だけ反応するためのフラグ
B = 0


def send_mes(mes):
    ai = socket.getaddrinfo("192.168.0.24", 80) # Address of Web Server
    addr = ai[0][-1]

    # Create a socket and make a HTTP request
    s = socket.socket() # Open socket
    s.connect(addr)

    b = bytes(mes, 'utf-8')
    s.send(b)
    
    ss=s.recv(512)
    # Print what we received
    print(ss)
    s1 = str(ss,'UTF-8')
    #点滅
    for i in range(3):
        time.sleep(0.1)
        led.value(1)
        time.sleep(0.1)
        led.value(0)

    s.close()          # Close socket


wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(ssid, password)
while not wlan.isconnected() and wlan.status() >= 0:
    print("Waiting to connect:")
    time.sleep(1)
 
# Should be connected and have an IP address
wlan.status() # 3 == success
wlan.ifconfig()
print(wlan.ifconfig())


while True:
    try:
        b1 = button.value()
        
        if not b1:
            if B == 0:
                print('Button pressed!')
                send_mes("Hello!!")
                B = 1
        
        else:
            if B == 1:
                print('Button released!')
                B = 0
            
        time.sleep(0.2)# wait
    
    except KeyboardInterrupt:
        break

import network

import time

from secret import ssid, password

import socket

import machine

# LED

led = machine.Pin("LED", machine.Pin.OUT)

#プッシュボタン

button = machine.Pin(16, machine.Pin.IN, machine.Pin.PULL_UP)

#ボタンが１回だけ反応するためのフラグ

B = 0

def send_mes(mes):

ai = socket.getaddrinfo("192.168.0.24", 80) # Address of Web Server

addr = ai[0][-1]

# Create a socket and make a HTTP request

s = socket.socket() # Open socket

s.connect(addr)

b = bytes(mes, 'utf-8')

s.send(b)

ss=s.recv(512)

# Print what we received

print(ss)

s1 = str(ss,'UTF-8')

#点滅

for i in range(3):

time.sleep(0.1)

led.value(1)

time.sleep(0.1)

led.value(0)

s.close() # Close socket

wlan = network.WLAN(network.STA_IF)

wlan.active(True)

wlan.connect(ssid, password)

while not wlan.isconnected() and wlan.status() >= 0:

print("Waiting to connect:")

time.sleep(1)

# Should be connected and have an IP address

wlan.status() # 3 == success

wlan.ifconfig()

print(wlan.ifconfig())

while True:

try:

b1 = button.value()

if not b1:

if B == 0:

print('Button pressed!')

send_mes("Hello!!")

B = 1

else:

if B == 1:

print('Button released!')

B = 0

time.sleep(0.2)# wait

except KeyboardInterrupt:

break

クライアントとしてのラズパイやデスクトップPC

Wi-Fi 環境は事前に設定済みとしておきます。

ターミナルから以下を実行

【send.py】

import socket

ai = socket.getaddrinfo("192.168.0.24", 80)
addr = ai[0][-1]
s = socket.socket()
s.connect(addr)
mes = "Hello"
b = bytes(mes, 'utf-8')
s.send(b)
ss = s.recv(512)

print(ss)

import socket

ai = socket.getaddrinfo("192.168.0.24", 80)

addr = ai[0][-1]

s = socket.socket()

s.connect(addr)

mes = "Hello"

b = bytes(mes, 'utf-8')

s.send(b)

ss = s.recv(512)

print(ss)

python3 send.py

クライアントとしてのアンドロイド・スマホ

スマホはWi-Fi に接続しておきます。

Google PlayストアでTermux を検索してインストール

Termux を起動

こういうソフトキーボードの入力がしづらいと思われたら、ブルートゥースキーボードを使いましょう（ダイソーでも1000円で売ってます）。

CUI からpython をインストールします。

pkg install -y python

1	pkg install -y python

編集用にnanoインストール

pkg install -y nano

1	pkg install -y nano

send.pyという名前でファイル作成
nano send.py

import socket

ai = socket.getaddrinfo("192.168.0.24", 80)
addr = ai[0][-1]
s = socket.socket()
s.connect(addr)
mes = "Hello"
b = bytes(mes, 'utf-8')
s.send(b)
ss = s.recv(512)

print(ss)

import socket

ai = socket.getaddrinfo("192.168.0.24", 80)

addr = ai[0][-1]

s = socket.socket()

s.connect(addr)

mes = "Hello"

b = bytes(mes, 'utf-8')

s.send(b)

ss = s.recv(512)

print(ss)

以下を実行

python send.py

Termuxを抜けるには、exit を実行。

2023/05/23現在のPython のバージョンは3.11.3です。

スマホからいろいろなセンサーを付けた複数のPico W にアクセスして情報を取得するというのは面白い使い方かもしれません。Pico W なら単三電池２本で駆動しますし….。

FRONT

地図と画像のサイト

Pi Pico W にサーバーを建ててクライアントと双方向通信

Be the first to comment

Leave a Reply コメントをキャンセル