Initial commit

.gitignore

@@ -0,0 +1 @@
+secrets.py

.micropico

@@ -0,0 +1,3 @@
+{
+    "info": "This file is just used to identify a project folder."
+}

.vscode/extensions.json

@@ -0,0 +1,8 @@
+{
+    "recommendations": [
+        "ms-python.python",
+        "visualstudioexptteam.vscodeintellicode",
+        "ms-python.vscode-pylance",
+        "paulober.pico-w-go"
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,15 @@
+{
+    "python.languageServer": "Pylance",
+    "python.analysis.typeCheckingMode": "basic",
+    "python.analysis.diagnosticSeverityOverrides": {
+        "reportMissingModuleSource": "none"
+    },
+    "python.terminal.activateEnvironment": false,
+    "micropico.openOnStart": true,
+    "python.analysis.typeshedPaths": [
+        "~/.micropico-stubs/included"
+    ],
+    "python.analysis.extraPaths": [
+        "~/.micropico-stubs/included"
+    ]
+}

README.md

@@ -0,0 +1,7 @@
+# Sensor Code
+
+The outdoor temp/humidity sensor code written in microPython
+
+## Installation
+
+Create a file ``secrets.py`` and add two variables ``SSID`` and ``PASSWORD``

dht20.py

@@ -0,0 +1,95 @@
+# dht20.py
+
+from machine import I2C
+from utime import sleep_ms
+
+
+class DHT20:
+    """Class for the DHT20 Temperature and Humidity Sensor."""
+
+    def __init__(self, address: int, i2c: I2C):
+        self._address = address
+        self._i2c = i2c
+        sleep_ms(100)
+
+        if not self.is_ready:
+            self._initialize()
+            sleep_ms(100)
+
+            if not self.is_ready:
+                raise RuntimeError("Could not initialize the DHT20.")
+
+    @property
+    def is_ready(self) -> bool:
+        """Check if the DHT20 is ready."""
+        self._i2c.writeto(self._address, bytearray(b"\x71"))
+        return self._i2c.readfrom(self._address, 1)[0] == 0x18
+
+    def _initialize(self):
+        buffer = bytearray(b"\x00\x00")
+        self._i2c.writeto_mem(self._address, 0x1B, buffer)
+        self._i2c.writeto_mem(self._address, 0x1C, buffer)
+        self._i2c.writeto_mem(self._address, 0x1E, buffer)
+
+    def _trigger_measurements(self):
+        self._i2c.writeto_mem(self._address, 0xAC, bytearray(b"\x33\x00"))
+
+    def _read_measurements(self):
+        buffer = self._i2c.readfrom(self._address, 7)
+        return buffer, buffer[0] & 0x80 == 0
+
+    def _crc_check(self, input_bitstring: str, check_value: str) -> bool:
+        """Calculate the CRC check of a string of bits using a fixed polynomial."""
+
+        polynomial_bitstring = "100110001"
+        len_input = len(input_bitstring)
+        initial_padding = check_value
+        input_padded_array = list(input_bitstring + initial_padding)
+
+        while "1" in input_padded_array[:len_input]:
+            cur_shift = input_padded_array.index("1")
+
+            for i in range(len(polynomial_bitstring)):
+                input_padded_array[cur_shift + i] = str(
+                    int(polynomial_bitstring[i] != input_padded_array[cur_shift + i])
+                )
+
+        return "1" not in "".join(input_padded_array)[len_input:]
+
+    @property
+    def measurements(self) -> dict:
+        """Get the temperature (°C) and relative humidity (%RH).
+
+        Returns a dictionary with the most recent measurements.
+
+        't': temperature (°C),
+        't_adc': the 'raw' temperature as produced by the ADC,
+        'rh': relative humidity (%RH),
+        'rh_adc': the 'raw' relative humidity as produced by the ADC,
+        'crc_ok': indicates if the data was received correctly
+        """
+        self._trigger_measurements()
+        sleep_ms(50)
+
+        data = self._read_measurements()
+        retry = 3
+
+        while not data[1]:
+            if not retry:
+                raise RuntimeError("Could not read measurements from the DHT20.")
+
+            sleep_ms(10)
+            data = self._read_measurements()
+            retry -= 1
+
+        buffer = data[0]
+        s_rh = buffer[1] << 12 | buffer[2] << 4 | buffer[3] >> 4
+        s_t = (buffer[3] << 16 | buffer[4] << 8 | buffer[5]) & 0xFFFFF
+        rh = (s_rh / 2**20) * 100
+        t = ((s_t / 2**20) * 200) - 50
+        crc_ok = self._crc_check(
+            f"{buffer[0] ^ 0xFF:08b}{buffer[1]:08b}{buffer[2]:08b}{buffer[3]:08b}{buffer[4]:08b}{buffer[5]:08b}",
+            f"{buffer[6]:08b}",
+        )
+
+        return {"t": t, "t_adc": s_t, "rh": rh, "rh_adc": s_rh, "crc_ok": crc_ok}

main.py

@@ -0,0 +1,176 @@
+# 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")

secrets.example.py

@@ -0,0 +1,4 @@
+# secrets.py
+
+SSID = "My WiFi Name"
+PASSWORD = "MySuperSecretPassword"