import time, math, board, smbus2, serial, psutil, pynmea2, bme680 import adafruit_ltr390, threading, os, statistics from datetime import datetime, timezone from collections import deque from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import SYNCHRONOUS from dotenv import load_dotenv from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent def cfg(): load_dotenv(BASE_DIR / '.env', override=True) load_dotenv(BASE_DIR / '.env.local', override=True) return { 'LATITUDE': float(os.getenv('LATITUDE', '0.0')), 'LONGITUDE': float(os.getenv('LONGITUDE', '0.0')), 'ALTITUDE': float(os.getenv('ALTITUDE', '0.0')), 'INFLUX_URL': os.getenv('INFLUX_URL', 'http://localhost:8086'), 'INFLUX_TOKEN': os.getenv('INFLUX_TOKEN', ''), 'INFLUX_ORG': os.getenv('INFLUX_ORG', 'weather'), 'INFLUX_BUCKET': os.getenv('INFLUX_BUCKET', 'station'), 'GPS_PORT': os.getenv('GPS_PORT', '/dev/ttyS0'), 'GPS_BAUD': int( os.getenv('GPS_BAUD', '9600')), 'GAS_REFERENCE': int( os.getenv('GAS_REFERENCE', '250000')), 'PRESSURE_TREND_WINDOW': int( os.getenv('PRESSURE_TREND_WINDOW', '60')), 'LUNA_CAL_SAMPLES': int( os.getenv('LUNA_CAL_SAMPLES', '20')), 'LUNA_MAX_DIST_CM': int( os.getenv('LUNA_MAX_DIST_CM', '800')), 'LUNA_MIN_DEPTH_CM': int( os.getenv('LUNA_MIN_DEPTH_CM', '2')), 'AS3935_ADDR': int( os.getenv('AS3935_ADDR', '0x03'), 16), 'AS3935_NOISE_FLOOR': int( os.getenv('AS3935_NOISE_FLOOR', '5')), 'AS3935_WATCHDOG': int( os.getenv('AS3935_WATCHDOG', '3')), 'AS3935_SPIKE_REJ': int( os.getenv('AS3935_SPIKE_REJ', '7')), 'SKIN_TYPE_2_MED': int( os.getenv('SKIN_TYPE_2_MED', '200')), 'LOOP_INTERVAL': float( os.getenv('LOOP_INTERVAL', '1.0')), 'FLOOD_TEMP_THRESHOLD': float( os.getenv('FLOOD_TEMP_THRESHOLD', '10.0')), 'SLUSH_TEMP_THRESHOLD': float( os.getenv('SLUSH_TEMP_THRESHOLD', '0.0')), 'SNOW_STRENGTH_MIN': int( os.getenv('SNOW_STRENGTH_MIN', '700')), 'SEISMIC_NOISE_FLOOR': float( os.getenv('SEISMIC_NOISE_FLOOR', '0.02')), 'TEMP_CALIBRATION': float( os.getenv('TEMP_CALIBRATION', '-7.5')), 'CLOUD_BUCKET_SIZE': int( os.getenv('CLOUD_BUCKET_SIZE', '5')), 'CLOUD_HISTORY': int( os.getenv('CLOUD_HISTORY', '20')), 'CLOUD_BASELINE_PCT': float( os.getenv('CLOUD_BASELINE_PCT', '0.9')), 'CLOUD_VOLATILITY_THRESH':float( os.getenv('CLOUD_VOLATILITY_THRESH', '0.3')), 'CLOUD_CLEAR_MAX': float( os.getenv('CLOUD_CLEAR_MAX', '15.0')), 'CLOUD_MOSTLY_CLEAR_MAX': float( os.getenv('CLOUD_MOSTLY_CLEAR_MAX', '40.0')), 'CLOUD_CLOUDY_MAX': float( os.getenv('CLOUD_CLOUDY_MAX', '80.0')), 'CLOUD_MIN_SAMPLES': int( os.getenv('CLOUD_MIN_SAMPLES', '5')), 'CLOUD_LUX_WINDOW': int( os.getenv('CLOUD_LUX_WINDOW', '10')), } # ── InfluxDB ────────────────────────────────────────────────────────────────── _influx_client = None _influx_writer = None _influx_url = None _influx_token = None def get_writer(c): global _influx_client, _influx_writer, _influx_url, _influx_token if c['INFLUX_URL'] != _influx_url or c['INFLUX_TOKEN'] != _influx_token: if _influx_client: _influx_client.close() _influx_client = InfluxDBClient(url=c['INFLUX_URL'], token=c['INFLUX_TOKEN'], org=c['INFLUX_ORG']) _influx_writer = _influx_client.write_api(write_options=SYNCHRONOUS) _influx_url, _influx_token = c['INFLUX_URL'], c['INFLUX_TOKEN'] return _influx_writer def write(c, measurement, fields, tags={}): try: p = Point(measurement).time(datetime.now(timezone.utc)) for k, v in tags.items(): p = p.tag(k, v) for k, v in fields.items(): if v is not None: p = p.field(k, float(v) if isinstance(v, (int, float)) else v) get_writer(c).write(bucket=c['INFLUX_BUCKET'], org=c['INFLUX_ORG'], record=p) except Exception as e: print(f"[InfluxDB] {measurement}: {e}") # ── Helpers ─────────────────────────────────────────────────────────────────── def try_init(fn, name): try: r = fn() print(f" ✅ {name}") return r except Exception as e: print(f" ❌ {name}: {e}") return None def try_read(fn): try: return fn() except: return None def s16(v): return v - 65536 if v > 32767 else v try: bus = smbus2.SMBus(1) except: bus = None; print("[I2C] SMBus init failed") try: i2c = board.I2C() except: i2c = None; print("[I2C] board.I2C init failed") # ── BME680 ──────────────────────────────────────────────────────────────────── pressure_buffer = deque(maxlen=60) def init_bme(): sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY) sensor.set_humidity_oversample(bme680.OS_2X) sensor.set_pressure_oversample(bme680.OS_4X) sensor.set_temperature_oversample(bme680.OS_8X) sensor.set_filter(bme680.FILTER_SIZE_3) sensor.set_gas_status(bme680.ENABLE_GAS_MEAS) sensor.set_gas_heater_temperature(320) sensor.set_gas_heater_duration(150) sensor.select_gas_heater_profile(0) return sensor def heat_index(tc, rh): tf = tc * 9/5 + 32 if tf < 80: return round(tc, 2) hi = (-42.379 + 2.04901523*tf + 10.14333127*rh - 0.22475541*tf*rh - 0.00683783*tf**2 - 0.05481717*rh**2 + 0.00122874*tf**2*rh + 0.00085282*tf*rh**2 - 0.00000199*tf**2*rh**2) return round((hi - 32) * 5/9, 2) def absolute_humidity(tc, rh): return round((6.112 * math.exp((17.67*tc)/(tc+243.5)) * rh * 2.1674) / (273.15+tc), 3) def vpd(tc, rh): es = 0.6108 * math.exp((17.27*tc)/(tc+237.3)) return round(es - (rh/100)*es, 3) def gas_to_aqi(gas_ohms, humidity, gas_reference): if gas_ohms is None: return None, None gas_score = min(gas_ohms / gas_reference, 1.0) * 75 hum_score = 25 - abs(humidity - 40) * 0.5 quality = min(max(gas_score + hum_score, 0), 100) aqi = round((1 - quality / 100) * 500) if aqi <= 50: label = 'Good' elif aqi <= 100: label = 'Standard' elif aqi <= 150: label = 'Warning' elif aqi <= 200: label = 'Unhealthy' elif aqi <= 300: label = 'Very Unhealthy' else: label = 'Hazardous' return aqi, label def sea_level_pressure(pressure_hpa, alt_m): if alt_m is None: return None return round(pressure_hpa / math.pow(1 - (alt_m / 44330.0), 5.255), 2) def frost_risk(temp_c, dew_point_c): margin = temp_c - dew_point_c if temp_c <= 0: return 'High' if temp_c <= 3 and margin < 2: return 'Moderate' if temp_c <= 5: return 'Low' return 'None' def get_pressure_trend(window): if pressure_buffer.maxlen != window: pressure_buffer.__init__(window) if len(pressure_buffer) < 2: return None, None delta = pressure_buffer[-1] - pressure_buffer[0] label = 'Rising' if delta > 1.0 else 'Falling' if delta < -1.0 else 'Stable' return round(delta, 2), label def read_bme(c, sensor, alt_m=None): null = { 'temperature': None, 'humidity': None, 'humidity_abs': None, 'dew_point': None, 'pressure_hpa': None, 'pressure_slp': None, 'pressure_rate': None, 'vapor_pressure_deficit': None, 'heat_index': None, 'air_quality_ohms': None, 'air_quality': None, } if not sensor: write(c, 'environment', null) return None, None try: if not sensor.get_sensor_data(): return None, None t, rh, p = sensor.data.temperature, sensor.data.humidity, sensor.data.pressure gas = round(sensor.data.gas_resistance, 0) if sensor.data.heat_stable else None aqi_score, aqi_label = gas_to_aqi(gas, rh, c['GAS_REFERENCE']) pressure_buffer.append(p) p_rate, p_label = get_pressure_trend(c['PRESSURE_TREND_WINDOW']) dp = round(t - (100-rh)/5.0, 2) write(c, 'environment', { 'temperature': round(t + c['TEMP_CALIBRATION'], 2), 'humidity': round(rh, 2), 'dew_point': dp, 'pressure_hpa': round(p, 2), 'pressure_slp': sea_level_pressure(p, alt_m), 'pressure_rate': p_rate, 'humidity_abs': absolute_humidity(t, rh), 'vapor_pressure_deficit': vpd(t, rh), 'heat_index': heat_index(t, rh), 'air_quality_ohms': gas, 'air_quality': aqi_score, }, tags={ 'air_quality_label': aqi_label or '', 'pressure_trend': p_label or '', 'frost_risk': frost_risk(t, dp), }) return t, rh except Exception as e: print(f"[BME680] {e}") return None, None # ── MPU6050 ─────────────────────────────────────────────────────────────────── MPU_ADDR = 0x68 _mpu_peak = None _mpu_lock = threading.Lock() _mpu_offsets = None def calibrate_mpu(samples=200): if not bus: return None print(" Calibrating MPU6050, keep still...") sums = [0.0]*6 n = 0 for _ in range(samples): d = try_read(lambda: bus.read_i2c_block_data(MPU_ADDR, 0x3B, 14)) if d: sums[0] += s16((d[0] << 8)|d[1]) / 16384.0 sums[1] += s16((d[2] << 8)|d[3]) / 16384.0 sums[2] += s16((d[4] << 8)|d[5]) / 16384.0 sums[3] += s16((d[8] << 8)|d[9]) / 131.0 sums[4] += s16((d[10] << 8)|d[11]) / 131.0 sums[5] += s16((d[12] << 8)|d[13]) / 131.0 n += 1 time.sleep(0.005) if n == 0: return None return [sums[0]/n, sums[1]/n, sums[2]/n-1.0, sums[3]/n, sums[4]/n, sums[5]/n] def mpu_loop(): global _mpu_peak while True: if bus and _mpu_offsets: d = try_read(lambda: bus.read_i2c_block_data(MPU_ADDR, 0x3B, 14)) if d: o = _mpu_offsets ax = s16((d[0] << 8)|d[1]) / 16384.0 - o[0] ay = s16((d[2] << 8)|d[3]) / 16384.0 - o[1] az = s16((d[4] << 8)|d[5]) / 16384.0 - o[2] mag = math.sqrt(ax**2 + ay**2 + (az - 1.0)**2) with _mpu_lock: if _mpu_peak is None or mag > _mpu_peak['magnitude']: _mpu_peak = { 'seismic_x': round(ax, 3), 'seismic_y': round(ay, 3), 'seismic_z': round(az, 3), 'magnitude': round(mag, 4), } time.sleep(0.01) def flush_mpu(c): global _mpu_peak with _mpu_lock: data, _mpu_peak = _mpu_peak, None if data and data['magnitude'] > c['SEISMIC_NOISE_FLOOR']: write(c, 'seismic', data) # ── QMC5883L ────────────────────────────────────────────────────────────────── def read_compass(c): null = {'x': None, 'y': None, 'z': None, 'heading': None} if not bus: write(c, 'compass', null); return try: addr = 0x1E try_read(lambda: bus.write_byte_data(addr, 0x09, 0x1D)) d = try_read(lambda: bus.read_i2c_block_data(addr, 0x00, 6)) if not d: write(c, 'compass', null); return x = s16((d[1] << 8)|d[0]) y = s16((d[3] << 8)|d[2]) z = s16((d[5] << 8)|d[4]) heading = math.degrees(math.atan2(y, x)) if heading < 0: heading += 360 write(c, 'compass', { 'compass_x': x, 'compass_y': y, 'compass_z': z, 'heading': round(heading, 1), }) except Exception as e: print(f"[QMC5883L] {e}") write(c, 'compass', null) # ── LTR390 ─────────────────────────────────────────────────────────────────── uv_dose_mj = 0.0 uv_dose_date = datetime.now().date() last_uv_time = time.time() dli_lux_acc = 0.0 dli_date = datetime.now().date() last_lux_time = time.time() daylight_start = None DAYLIGHT_LUX_THRESHOLD = 50 _cloud_baseline = {} _cloud_lux_window = deque(maxlen=10) def _el_bucket(solar_el, bucket_size): if solar_el is None: return None return round(solar_el / bucket_size) * bucket_size def moon_phase_factor(): known_new = datetime(2024, 1, 11, tzinfo=timezone.utc) lunar_cycle = 29.53058867 days = (datetime.now(timezone.utc) - known_new).total_seconds() / 86400 phase = (days % lunar_cycle) / lunar_cycle return round(math.cos(math.pi * (phase - 0.5)) * 0.5 + 0.5, 3) def update_cloud_baseline(c, solar_el, lux): bucket = _el_bucket(solar_el, c['CLOUD_BUCKET_SIZE']) if bucket is None: return if bucket not in _cloud_baseline: _cloud_baseline[bucket] = deque(maxlen=c['CLOUD_HISTORY']) _cloud_baseline[bucket].append(lux) def get_expected_lux(c, solar_el): bucket = _el_bucket(solar_el, c['CLOUD_BUCKET_SIZE']) if bucket is None: return None history = _cloud_baseline.get(bucket) if not history or len(history) < c['CLOUD_MIN_SAMPLES']: return None sorted_vals = sorted(history) idx = max(0, int(len(sorted_vals) * c['CLOUD_BASELINE_PCT']) - 1) baseline = sorted_vals[idx] if solar_el is not None and solar_el < 0: baseline = baseline * moon_phase_factor() return round(baseline, 2) def classify_clouds(c, lux, solar_el): # Resize window if config changed if _cloud_lux_window.maxlen != c['CLOUD_LUX_WINDOW']: _cloud_lux_window.__init__(maxlen=c['CLOUD_LUX_WINDOW']) _cloud_lux_window.append(lux) expected = get_expected_lux(c, solar_el) if expected is None or expected < 0.1: return None, None, None ratio = min(lux / expected, 1.0) cloud_pct = round((1 - ratio) * 100, 1) volatility = None if len(_cloud_lux_window) >= 3: mean = statistics.mean(_cloud_lux_window) if mean > 0: volatility = round(statistics.stdev(_cloud_lux_window) / mean, 3) vt = c['CLOUD_VOLATILITY_THRESH'] if volatility is not None and volatility > vt and c['CLOUD_CLEAR_MAX'] < cloud_pct < c['CLOUD_CLOUDY_MAX']: label = 'Partly Cloudy' elif cloud_pct < c['CLOUD_CLEAR_MAX']: label = 'Clear' elif cloud_pct < c['CLOUD_MOSTLY_CLEAR_MAX']: label = 'Mostly Clear' elif cloud_pct < c['CLOUD_CLOUDY_MAX']: label = 'Cloudy' else: label = 'Overcast' return cloud_pct, label, volatility def visibility_estimate(lux, humidity, solar_el=None): base_km = 50.0 hum_factor = max(0, 1 - ((humidity - 40) / 60)) if humidity > 40 else 1.0 if solar_el is not None and solar_el > 5: lux_factor = min(lux / 10000, 1.0) else: lux_factor = 1.0 return round(base_km * hum_factor * lux_factor, 1) def uv_index_label(uvi): if uvi < 3: return 'Low' elif uvi < 6: return 'Moderate' elif uvi < 8: return 'High' elif uvi < 11: return 'Very High' else: return 'Extreme' def read_ltr(c, sensor, solar_el, humidity=None): global uv_dose_mj, uv_dose_date, last_uv_time global dli_lux_acc, dli_date, last_lux_time, daylight_start null = { 'lux': None, 'uv_index': None, 'solar_wm2': None, 'uv_dose': None, 'clouds': None, 'daily_light_integral': None, 'daylight': None, 'visibility': None, } if not sensor: write(c, 'light', null); return try: lux = round(sensor.lux, 2) uvi = round(sensor.uvi, 2) except Exception as e: print(f"[LTR390] {e}") write(c, 'light', null) return now = time.time() today = datetime.now().date() if today != uv_dose_date: uv_dose_mj, uv_dose_date = 0.0, today if today != dli_date: dli_lux_acc, dli_date, daylight_start = 0.0, today, None dt = now - last_uv_time uv_dose_mj += uvi * 25 * dt dli_lux_acc += lux * (now - last_lux_time) last_uv_time = now last_lux_time = now dli = round(dli_lux_acc / 54 / 1_000_000, 4) if lux >= DAYLIGHT_LUX_THRESHOLD and daylight_start is None: daylight_start = now daylight_hours = round((now - daylight_start) / 3600, 2) if daylight_start else 0.0 update_cloud_baseline(c, solar_el, lux) cloud_pct, cloud_label, volatility = classify_clouds(c, lux, solar_el) write(c, 'light', { 'lux': lux, 'uv_index': uvi, 'solar_wm2': round(lux / 120, 2), 'uv_dose': round(uv_dose_mj, 2), 'clouds': cloud_pct, 'daily_light_integral': dli, 'daylight': daylight_hours, 'visibility': visibility_estimate(lux, humidity, solar_el) if humidity else None, }, tags={ 'clouds_label': cloud_label or '' 'uv_index_label': uv_index_label(uvi), }) # ── TF-Luna ─────────────────────────────────────────────────────────────────── luna_baseline_cm = None def calibrate_luna(c): global luna_baseline_cm if not bus: print(" ❌ TF-Luna: no I2C bus"); return print(" Calibrating TF-Luna...") readings = [] for _ in range(c['LUNA_CAL_SAMPLES']): d = try_read(lambda: bus.read_i2c_block_data(0x10, 0x00, 6)) if d: dist, strength = d[0]|(d[1]<<8), d[2]|(d[3]<<8) if strength >= 100 and 0 < dist <= c['LUNA_MAX_DIST_CM']: readings.append(dist) time.sleep(0.1) if readings: luna_baseline_cm = round(sum(readings)/len(readings), 1) print(f" ✅ TF-Luna baseline: {luna_baseline_cm} cm") else: print(" ❌ TF-Luna: calibration failed") def read_luna(c, temp_c): null = {'ground_distance': None, 'lidar_strength': None} if not bus: write(c, 'accumulation', null); return try: d = try_read(lambda: bus.read_i2c_block_data(0x10, 0x00, 6)) if not d: write(c, 'ground', null); return dist = d[0]|(d[1]<<8) strength = d[2]|(d[3]<<8) if strength < 100 or dist <= 0 or dist > c['LUNA_MAX_DIST_CM']: write(c, 'accumulation', null); return fields = {'ground_distance': dist, 'ground_calibration': luna_baseline_cm, 'lidar_strength': strength} kind = 'none' if luna_baseline_cm and (luna_baseline_cm - dist) >= c['LUNA_MIN_DEPTH_CM']: t = temp_c depth = round(luna_baseline_cm - dist, 1) kind = ( 'flood' if t is not None and t > c['FLOOD_TEMP_THRESHOLD'] else 'slush' if t is not None and t > c['SLUSH_TEMP_THRESHOLD'] else 'snow' if strength > c['SNOW_STRENGTH_MIN'] else 'ice' ) fields['accumulation'] = depth else: fields['accumulation'] = 0 write(c, 'accumulation', fields, tags={'accumulation_type': kind}) except Exception as e: print(f"[TF-Luna] {e}") write(c, 'accumulation', null) # ── AS3935 ─────────────────────────────────────────────────────────────────── lightning_distances = deque(maxlen=10) lightning_count = 0 lightning_window_start = time.time() def init_as3935(c): if not bus: return try: addr = c['AS3935_ADDR'] val = (c['AS3935_NOISE_FLOOR'] << 4) | c['AS3935_WATCHDOG'] bus.write_byte_data(addr, 0x01, val) bus.write_byte_data(addr, 0x02, c['AS3935_SPIKE_REJ']) bus.write_byte_data(addr, 0x00, 0x24) print(" ✅ AS3935") except Exception as e: print(f" ❌ AS3935: {e}") def read_as3935(c): global lightning_count, lightning_window_start if not bus: return try: addr = c['AS3935_ADDR'] data = try_read(lambda: bus.read_i2c_block_data(addr, 0x00, 9)) if not data: return interrupt = data[3] & 0x0F energy = ((data[5] & 0x1F) << 16) | (data[4] << 8) | data[3] try_read(lambda: bus.read_byte_data(addr, 0x03)) if interrupt not in (0x04, 0x08) or energy < 1000: return dist = data[6] & 0x3F lightning_distances.append(dist) lightning_count += 1 elapsed = time.time() - lightning_window_start if elapsed >= 3600: lightning_count = 1 lightning_window_start = time.time() elapsed = 1 strike_rate = round(lightning_count / (elapsed / 3600), 1) trend = 'Stationary' if len(lightning_distances) >= 3: avg_early = sum(list(lightning_distances)[:3]) / 3 avg_late = sum(list(lightning_distances)[-3:]) / 3 if avg_late < avg_early - 2: trend = 'Approaching' elif avg_late > avg_early + 2: trend = 'Retreating' write(c, 'lightning', { 'lightning_distance': dist, 'lightning_energy': energy, 'lightning_rate': strike_rate, }, tags={'storm_direction': trend}) except Exception as e: print(f"[AS3935] {e}") # ── GPS ─────────────────────────────────────────────────────────────────────── gps_cache = {} def read_gps(c): try: ser = serial.Serial(c['GPS_PORT'], c['GPS_BAUD'], timeout=1) line = ser.readline().decode('ascii', errors='replace').strip() ser.close() if 'GGA' in line: msg = pynmea2.parse(line) gps_cache.update({ 'latitude': msg.latitude, 'longitude': msg.longitude, 'altitude': msg.altitude, 'gps_satellites': int(msg.num_sats), }) if gps_cache: write(c, 'gps', {k: v for k, v in gps_cache.items() if v is not None}) except Exception as e: print(f"[GPS] {e}") write(c, 'gps', {'latitude': None, 'longitude': None, 'altitude': None, 'gps_satellites': None}) return gps_cache def solar_elevation(lat, lon): now = datetime.now(timezone.utc) day = now.timetuple().tm_yday decl = math.radians(23.45 * math.sin(math.radians((360/365)*(day-81)))) hour_angle = math.radians(((now.hour + now.minute/60) - 12) * 15 + lon) lat_r = math.radians(lat) el = math.degrees(math.asin( math.sin(lat_r)*math.sin(decl) + math.cos(lat_r)*math.cos(decl)*math.cos(hour_angle) )) return round(el, 2) # ── Rain ────────────────────────────────────────────────────────────────────── def read_rain(c): try: write(c, 'rain', {'precipitation': 0}, tags={'raining': False}) except Exception as e: print(f"[Rain] {e}") # ── Wind ────────────────────────────────────────────────────────────────────── def read_wind(c): try: write(c, 'wind', {'wind_direction': 0, 'wind_speed': 0, 'wind_gusts': 0}) except Exception as e: print(f"[Wind] {e}") # ── System ──────────────────────────────────────────────────────────────────── def read_system(c): try: cpu_temp = None temps = psutil.sensors_temperatures() for key in ('cpu_thermal', 'coretemp', 'k10temp', 'acpitz'): if key in temps: cpu_temp = round(temps[key][0].current, 1) break cpu_freq = psutil.cpu_freq() vm = psutil.virtual_memory() disk = psutil.disk_usage('/') write(c, 'system', { 'cpu_temp': cpu_temp, 'cpu_usage': round(psutil.cpu_percent(interval=None), 1), 'cpu_freq_mhz': round(cpu_freq.current, 1) if cpu_freq else None, 'mem_used_mb': round(vm.used / 1024**2, 1), 'mem_total_mb': round(vm.total / 1024**2, 1), 'mem_percent': vm.percent, 'disk_used_gb': round(disk.used / 1024**3, 2), 'disk_total_gb': round(disk.total / 1024**3, 2), 'disk_percent': disk.percent, }) except Exception as e: print(f"[System] {e}") # ── Main ────────────────────────────────────────────────────────────────────── if __name__ == '__main__': print("\n🌦 Weather Station — Initialising...\n") c = cfg() bme_sensor = try_init(init_bme, 'BME680') ltr_sensor = try_init(lambda: adafruit_ltr390.LTR390(i2c) if i2c else (_ for _ in ()).throw(Exception('no i2c')), 'LTR390') if bus: try_read(lambda: bus.write_byte_data(MPU_ADDR, 0x6B, 0)) time.sleep(0.1) _mpu_offsets = try_init(calibrate_mpu, 'MPU6050') else: _mpu_offsets = None print(" ❌ MPU6050: no I2C bus") threading.Thread(target=mpu_loop, daemon=True).start() init_as3935(c) calibrate_luna(c) print("\n🚀 Starting sensor loop...\n") last_temp = None last_rh = None while True: c = cfg() gps = read_gps(c) lat = gps.get('latitude') or c['LATITUDE'] lon = gps.get('longitude') or c['LONGITUDE'] alt = gps.get('altitude') or c['ALTITUDE'] sol = solar_elevation(lat, lon) if lat and lon else None result = read_bme(c, bme_sensor, alt_m=alt) if result and result[0] is not None: last_temp, last_rh = result flush_mpu(c) read_compass(c) read_ltr(c, ltr_sensor, sol, humidity=last_rh) read_luna(c, last_temp) read_as3935(c) read_rain(c) read_wind(c) read_system(c) time.sleep(c['LOOP_INTERVAL'])