From 2768616be445ad12cf0df90e03957baeb1cc13f9 Mon Sep 17 00:00:00 2001 From: Gregory Ballantine Date: Sat, 12 Sep 2026 17:25:55 -0400 Subject: [PATCH] Refactored some classes into separate files in lib/ --- .gitignore | 1 - install.sh | 4 +- lib/bme280.py | 101 +++++++++++++++++++ lib/sht41.py | 36 +++++++ lib/simple_mqtt.py | 78 +++++++++++++++ main.py | 238 +++------------------------------------------ 6 files changed, 228 insertions(+), 230 deletions(-) create mode 100644 lib/bme280.py create mode 100644 lib/sht41.py create mode 100644 lib/simple_mqtt.py diff --git a/.gitignore b/.gitignore index 2b96b49..098d98c 100644 --- a/.gitignore +++ b/.gitignore @@ -18,7 +18,6 @@ dist/ downloads/ eggs/ .eggs/ -lib/ lib64/ parts/ sdist/ diff --git a/install.sh b/install.sh index 964b9aa..a28b79e 100755 --- a/install.sh +++ b/install.sh @@ -1,10 +1,10 @@ #!/bin/bash -APP_FILES=('main.py' 'config.py') +APP_FILES=('main.py' 'config.py' 'lib/') APP_PACKAGES='./package.json' # Copy required files -mpremote cp "${APP_FILES[@]}" : +mpremote cp -r "${APP_FILES[@]}" : # Install dependencies mpremote mip install ./package.json diff --git a/lib/bme280.py b/lib/bme280.py new file mode 100644 index 0000000..fb7599a --- /dev/null +++ b/lib/bme280.py @@ -0,0 +1,101 @@ +import struct +import time + + +class BME280: + DEFAULT_ADDR = 0x77 + + def __init__(self, i2c, addr=DEFAULT_ADDR): + self.i2c = i2c + self.addr = addr + + # Reset sensor + self.i2c.writeto_mem(self.addr, 0xE0, b"\xb6") + time.sleep(0.1) + + self._read_calibration_data() + + # Humidity oversampling x1 + self.i2c.writeto_mem(self.addr, 0xF2, b"\x01") + # Pressure oversampling x1, Temp oversampling x1, Normal mode + self.i2c.writeto_mem(self.addr, 0xF4, b"\x27") + # Standby 1000ms, filter off + self.i2c.writeto_mem(self.addr, 0xF5, b"\xa0") + + def _read_calibration_data(self): + calib = self.i2c.readfrom_mem(self.addr, 0x88, 24) + c = struct.unpack("> 4) | (e5_e6[1] << 4) + + self.dig_h6 = struct.unpack("> 4 + raw_t = ((raw[3] << 16) | (raw[4] << 8) | raw[5]) >> 4 + raw_h = (raw[6] << 8) | raw[7] + + # Temperature calculation (Celsius) + var1 = (((raw_t >> 3) - (self.dig_t1 << 1)) * self.dig_t2) >> 11 + var2 = ( + (((raw_t >> 4) - self.dig_t1) * ((raw_t >> 4) - self.dig_t1)) >> 12 + ) * self.dig_t3 >> 14 + t_fine = var1 + var2 + temp = ((t_fine * 5 + 128) >> 8) / 100.0 + + # Pressure calculation (hPa) + p_var1 = t_fine - 128000 + p_var2 = ( + p_var1 * p_var1 * self.dig_p6 + ((p_var1 * self.dig_p5) << 17) + (self.dig_p4 << 35) + ) + p_var1 = ((p_var1 * p_var1 * self.dig_p3) >> 8) + ((p_var1 * self.dig_p2) << 12) + p_var1 = (((1 << 47) + p_var1) * self.dig_p1) >> 33 + + if p_var1 == 0: + pressure = 0.0 + else: + p = 1048576 - raw_p + p = (((p << 31) - p_var2) * 3125) // p_var1 + p_var1 = (self.dig_p9 * (p >> 13) * (p >> 13)) >> 25 + p_var2 = (self.dig_p8 * p) >> 19 + p = ((p + p_var1 + p_var2) >> 8) + (self.dig_p7 << 4) + pressure = (p / 256.0) / 100.0 + + # Humidity calculation (%) + h_var = t_fine - 76800 + h_var = ((((raw_h << 14) - (self.dig_h4 << 20) - (self.dig_h5 * h_var)) + 16384) >> 15) * ( + ( + ( + ( + (((h_var * self.dig_h6) >> 10) * (((h_var * self.dig_h3) >> 11) + 32768)) + >> 10 + ) + + 2097152 + ) + * self.dig_h2 + + 8192 + ) + >> 14 + ) + h_var = h_var - (((((h_var >> 15) * (h_var >> 15)) >> 7) * self.dig_h1) >> 4) + h_var = max(0, min(h_var, 419430400)) + humidity = (h_var >> 12) / 1024.0 + + return { + "temperature": round(temp, 2), + "pressure": round(pressure, 2), + "humidity": round(humidity, 2), + } diff --git a/lib/sht41.py b/lib/sht41.py new file mode 100644 index 0000000..410d98e --- /dev/null +++ b/lib/sht41.py @@ -0,0 +1,36 @@ +import time + + +class SHT41: + DEFAULT_ADDR = 0x44 + CMD_MEASURE_HIGH_PRECISION = 0xFD + + def __init__(self, i2c, addr=DEFAULT_ADDR): + self.i2c = i2c + self.addr = addr + # Soft reset + self.i2c.writeto(self.addr, b"\x94") + time.sleep(0.01) + + def read_values(self): + # Trigger high precision measurement + self.i2c.writeto(self.addr, bytes([self.CMD_MEASURE_HIGH_PRECISION])) + # Conversion takes up to 8.2ms + time.sleep(0.01) + + # 6 bytes: [Temp_MSB, Temp_LSB, CRC, Hum_MSB, Hum_LSB, CRC] + data = self.i2c.readfrom(self.addr, 6) + + raw_t = (data[0] << 8) | data[1] + raw_h = (data[3] << 8) | data[4] + + # Sensirion SHT4x conversion equations + temp = -45.0 + 175.0 * (raw_t / 65535.0) + hum = -6.0 + 125.0 * (raw_h / 65535.0) + hum = max(0.0, min(hum, 100.0)) + + return { + "temperature": round(temp, 2), + "humidity": round(hum, 2), + "pressure": None, + } diff --git a/lib/simple_mqtt.py b/lib/simple_mqtt.py new file mode 100644 index 0000000..827d32e --- /dev/null +++ b/lib/simple_mqtt.py @@ -0,0 +1,78 @@ +import socket +import struct + + +class SimpleMQTT: + def __init__(self, host, port=1883): + self.host = host + self.port = port + self.sock = None + + def connect(self, client_id, username=None, password=None): + addr_info = socket.getaddrinfo(self.host, self.port) + addr = addr_info[0][-1] + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.connect(addr) + + # Flags: Clean session (bit 1) + optional credentials + flags = 0x02 + if username: + flags |= 0x80 + if password: + flags |= 0x40 + + # Variable header: Protocol Name (MQTT) + Level (4) + Flags + KeepAlive (60s) + var_header = bytearray( + [0x00, 0x04, ord("M"), ord("Q"), ord("T"), ord("T"), 0x04, flags, 0x00, 0x3C] + ) + + payload = bytearray() + payload.extend(self._encode_str(client_id)) + if username: + payload.extend(self._encode_str(username)) + if password: + payload.extend(self._encode_str(password)) + + body = var_header + payload + packet = bytearray([0x10]) + self._encode_len(len(body)) + body + self.sock.write(packet) + + # Read CONNACK (4 bytes: 0x20, 0x02, ack_flags, return_code) + resp = self.sock.read(4) + if not resp or resp[0] != 0x20 or resp[3] != 0x00: + code = resp[3] if resp and len(resp) >= 4 else "nil" + raise RuntimeError(f"MQTT connection failed: code {code}") + + def publish(self, topic, payload, retain=False): + cmd = 0x30 | (0x01 if retain else 0x00) + if isinstance(payload, str): + payload = payload.encode("utf-8") + body = self._encode_str(topic) + payload + packet = bytearray([cmd]) + self._encode_len(len(body)) + body + self.sock.write(packet) + + def disconnect(self): + try: + if self.sock: + self.sock.write(bytearray([0xE0, 0x00])) + self.sock.close() + except Exception: + pass + finally: + self.sock = None + + def _encode_str(self, s): + raw = s.encode("utf-8") + return struct.pack("!H", len(raw)) + raw + + def _encode_len(self, length): + encoded = bytearray() + while True: + digit = length % 128 + length //= 128 + if length > 0: + digit |= 0x80 + encoded.append(digit) + if length == 0: + break + return encoded diff --git a/main.py b/main.py index 8c3c55c..adca8e2 100644 --- a/main.py +++ b/main.py @@ -1,8 +1,7 @@ -import socket -import struct import time from machine import Pin, SoftI2C import network +from simple_mqtt import SimpleMQTT # ------------------------------------------------------------- # Configuration Loader @@ -19,221 +18,13 @@ NODE_NAME = getattr(Config, "NODE_NAME", Config.NODE_ID) SENSOR_TYPE = getattr(Config, "SENSOR_TYPE", "bme280").lower() STATE_TOPIC = f"homeassistant/sensor/{Config.NODE_ID}/state" - -# ------------------------------------------------------------- -# Minimal MQTT 3.1.1 Client -# ------------------------------------------------------------- -class SimpleMQTT: - def __init__(self, host, port=1883): - self.host = host - self.port = port - self.sock = None - - def connect(self, client_id, username=None, password=None): - addr_info = socket.getaddrinfo(self.host, self.port) - addr = addr_info[0][-1] - self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.sock.connect(addr) - - flags = 0x02 - if username: - flags |= 0x80 - if password: - flags |= 0x40 - - var_header = bytearray( - [0x00, 0x04, ord("M"), ord("Q"), ord("T"), ord("T"), 0x04, flags, 0x00, 0x3C] - ) - - payload = bytearray() - payload.extend(self._encode_str(client_id)) - if username: - payload.extend(self._encode_str(username)) - if password: - payload.extend(self._encode_str(password)) - - body = var_header + payload - packet = bytearray([0x10]) + self._encode_len(len(body)) + body - self.sock.write(packet) - - resp = self.sock.read(4) - if not resp or resp[0] != 0x20 or resp[3] != 0x00: - code = resp[3] if resp and len(resp) >= 4 else "nil" - raise RuntimeError(f"MQTT connection failed: code {code}") - - def publish(self, topic, payload, retain=False): - cmd = 0x30 | (0x01 if retain else 0x00) - if isinstance(payload, str): - payload = payload.encode("utf-8") - body = self._encode_str(topic) + payload - packet = bytearray([cmd]) + self._encode_len(len(body)) + body - self.sock.write(packet) - - def disconnect(self): - try: - if self.sock: - self.sock.write(bytearray([0xE0, 0x00])) - self.sock.close() - except Exception: - pass - finally: - self.sock = None - - def _encode_str(self, s): - raw = s.encode("utf-8") - return struct.pack("!H", len(raw)) + raw - - def _encode_len(self, length): - encoded = bytearray() - while True: - digit = length % 128 - length //= 128 - if length > 0: - digit |= 0x80 - encoded.append(digit) - if length == 0: - break - return encoded - - -# ------------------------------------------------------------- -# BME280 Driver (I2C) -# ------------------------------------------------------------- -class BME280: - DEFAULT_ADDR = 0x77 - - def __init__(self, i2c, addr=DEFAULT_ADDR): - self.i2c = i2c - self.addr = addr - - # Reset sensor - self.i2c.writeto_mem(self.addr, 0xE0, b"\xb6") - time.sleep(0.1) - - self._read_calibration_data() - - # Humidity oversampling x1 - self.i2c.writeto_mem(self.addr, 0xF2, b"\x01") - # Pressure oversampling x1, Temp oversampling x1, Normal mode - self.i2c.writeto_mem(self.addr, 0xF4, b"\x27") - # Standby 1000ms, filter off - self.i2c.writeto_mem(self.addr, 0xF5, b"\xa0") - - def _read_calibration_data(self): - calib = self.i2c.readfrom_mem(self.addr, 0x88, 24) - c = struct.unpack("> 4) | (e5_e6[1] << 4) - - self.dig_h6 = struct.unpack("> 4 - raw_t = ((raw[3] << 16) | (raw[4] << 8) | raw[5]) >> 4 - raw_h = (raw[6] << 8) | raw[7] - - # Temperature calculation (Celsius) - var1 = (((raw_t >> 3) - (self.dig_t1 << 1)) * self.dig_t2) >> 11 - var2 = ( - (((raw_t >> 4) - self.dig_t1) * ((raw_t >> 4) - self.dig_t1)) >> 12 - ) * self.dig_t3 >> 14 - t_fine = var1 + var2 - temp = ((t_fine * 5 + 128) >> 8) / 100.0 - - # Pressure calculation (hPa) - p_var1 = t_fine - 128000 - p_var2 = ( - p_var1 * p_var1 * self.dig_p6 + ((p_var1 * self.dig_p5) << 17) + (self.dig_p4 << 35) - ) - p_var1 = ((p_var1 * p_var1 * self.dig_p3) >> 8) + ((p_var1 * self.dig_p2) << 12) - p_var1 = (((1 << 47) + p_var1) * self.dig_p1) >> 33 - - if p_var1 == 0: - pressure = 0.0 - else: - p = 1048576 - raw_p - p = (((p << 31) - p_var2) * 3125) // p_var1 - p_var1 = (self.dig_p9 * (p >> 13) * (p >> 13)) >> 25 - p_var2 = (self.dig_p8 * p) >> 19 - p = ((p + p_var1 + p_var2) >> 8) + (self.dig_p7 << 4) - pressure = (p / 256.0) / 100.0 - - # Humidity calculation (%) - h_var = t_fine - 76800 - h_var = ((((raw_h << 14) - (self.dig_h4 << 20) - (self.dig_h5 * h_var)) + 16384) >> 15) * ( - ( - ( - ( - (((h_var * self.dig_h6) >> 10) * (((h_var * self.dig_h3) >> 11) + 32768)) - >> 10 - ) - + 2097152 - ) - * self.dig_h2 - + 8192 - ) - >> 14 - ) - h_var = h_var - (((((h_var >> 15) * (h_var >> 15)) >> 7) * self.dig_h1) >> 4) - h_var = max(0, min(h_var, 419430400)) - humidity = (h_var >> 12) / 1024.0 - - return { - "temperature": round(temp, 2), - "pressure": round(pressure, 2), - "humidity": round(humidity, 2), - } - - -# ------------------------------------------------------------- -# SHT41 Driver (I2C) -# ------------------------------------------------------------- -class SHT41: - DEFAULT_ADDR = 0x44 - CMD_MEASURE_HIGH_PRECISION = 0xFD - - def __init__(self, i2c, addr=DEFAULT_ADDR): - self.i2c = i2c - self.addr = addr - # Soft reset - self.i2c.writeto(self.addr, b"\x94") - time.sleep(0.01) - - def read_values(self): - # Trigger high precision measurement - self.i2c.writeto(self.addr, bytes([self.CMD_MEASURE_HIGH_PRECISION])) - # Conversion takes up to 8.2ms - time.sleep(0.01) - - # 6 bytes: [Temp_MSB, Temp_LSB, CRC, Hum_MSB, Hum_LSB, CRC] - data = self.i2c.readfrom(self.addr, 6) - - raw_t = (data[0] << 8) | data[1] - raw_h = (data[3] << 8) | data[4] - - # Official Sensirion conversion equations - temp = -45.0 + 175.0 * (raw_t / 65535.0) - hum = -6.0 + 125.0 * (raw_h / 65535.0) - hum = max(0.0, min(hum, 100.0)) - - return { - "temperature": round(temp, 2), - "humidity": round(hum, 2), - "pressure": None, - } +# Dynamic sensor driver import +if SENSOR_TYPE == "sht41": + from sht41 import SHT41 +elif SENSOR_TYPE == "bme280": + from bme280 import BME280 +else: + raise ValueError(f"Unsupported SENSOR_TYPE '{SENSOR_TYPE}'. Use 'bme280' or 'sht41'.") # ------------------------------------------------------------- @@ -266,7 +57,6 @@ def register_ha_discovery(mqtt, sensor_type): }, ] - # Only register atmospheric pressure if the sensor provides it if sensor_type == "bme280": configs.append( { @@ -297,7 +87,7 @@ def register_ha_discovery(mqtt, sensor_type): # ------------------------------------------------------------- -# Instantiate Wi-Fi connection +# Wi-Fi Connection # ------------------------------------------------------------- def connect_wifi(): wlan = network.WLAN(network.STA_IF) @@ -316,7 +106,7 @@ def connect_wifi(): # ------------------------------------------------------------- -# Panic when the sensor isn't ready +# Panic Handler # ------------------------------------------------------------- def panic(err=None): if err: @@ -333,15 +123,12 @@ def panic(err=None): def main(): i2c = SoftI2C(sda=Pin(4), scl=Pin(5), freq=100000) - # Initialize selected sensor if SENSOR_TYPE == "sht41": print("Using SHT41 sensor (0x44)") sensor = SHT41(i2c) - elif SENSOR_TYPE == "bme280": + else: print("Using BME280 sensor (0x77)") sensor = BME280(i2c) - else: - raise ValueError(f"Unsupported SENSOR_TYPE '{SENSOR_TYPE}'. Use 'bme280' or 'sht41'.") connect_wifi() @@ -354,16 +141,13 @@ def main(): mqtt.connect(Config.CLIENT_ID, mqtt_user, mqtt_pass) print("MQTT connected.") - # Publish discovery retained messages register_ha_discovery(mqtt, SENSOR_TYPE) - # Telemetry loop interval = getattr(Config, "UPDATE_INTERVAL", 30) while True: try: data = sensor.read_values() - # Format payload conditionally based on available readings if data.get("pressure") is not None: state_json = ( f'{{"temperature":{data["temperature"]},'