weather-sensor/main.py

# main.py

# Import necessary modules
import network
import asyncio
import json
from math import log
from machine import ADC, I2C, Pin
from collections import OrderedDict

import secrets
from dht20 import DHT20

# Set pinouts
led = Pin("LED", Pin.OUT)
i2c0_sda = Pin(0)
i2c0_scl = Pin(1)
i2c0 = I2C(0, sda=i2c0_sda, scl=i2c0_scl)

# Initiate sensors
temp_sensor = DHT20(0x38, i2c0)
UV = ADC(Pin(26))

# Dict for WLAN status codes
WLAN_STAT = {
    network.STAT_NO_AP_FOUND: "Wi-Fi not found",
    network.STAT_CONNECT_FAIL: "Connection failed!",
    network.STAT_CONNECTING: "Connecting",
    network.STAT_GOT_IP: "Got IP-address",
    network.STAT_IDLE: "Idle",
    network.STAT_WRONG_PASSWORD: "Wrong Password!",
}


# Get sensor data
def get_sensor_data():
    try:
        measurements = temp_sensor.measurements
        temp = round(measurements["t"], 1)
        humidity = round(measurements["rh"], 1)
    except Exception as e:
        print(f"DHT20 Error: {e}")
        temp = "--"
        humidity = "--"

    try:
        uv_index = int((UV.read_u16() * 3.3 / 65535) / 0.1)
    except Exception as e:
        print(f"UV Sensor Error: {e}")
        uv_index = "--"

    try:
        # Ensure we only calculate dewpoint if we have real numbers
        if temp != "--" and humidity != "--":
            gamma = (17.62 * float(temp) / (243.12 + float(temp))) + log(
                float(humidity) / 100.0
            )
            dewpoint = round((243.12 * gamma) / (17.62 - gamma), 1)
        else:
            dewpoint = "--"
    except Exception as e:
        print(f"Dewpoint Calc Error: {e}")
        dewpoint = "--"

    return temp, humidity, dewpoint, uv_index


# Generate JSON payload with sensor data
def json_payload():
    temp, humidity, dewpoint, uv_index = get_sensor_data()
    data = OrderedDict(
        [
            ("device_id", "ALPHA_1"),
            ("temperature", temp),
            ("humidity", humidity),
            ("dew", dewpoint),
            ("uv_index", uv_index),
        ]
    )
    return json.dumps(data)


# Asynchronous LED blinking function
async def async_blink_led(blinks=1, time_ms=100):
    for _ in range(blinks):
        led.on()
        await asyncio.sleep(time_ms / 1000)
        led.off()
        await asyncio.sleep(time_ms / 1000)


# Init Wi-Fi Interface (Converted to async to prevent blocking)
async def init_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.config(hostname="weathersensor")
    wlan.active(True)

    # Connect to your network using secrets.py
    wlan.connect(secrets.SSID, secrets.PASSWORD)

    # Wait for Wi-Fi connection
    connection_timeout = 10
    for x in range(connection_timeout):
        print(
            f"Waiting for Wi-Fi connection... timeout in {(connection_timeout-x):02d} sec"
        )
        print(WLAN_STAT.get(wlan.status(), "Unknown Status"), end="\033[F")
        if wlan.status() >= network.STAT_GOT_IP:
            break
        await async_blink_led()
        await asyncio.sleep(0.9)

    # Check if connection is successful
    if wlan.status() != network.STAT_GOT_IP:
        print("\x1b[2KFailed to connect to Wi-Fi")
        return False
    else:
        print("\x1b[2KConnection successful!")
        network_info = wlan.ifconfig()
        print(f"IP address: {network_info[0]}")
        await async_blink_led(blinks=5)
        return True


# Asynchronous function to handle client requests
async def handle_client(reader, writer):
    print("Sending data... ", end="")
    # Skip HTTP request headers
    while await reader.readline() != b"\r\n":
        pass

    # Generate JSON response
    response = json_payload()

    # Send the JSON response and close the connection
    writer.write("HTTP/1.0 200 OK\r\nContent-type: application/json\r\n\r\n")
    writer.write(response)
    await writer.drain()
    writer.close()
    await writer.wait_closed()

    print("Success")
    await async_blink_led(2)


# Main function
async def main():
    print("Starting up the script")
    await async_blink_led(blinks=3)

    if not await init_wifi():
        print("Exiting program due to network failure...")
        return

    # Start the server using the modern uasyncio method
    print("Setting up server")
    server = await asyncio.start_server(handle_client, "0.0.0.0", 80)

    try:
        # Keep the main loop alive so the background server can keep running
        while True:
            await asyncio.sleep(3600)
    finally:
        # If the loop breaks (e.g., KeyboardInterrupt), gracefully close the server
        print("Shutting down server...")
        server.close()
        await server.wait_closed()


# Modern Asyncio Execution
try:
    asyncio.run(main())
except Exception as e:
    print(f"Error occurred: {e}")
except KeyboardInterrupt:
    print("Program Interrupted by the user")