M5Stackでセンサーアプリを作る

import M5
from M5 import Widgets, Mic, Speaker, Imu
import network
import time
import math
import struct
import requests2
from hardware import I2C, Pin

WIFI_SSID = "YOUR_WIFI_SSID"
WIFI_PASSWORD = "YOUR_WIFI_PASSWORD"

GAS_URL = "https://script.google.com/macros/s/YOUR_DEPLOYMENT_ID/exec"
SECRET = "test"

# ===== 感度の初期値（設置場所で調整してください） =====
# ONを超えたら検知、OFF以下に戻った状態が続くと次を検知できます。
NEAR_ON, NEAR_OFF = 200, 120       # 近接センサー生値。cmではありません。
SOUND_ON, SOUND_OFF = 1500, 800   # 16bit音声のRMS。dBではありません。
SHAKE_ON, SHAKE_OFF = 0.35, 0.12  # 前回からの加速度ベクトル変化（g）
COOLDOWN_MS = 3000               # 同じイベントは最低3秒間隔
RELEASE_MS = 500                 # 解除条件が0.5秒続いてから再検知
SEND_ENABLED = False            # 最初は画面確認のみ。送信するときにTrueへ変更

labels = {}
wlan = None
proximity_bus = None
audio_buffer = bytearray(800)    # 400 samples / 8000Hz = 50ms、符号付き16bit
previous_accel = None
gates = {}
saved_count = 0
failed_count = 0
last_display = None


class EventGate:
    def __init__(self, on, off):
        if off >= on:
            raise ValueError("OFF must be lower than ON")
        self.on = on
        self.off = off
        self.active = False
        self.low_since = None
        self.last_event = None

    def update(self, value, now):
        if self.active:
            if value <= self.off:
                if self.low_since is None:
                    self.low_since = now
                elif time.ticks_diff(now, self.low_since) >= RELEASE_MS:
                    self.active = False
                    self.low_since = None
            else:
                self.low_since = None
            return False

        if value >= self.on:
            # クールダウン中の新しい動作も一度ラッチし、遅れて送信しない。
            self.active = True
            allowed = (self.last_event is None or
                       time.ticks_diff(now, self.last_event) >= COOLDOWN_MS)
            if allowed:
                self.last_event = now
                return True
        return False


def status(message):
    print(message)
    if "status" in labels:
        labels["status"].setText(message)


def connect_wifi():
    global wlan
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if wlan.isconnected():
        return
    status("Connecting Wi-Fi...")
    wlan.connect(WIFI_SSID, WIFI_PASSWORD)
    started = time.ticks_ms()
    while not wlan.isconnected():
        M5.update()
        if time.ticks_diff(time.ticks_ms(), started) > 30000:
            raise RuntimeError("Wi-Fi timeout")
        time.sleep_ms(100)


def get_location(response):
    for key, value in response.headers.items():
        if isinstance(key, bytes):
            key = key.decode()
        if key.lower() == "location":
            return value.decode() if isinstance(value, bytes) else value
    return None


def send_event(event, value):
    # 記録済みで応答だけ失われた可能性があるため、自動再送はしません。
    connect_wifi()
    response = None
    try:
        response = requests2.post(
            GAS_URL,
            json={"token": SECRET, "event": event, "value": value},
            headers={"Content-Type": "application/json"},
            timeout=15
        )
        # requests2の版により自動追跡されます。未追跡の場合だけGET。
        for _ in range(3):
            if response.status_code not in (301, 302, 303):
                break
            target = get_location(response)
            if not target or not target.startswith(
                "https://script.googleusercontent.com/"
            ):
                raise RuntimeError("Check GAS access settings")
            response.close()
            response = None
            response = requests2.get(target, timeout=15)

        if response.status_code != 200:
            raise RuntimeError("HTTP " + str(response.status_code))
        try:
            result = response.json()
        except ValueError:
            print("Non-JSON response:", response.text[:300])
            raise RuntimeError("Response is not JSON")
        if not isinstance(result, dict) or result.get("ok") is not True:
            print("GAS response:", result)
            raise RuntimeError("GAS could not save event")
    finally:
        if response is not None:
            response.close()


def init_proximity():
    global proximity_bus
    # CoreS3 Lite 内蔵バス: I2C1 / SCL=11 / SDA=12 / LTR553=0x23
    proximity_bus = I2C(1, scl=Pin(11), sda=Pin(12), freq=100000)
    # 読み取りに失敗すればエラーにする（0を正常値として扱わない）。
    part_id = proximity_bus.readfrom_mem(0x23, 0x86, 1)[0]
    manufacturer = proximity_bus.readfrom_mem(0x23, 0x87, 1)[0]
    print("LTR553 IDs:", hex(part_id), hex(manufacturer))
    if manufacturer != 0x05 or (part_id & 0xF0) != 0x90:
        raise RuntimeError("LTR553 ID mismatch")
    control = proximity_bus.readfrom_mem(0x23, 0x81, 1)[0]
    proximity_bus.writeto_mem(0x23, 0x81, bytes([control & 0xFC]))
    proximity_bus.writeto_mem(0x23, 0x82, b"\x7b")  # 60kHz, 100%, 50mA
    proximity_bus.writeto_mem(0x23, 0x83, b"\x08")  # 8 pulses
    proximity_bus.writeto_mem(0x23, 0x84, b"\x00")  # 50ms
    proximity_bus.writeto_mem(0x23, 0x81, bytes([(control & 0xFC) | 0x03]))
    time.sleep_ms(150)


def read_proximity():
    data = proximity_bus.readfrom_mem(0x23, 0x8D, 2)
    return data[0] | ((data[1] & 0x07) << 8)


def read_shake():
    global previous_accel
    accel = Imu.getAccel()
    if previous_accel is None:
        previous_accel = accel
        return 0.0
    delta = math.sqrt(sum((accel[i] - previous_accel[i]) ** 2
                          for i in range(3)))
    previous_accel = accel
    return delta


def read_sound_and_shake():
    # 録音待ちの間もIMUを読み、短い揺れの取りこぼしを減らす。
    if not Mic.record(audio_buffer, 8000, False):
        raise RuntimeError("Mic.record failed")
    started = time.ticks_ms()
    peak_shake = 0.0
    while Mic.isRecording():
        M5.update()
        peak_shake = max(peak_shake, read_shake())
        if time.ticks_diff(time.ticks_ms(), started) > 2000:
            raise RuntimeError("Microphone timeout")
        time.sleep_ms(5)
    M5.update()
    peak_shake = max(peak_shake, read_shake())

    total = 0
    squares = 0
    count = len(audio_buffer) // 2
    for offset in range(0, len(audio_buffer), 2):
        sample = struct.unpack_from("<h", audio_buffer, offset)[0]
        total += sample
        squares += sample * sample
    mean = total / count
    rms = math.sqrt(max(0.0, squares / count - mean * mean))
    return int(rms), peak_shake


def setup():
    global gates, previous_accel
    M5.begin()
    Widgets.fillScreen(0x222222)
    for name, y, text in (
        ("title", 10, "CoreS3 Sensor Logger"),
        ("near", 50, "Near: --"),
        ("sound", 80, "Sound: --"),
        ("shake", 110, "Shake: --"),
        ("status", 150, "Starting..."),
        ("count", 185, "Saved: 0  Failed: 0")
    ):
        labels[name] = Widgets.Label(
            text, 8, y, 1.0, 0xFFFFFF, 0x222222, Widgets.FONTS.DejaVu18
        )
    if SEND_ENABLED:
        if not GAS_URL.startswith("https://script.google.com/macros/s/") or not GAS_URL.endswith("/exec"):
            raise ValueError("Set GAS_URL to the working /exec URL")
        connect_wifi()
    init_proximity()
    Speaker.end()
    if not Mic.begin():
        raise RuntimeError("Mic.begin failed")
    gates = {
        "near": EventGate(NEAR_ON, NEAR_OFF),
        "sound": EventGate(SOUND_ON, SOUND_OFF),
        "shake": EventGate(SHAKE_ON, SHAKE_OFF)
    }
    status("Warming up...")
    # マイク起動直後のノイズを捨てる。
    for _ in range(10):
        read_sound_and_shake()
    previous_accel = None
    status("Monitoring" if SEND_ENABLED else "Local only")


def loop():
    global last_display, previous_accel, saved_count, failed_count
    M5.update()
    sound, shake = read_sound_and_shake()
    near = read_proximity()
    now = time.ticks_ms()
    values = {"near": near, "sound": sound, "shake": round(shake, 3)}
    # 各イベントを先に判定。1回の観測で複数種類を記録可能。
    events = [(name, value) for name, value in values.items()
              if gates[name].update(value, now)]
    if last_display is None or time.ticks_diff(now, last_display) >= 250 or events:
        for name, value in values.items():
            labels[name].setText(name + ": " + str(value))
        print("near=%d sound=%d shake=%.3f" % (near, sound, shake))
        last_display = now

    for event, value in events:
        print("EVENT:", event, value)
        if not SEND_ENABLED:
            status("Detected: " + event)
            continue
        status("Sending: " + event)
        try:
            send_event(event, value)
            saved_count += 1
            status("Saved: " + event)
        except Exception as error:
            failed_count += 1
            print("Send error:", error)
            print("Event not confirmed; check sheet. No automatic retry.")
            status("Send error: " + event)
        labels["count"].setText("Saved: %d Failed: %d" % (saved_count, failed_count))

    if events and SEND_ENABLED:
        # 通信中は観測していないので、解除時間や加速度差を持ち越さない。
        previous_accel = None
        for gate in gates.values():
            gate.low_since = None
    time.sleep_ms(5)


if __name__ == "__main__":
    try:
        setup()
        while True:
            loop()
    except KeyboardInterrupt:
        status("Stopped")
    except Exception as error:
        import sys
        status("Error: check terminal")
        sys.print_exception(error)
    finally:
        Mic.end()