HW-416-BのデータをESP32を使用してサーバーに送信する方法

Last updated at 2025-11-02Posted at 2025-11-02

はじめに

今回は，HW-416-BとESP32でデータをとり，サーバーに送信するまでの流れを紹介します．

HW-416-Bとは，人感センサーの1つです．赤外線などを利用し，周辺温度と温度差のあるものが検知範囲内で動いた場合の温度変化を検知する仕組みです．

この性質上，動かないものは検知せず，"検知範囲内で動く生物を検知するセンサー"となっています．

人感センサーの使用例としては，自動点灯ライトや商業施設のトイレなどがあげられます．

今回の実験環境について

今回，クライアントはESP32とHW-416-B，サーバはESXiの仮想マシンを使用します．
クライアントはMicroPythonで実装し，サーバはPythonで実装しています．
今回使用したクライアント機器

ESP32(黒い長方形の機器)
ブレッドボード(白いボード)
HW-416-B(白いカバーのついた機器)

クライアント(ESP32+HW-416-B)

client.py

import socket
import time
import network
import ntptime  # NTP用
from machine import Pin

# --- Wi-Fi設定 ---
SSID = 'Wi-FiのSSID'
PASSWORD = 'Wi-Fiのパスワード'
SERVER_IP = "サーバのIPアドレス" 
SERVER_PORT = 50000            # サーバのポート番号

# PIRセンサー(HW-416-B)
pir = Pin(4, Pin.IN)  # GPIO4に接続

# 動作確認用LED
led = Pin(2, Pin.OUT)

# --- Wi-Fi接続 ---
def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print("📡 Wi-Fi接続中...")
        wlan.connect(SSID, PASSWORD)
        while not wlan.isconnected():
            time.sleep(0.5)
    print("✅ Wi-Fi接続成功:", wlan.ifconfig())

# --- NTPで時刻同期 ---
def sync_time():
    try:
        print("⏳ NTPで時刻同期中...")
        ntptime.settime()
        print("✅ 時刻同期完了")
    except Exception as e:
        print(f"⚠ 時刻同期失敗: {e}")

# --- サーバ送信 ---
def send_to_server(message):
    try:
        addr = socket.getaddrinfo(SERVER_IP, SERVER_PORT)[0][-1]
        s = socket.socket()
        s.connect(addr)
        s.send(bytes(message, "utf-8"))
        s.close()
        print(f"✅ サーバ送信成功: {message}")
        led.value(1)  # LED ON
        time.sleep(0.2)
        led.value(0)  # LED OFF
    except Exception as e:
        print(f"❌ サーバ送信失敗: {e}")

# --- メイン処理 ---
connect_wifi()
sync_time()

print("🚨 HW-416-B監視開始")
while True:
    if pir.value() == 1:  # PIRが人を検知
        # 検知した時刻を取得
        timestamp = time.localtime()
        timestr = "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
            timestamp[0], timestamp[1], timestamp[2],
            timestamp[3], timestamp[4], timestamp[5]
        )
        # メッセージ作成
        message = f"{timestr} MOTION_DETECTED"

        # コンソール表示
        print(f"👀 {timestr} 人を検知 → サーバ送信")

        # サーバ送信
        send_to_server(message)

        time.sleep(3)  # 検知後3秒間は連続送信防止
    time.sleep(0.1)  # ポーリング間隔

サーバ

server.py

import socket
import datetime

# --- サーバ設定 ---
HOST = "0.0.0.0"  # 全てのインターフェースで待機
PORT = 50000
LOG_FILE = "motion_log.txt"

# --- ソケット作成 ---
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((HOST, PORT))
server_socket.listen(5)
print(f"🚀 サーバ起動: {HOST}:{PORT}")

try:
    while True:
        client_socket, addr = server_socket.accept()
        print(f"✅ 接続: {addr}")

        data = client_socket.recv(1024).decode("utf-8")
        if data:
            # コンソール表示
            print(f"📩 受信データ: {data}")

            # ログファイルに追記
            with open(LOG_FILE, "a") as f:
                now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                f.write(f"{now} {addr[0]}: {data}\n")
        client_socket.close()
except KeyboardInterrupt:
    print("\n🛑 サーバ停止")
finally:
    server_socket.close()