BitGoblin/pico-thermo
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Refactored some classes into separate files in lib/
ci/woodpecker/push/woodpecker Pipeline failed
Details

Browse Source
This commit is contained in:
Gregory Ballantine
2026-09-12 17:25:55 -04:00
parent 31c7f870e0
commit 2768616be4
6 changed files with 228 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
-1
View File
@@ -18,7 +18,6 @@ dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
install.sh
+2 -2
View File
@@ -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
lib/bme280.py
+101
View File
@@ -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("<HhhHhhhhhhhh", calib)
self.dig_t1, self.dig_t2, self.dig_t3 = c[0], c[1], c[2]
self.dig_p1, self.dig_p2, self.dig_p3 = c[3], c[4], c[5]
self.dig_p4, self.dig_p5, self.dig_p6 = c[6], c[7], c[8]
self.dig_p7, self.dig_p8, self.dig_p9 = c[9], c[10], c[11]
self.dig_h1 = self.i2c.readfrom_mem(self.addr, 0xA1, 1)[0]
h = struct.unpack("<hB", self.i2c.readfrom_mem(self.addr, 0xE1, 3)[:3])
self.dig_h2, self.dig_h3 = h[0], h[1]
e4 = self.i2c.readfrom_mem(self.addr, 0xE4, 2)
self.dig_h4 = (e4[0] << 4) | (e4[1] & 0x0F)
e5_e6 = self.i2c.readfrom_mem(self.addr, 0xE5, 2)
self.dig_h5 = (e5_e6[0] >> 4) | (e5_e6[1] << 4)
self.dig_h6 = struct.unpack("<b", self.i2c.readfrom_mem(self.addr, 0xE7, 1))[0]
def read_values(self):
raw = self.i2c.readfrom_mem(self.addr, 0xF7, 8)
raw_p = ((raw[0] << 16) | (raw[1] << 8) | raw[2]) >> 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),
}
lib/sht41.py
+36
View File
@@ -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,
}
lib/simple_mqtt.py
+78
View File
@@ -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
main.py
+10 -226
View File
@@ -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("<HhhHhhhhhhhh", calib)
self.dig_t1, self.dig_t2, self.dig_t3 = c[0], c[1], c[2]
self.dig_p1, self.dig_p2, self.dig_p3 = c[3], c[4], c[5]
self.dig_p4, self.dig_p5, self.dig_p6 = c[6], c[7], c[8]
self.dig_p7, self.dig_p8, self.dig_p9 = c[9], c[10], c[11]
self.dig_h1 = self.i2c.readfrom_mem(self.addr, 0xA1, 1)[0]
h = struct.unpack("<hB", self.i2c.readfrom_mem(self.addr, 0xE1, 3)[:3])
self.dig_h2, self.dig_h3 = h[0], h[1]
e4 = self.i2c.readfrom_mem(self.addr, 0xE4, 2)
self.dig_h4 = (e4[0] << 4) | (e4[1] & 0x0F)
e5_e6 = self.i2c.readfrom_mem(self.addr, 0xE5, 2)
self.dig_h5 = (e5_e6[0] >> 4) | (e5_e6[1] << 4)
self.dig_h6 = struct.unpack("<b", self.i2c.readfrom_mem(self.addr, 0xE7, 1))[0]
def read_values(self):
raw = self.i2c.readfrom_mem(self.addr, 0xF7, 8)
raw_p = ((raw[0] << 16) | (raw[1] << 8) | raw[2]) >> 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
# Dynamic sensor driver import
if SENSOR_TYPE == "sht41":
from sht41 import SHT41
elif SENSOR_TYPE == "bme280":
from bme280 import BME280
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,
}
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"]},'