323 lines
12 KiB
Python
323 lines
12 KiB
Python
import time
|
|
import math
|
|
import board
|
|
import smbus2
|
|
import serial
|
|
import pynmea2
|
|
import bme680
|
|
import adafruit_ltr390
|
|
import threading
|
|
from datetime import datetime
|
|
|
|
bus = smbus2.SMBus(1)
|
|
i2c = board.I2C()
|
|
|
|
def try_read(fn):
|
|
try:
|
|
return fn()
|
|
except:
|
|
return None
|
|
|
|
# ── BME680 (0x77) ────────────────────────────────────────────────────────────
|
|
|
|
GAS_REFERENCE = 250000
|
|
|
|
def gas_to_aqi(gas_ohms, humidity):
|
|
if gas_ohms is None:
|
|
return None
|
|
gas_score = min(gas_ohms / GAS_REFERENCE, 1.0) * 75
|
|
hum_score = (25 - abs(humidity - 40) * 0.5)
|
|
aqi = min(max(round(gas_score + hum_score), 0), 100)
|
|
if aqi >= 80: label = "Excellent"
|
|
elif aqi >= 60: label = "Good"
|
|
elif aqi >= 40: label = "Fair"
|
|
elif aqi >= 20: label = "Poor"
|
|
else: label = "Very Poor"
|
|
return {'score': aqi, 'label': label}
|
|
|
|
class BME680:
|
|
def __init__(self):
|
|
self._cache = None
|
|
self.sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY)
|
|
self.sensor.set_humidity_oversample(bme680.OS_2X)
|
|
self.sensor.set_pressure_oversample(bme680.OS_4X)
|
|
self.sensor.set_temperature_oversample(bme680.OS_8X)
|
|
self.sensor.set_filter(bme680.FILTER_SIZE_3)
|
|
self.sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
|
|
self.sensor.set_gas_heater_temperature(320)
|
|
self.sensor.set_gas_heater_duration(150)
|
|
self.sensor.select_gas_heater_profile(0)
|
|
|
|
def read(self):
|
|
if self.sensor.get_sensor_data():
|
|
dew_point = self.sensor.data.temperature - ((100 - self.sensor.data.humidity) / 5.0)
|
|
gas_ohms = round(self.sensor.data.gas_resistance, 0) if self.sensor.data.heat_stable else None
|
|
self._cache = {
|
|
'temp_c': round(self.sensor.data.temperature, 2),
|
|
'temp_f': round(self.sensor.data.temperature * 9/5 + 32, 2),
|
|
'humidity': round(self.sensor.data.humidity, 2),
|
|
'dew_point': round(dew_point, 2),
|
|
'pressure': round(self.sensor.data.pressure, 2),
|
|
'gas_ohms': gas_ohms,
|
|
'aqi': gas_to_aqi(gas_ohms, self.sensor.data.humidity),
|
|
}
|
|
return self._cache
|
|
|
|
# ── MPU6050 (0x68) ───────────────────────────────────────────────────────────
|
|
|
|
class MPU6050:
|
|
ADDR = 0x68
|
|
CALIBRATION_SAMPLES = 200
|
|
|
|
def __init__(self):
|
|
try_read(lambda: bus.write_byte_data(self.ADDR, 0x6B, 0))
|
|
time.sleep(0.1)
|
|
self.offsets = self._calibrate()
|
|
self._peak = None
|
|
self._lock = threading.Lock()
|
|
threading.Thread(target=self._sample_loop, daemon=True).start()
|
|
|
|
def _calibrate(self):
|
|
print("Calibrating MPU6050, keep still...")
|
|
sums = [0] * 6
|
|
for _ in range(self.CALIBRATION_SAMPLES):
|
|
d = try_read(lambda: bus.read_i2c_block_data(self.ADDR, 0x3B, 14))
|
|
if d:
|
|
def s(v): return v - 65536 if v > 32767 else v
|
|
sums[0] += s((d[0] << 8) | d[1]) / 16384.0
|
|
sums[1] += s((d[2] << 8) | d[3]) / 16384.0
|
|
sums[2] += s((d[4] << 8) | d[5]) / 16384.0
|
|
sums[3] += s((d[8] << 8) | d[9]) / 131.0
|
|
sums[4] += s((d[10] << 8) | d[11]) / 131.0
|
|
sums[5] += s((d[12] << 8) | d[13]) / 131.0
|
|
time.sleep(0.005)
|
|
n = self.CALIBRATION_SAMPLES
|
|
return [sums[0]/n, sums[1]/n, sums[2]/n - 1.0,
|
|
sums[3]/n, sums[4]/n, sums[5]/n]
|
|
|
|
def _sample_loop(self):
|
|
while True:
|
|
d = try_read(lambda: bus.read_i2c_block_data(self.ADDR, 0x3B, 14))
|
|
if d:
|
|
def s(v): return v - 65536 if v > 32767 else v
|
|
o = self.offsets
|
|
ax = s((d[0] << 8) | d[1]) / 16384.0 - o[0]
|
|
ay = s((d[2] << 8) | d[3]) / 16384.0 - o[1]
|
|
az = s((d[4] << 8) | d[5]) / 16384.0 - o[2]
|
|
gx = s((d[8] << 8) | d[9]) / 131.0 - o[3]
|
|
gy = s((d[10] << 8) | d[11]) / 131.0 - o[4]
|
|
gz = s((d[12] << 8) | d[13]) / 131.0 - o[5]
|
|
# subtract gravity so idle magnitude ≈ 0
|
|
mag = math.sqrt(ax**2 + ay**2 + (az - 1.0)**2)
|
|
with self._lock:
|
|
if self._peak is None or mag > self._peak['magnitude']:
|
|
self._peak = {
|
|
'ax': round(ax, 3), 'ay': round(ay, 3), 'az': round(az, 3),
|
|
'gx': round(gx, 3), 'gy': round(gy, 3), 'gz': round(gz, 3),
|
|
'magnitude': round(mag, 4),
|
|
}
|
|
time.sleep(0.01)
|
|
|
|
def read(self):
|
|
with self._lock:
|
|
data = self._peak
|
|
self._peak = None # reset after each display read
|
|
return data
|
|
|
|
# ── QMC5883L Compass (0x1E) ──────────────────────────────────────────────────
|
|
|
|
def read_compass():
|
|
addr = 0x1E
|
|
try_read(lambda: bus.write_byte_data(addr, 0x09, 0x1D))
|
|
data = try_read(lambda: bus.read_i2c_block_data(addr, 0x00, 6))
|
|
if not data:
|
|
return None
|
|
def s(v): return v - 65536 if v > 32767 else v
|
|
x = s((data[1] << 8) | data[0])
|
|
y = s((data[3] << 8) | data[2])
|
|
z = s((data[5] << 8) | data[4])
|
|
heading = math.degrees(math.atan2(y, x))
|
|
if heading < 0:
|
|
heading += 360
|
|
return {'x': x, 'y': y, 'z': z, 'heading': round(heading, 1)}
|
|
|
|
# ── LTR390 (0x53) ────────────────────────────────────────────────────────────
|
|
|
|
UV_LABELS = [(0,'Low'),(3,'Moderate'),(6,'High'),(8,'Very High'),(11,'Extreme')]
|
|
|
|
def uv_label(uvi):
|
|
label = 'Low'
|
|
for threshold, name in UV_LABELS:
|
|
if uvi >= threshold:
|
|
label = name
|
|
return label
|
|
|
|
class LTR390:
|
|
def __init__(self):
|
|
self.sensor = try_read(lambda: adafruit_ltr390.LTR390(i2c))
|
|
|
|
def read(self):
|
|
if not self.sensor:
|
|
return None
|
|
return try_read(lambda: {
|
|
'lux': round(self.sensor.lux, 2),
|
|
'uvi': round(self.sensor.uvi, 2),
|
|
'uv_label': uv_label(self.sensor.uvi),
|
|
'solar_wm2': round(self.sensor.lux / 120, 2),
|
|
})
|
|
|
|
# ── TF-Luna (0x10) ───────────────────────────────────────────────────────────
|
|
|
|
def read_tfluna():
|
|
data = try_read(lambda: bus.read_i2c_block_data(0x10, 0x00, 6))
|
|
if not data:
|
|
return None
|
|
dist = data[0] | (data[1] << 8)
|
|
strength = data[2] | (data[3] << 8)
|
|
if strength < 100 or dist <= 0 or dist > 800:
|
|
return None
|
|
return {'distance_cm': dist, 'strength': strength}
|
|
|
|
# ── AS3935 (0x03) ────────────────────────────────────────────────────────────
|
|
|
|
AS3935_ADDR = 0x03
|
|
AS3935_NOISE_FLOOR = 7 # 0-7, raise if false positives persist
|
|
AS3935_WATCHDOG = 3 # 0-15
|
|
AS3935_SPIKE_REJ = 15 # 0-15, filters EMI from TF-Luna
|
|
|
|
def init_as3935():
|
|
val = (AS3935_NOISE_FLOOR << 4) | AS3935_WATCHDOG
|
|
try_read(lambda: bus.write_byte_data(AS3935_ADDR, 0x01, val))
|
|
try_read(lambda: bus.write_byte_data(AS3935_ADDR, 0x02, AS3935_SPIKE_REJ))
|
|
try_read(lambda: bus.write_byte_data(AS3935_ADDR, 0x00, 0x24)) # indoor
|
|
|
|
def read_as3935():
|
|
data = try_read(lambda: bus.read_i2c_block_data(AS3935_ADDR, 0x00, 9))
|
|
if not data:
|
|
return None
|
|
interrupt = data[3] & 0x0F
|
|
energy = ((data[5] & 0x1F) << 16) | (data[4] << 8) | data[3]
|
|
try_read(lambda: bus.read_byte_data(AS3935_ADDR, 0x03)) # clear interrupt
|
|
if interrupt not in (0x04, 0x08) or energy < 1000:
|
|
return {'noise_floor': AS3935_NOISE_FLOOR, 'interrupt': 0, 'distance_km': 0, 'energy': 0}
|
|
return {
|
|
'noise_floor': AS3935_NOISE_FLOOR,
|
|
'interrupt': interrupt,
|
|
'distance_km': data[6] & 0x3F,
|
|
'energy': energy,
|
|
}
|
|
|
|
# ── GPS (ttyS0) ──────────────────────────────────────────────────────────────
|
|
|
|
gps_cache = {}
|
|
|
|
def read_gps():
|
|
try:
|
|
ser = serial.Serial('/dev/ttyS0', 9600, timeout=1)
|
|
line = ser.readline().decode('ascii', errors='replace').strip()
|
|
ser.close()
|
|
if 'GGA' in line:
|
|
msg = pynmea2.parse(line)
|
|
gps_cache.update({
|
|
'lat': msg.latitude,
|
|
'lon': msg.longitude,
|
|
'alt': msg.altitude,
|
|
'satellites': msg.num_sats,
|
|
})
|
|
elif 'VTG' in line:
|
|
msg = pynmea2.parse(line)
|
|
gps_cache['speed_kmh'] = msg.spd_over_grnd_kmph
|
|
elif 'RMC' in line:
|
|
msg = pynmea2.parse(line)
|
|
gps_cache['heading'] = msg.true_course
|
|
except:
|
|
pass
|
|
return gps_cache or None
|
|
|
|
# ── Display ──────────────────────────────────────────────────────────────────
|
|
|
|
def fmt(val, unit='', unavailable='—'):
|
|
return f"{val}{unit}" if val is not None else unavailable
|
|
|
|
def section(title):
|
|
print(f"\n {title}")
|
|
print(f" {'─' * 40}")
|
|
|
|
def print_dashboard(bme_data, mpu_data):
|
|
compass_data = read_compass()
|
|
ltr_data = ltr.read()
|
|
luna_data = read_tfluna()
|
|
as_data = read_as3935()
|
|
gps_data = read_gps()
|
|
|
|
print('\033[2J\033[H', end='')
|
|
print(f"╔{'═'*50}╗")
|
|
print(f"║ 🌦 Weather Station — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ║")
|
|
print(f"╚{'═'*50}╝")
|
|
|
|
section("🧭 BN-880 — GPS / Compass")
|
|
if gps_data:
|
|
print(f" Heading: {compass_data['heading']}°")
|
|
print(f" Lat/Lon: {fmt(gps_data.get('lat'))}°, {fmt(gps_data.get('lon'))}°")
|
|
print(f" Altitude: {fmt(gps_data.get('alt'), 'm')}")
|
|
print(f" Speed: {fmt(gps_data.get('speed_kmh'), ' km/h')}")
|
|
print(f" Satellites: {fmt(gps_data.get('satellites'))}")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
section("🌡 BME680 — Temp / Humidity / Pressure / AQ")
|
|
if bme_data:
|
|
print(f" Temperature: {bme_data['temp_c']}°C ({bme_data['temp_f']}°F)")
|
|
print(f" Humidity: {bme_data['humidity']}%")
|
|
print(f" Dew Point: {bme_data['dew_point']}°C")
|
|
print(f" Pressure: {bme_data['pressure']} hPa")
|
|
print(f" Gas Res: {fmt(bme_data['gas_ohms'], ' Ω')}")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
section("☀️ LTR390 — Light / UV")
|
|
if ltr_data:
|
|
print(f" Lux: {ltr_data['lux']}")
|
|
print(f" UV Index: {ltr_data['uvi']} — {ltr_data['uv_label']}")
|
|
print(f" Solar W/m²: {ltr_data['solar_wm2']}")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
section("⚡ AS3935 — Lightning")
|
|
if as_data:
|
|
if as_data['interrupt'] == 0:
|
|
print(f" No event (noise floor: {as_data['noise_floor']}/7)")
|
|
else:
|
|
print(f" Distance: {as_data['distance_km']} km")
|
|
print(f" Energy: {as_data['energy']}")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
section("📐 MPU6050 — Seismic / IMU")
|
|
if mpu_data:
|
|
print(f" Magnitude: {mpu_data['magnitude']}g")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
section("📏 TF-Luna — Distance")
|
|
if luna_data:
|
|
print(f" Distance: {luna_data['distance_cm']} cm")
|
|
print(f" Strength: {luna_data['strength']}")
|
|
else:
|
|
print(" ❌ Unavailable")
|
|
|
|
print(f"\n{'═'*52}\n")
|
|
|
|
if __name__ == '__main__':
|
|
bme = BME680()
|
|
mpu = MPU6050()
|
|
ltr = LTR390()
|
|
init_as3935()
|
|
bme_data = None
|
|
mpu_data = None
|
|
while True:
|
|
bme_data = bme.read() or bme_data
|
|
mpu_data = mpu.read()
|
|
print_dashboard(bme_data, mpu_data)
|
|
time.sleep(1)
|