ブログ

import machine
import network
import ujson
import urequests
import utime

import config

# Seconds in a day
SECONDS_IN_A_DAY = 60 * 60 * 24

# Detect interval (sec)
INTERVAL_SEC = 5

# Number of times to check if the same result is obtained consecutively
HISTORY_LENGTH = 5

# Number of times an exception occurs before reset
MAX_EXCEPTIONS = 9

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

# Wireless LAN
wlan = network.WLAN(network.STA_IF)

# Liquid level sensor
sensor = machine.Pin(28, machine.Pin.IN, machine.Pin.PULL_UP)


def console_print(text):
    """Print to console.

    Args:
        text (str): message.
    """
    print(f"{utime.time()}: {text}")


def connect_wifi():
    """Connect to wifi network.

    Raises:
        RuntimeError: Network connect failure.
    """
    if wlan.status() == 3:
        return
    console_print("Connecting ...")
    led.on()
    wlan.active(True)
    wlan.connect(config.WLAN_SSID, config.WLAN_PASSWORD)
    count = 30
    while 0 < count:
        led.toggle()
        if wlan.status() < 0 or 3 <= wlan.status():
            led.off()
            wlan_status = wlan.ifconfig()
            console_print(f"Connected {wlan_status[0]}.")
            return
        count -= 1
        utime.sleep(0.5)
    led.off()
    raise RuntimeError("Connect failed.")


def disconnect_wifi():
    """Disconnect from wifi network."""
    wlan.disconnect()
    wlan.active(False)
    console_print("Disconnected.")


def send_message(message):
    """Send message using LINE Messaging API

    Args:
        message (str): message
    """
    connect_wifi()
    console_print(message)
    response = urequests.post(
        config.CHANNEL_ENDPOINT_URL,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {config.CHANNEL_ACCESS_TOKEN}",
        },
        data=ujson.dumps(
            {
                "to": config.CHANNEL_USER_ID,
                "messages": [{"type": "text", "text": message}],
            }
        ).encode("utf-8"),
    )
    response.close()
    disconnect_wifi


def get_liquid_level():
    """Detect liquid level

    Returns:
        bool: Liquid detect status
    """
    return sensor.value() == 0


# Initialize
led.off()
results = []
last_message = (None, 0)
exceptions = 0

# Loop
while True:
    try:
        # Get current result
        current_result = get_liquid_level()
        led.value(current_result)

        # Update results
        results.append(current_result)
        results = results[-HISTORY_LENGTH:]

        # Check continuity
        if len(set(results)) == 1:

            # Select message
            if current_result:
                message = (
                    "タンクは水が入っています",
                    utime.time() + 7 * SECONDS_IN_A_DAY,
                )
            else:
                message = (
                    "タンクの水が不足しています!!",
                    utime.time() + 1 * SECONDS_IN_A_DAY,
                )

            # Send message
            if last_message[0] != message[0] or last_message[1] < utime.time():
                send_message(message[0])
                last_message = message

        # Continue
        exceptions = 0
        utime.sleep(INTERVAL_SEC)

    except KeyboardInterrupt:

        # Break
        break

    except Exception as e:

        # Error
        console_print(e)

        # Check force reset
        exceptions += 1
        if MAX_EXCEPTIONS < exceptions:
            console_print("Reset.")
            machine.reset()

        # Continue
        retry_factor = 2 ** min(exceptions - 1, 8)
        retry_duration = INTERVAL_SEC * retry_factor
        console_print(f"Continue after {retry_duration} seconds.")
        utime.sleep(retry_duration)

# Break
console_print("Break.")

WLAN_SSID = "WiFi の SSID"
WLAN_PASSWORD = "WiFi のパスワード"

CHANNEL_ENDPOINT_URL = "https://api.line.me/v2/bot/message/push"
CHANNEL_ACCESS_TOKEN = "自分のチャネルアクセストークン"
CHANNEL_USER_ID = "自分の LINE のユーザ ID"