commit 29c698a12a90c4a8608843863337415f7dd1e127 Author: William Söderberg Date: Thu Mar 19 16:05:34 2026 +0100 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c6b6498 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +secrets.py \ No newline at end of file diff --git a/.micropico b/.micropico new file mode 100644 index 0000000..3de3977 --- /dev/null +++ b/.micropico @@ -0,0 +1,3 @@ +{ + "info": "This file is just used to identify a project folder." +} \ No newline at end of file diff --git a/.vscode/extensions.json b/.vscode/extensions.json new file mode 100644 index 0000000..fbc7999 --- /dev/null +++ b/.vscode/extensions.json @@ -0,0 +1,8 @@ +{ + "recommendations": [ + "ms-python.python", + "visualstudioexptteam.vscodeintellicode", + "ms-python.vscode-pylance", + "paulober.pico-w-go" + ] +} \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..03be782 --- /dev/null +++ b/.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" + ] +} \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..38b7702 --- /dev/null +++ b/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`` diff --git a/dht20.py b/dht20.py new file mode 100644 index 0000000..347756b --- /dev/null +++ b/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} diff --git a/main.py b/main.py new file mode 100644 index 0000000..e12be0e --- /dev/null +++ b/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") diff --git a/secrets.example.py b/secrets.example.py new file mode 100644 index 0000000..1b47506 --- /dev/null +++ b/secrets.example.py @@ -0,0 +1,4 @@ +# secrets.py + +SSID = "My WiFi Name" +PASSWORD = "MySuperSecretPassword"