#!/usr/bin/env python3 """ Web Dashboard for Lab Weather Tracker Provides a simple web interface to view current and historical weather data """ from flask import Flask, render_template, jsonify, request import csv import os from datetime import datetime, timedelta from collections import deque from smbus2 import SMBus try: from bme280 import BME280 except ImportError: from pimoroni_bme280 import BME280 try: from ltr559 import LTR559 ltr559 = LTR559() except ImportError: import ltr559 # Initialize sensors SENSORS_AVAILABLE = False try: bus = SMBus(1) bme280_sensor = BME280(i2c_dev=bus) SENSORS_AVAILABLE = True except (ImportError, RuntimeError, OSError) as e: SENSORS_AVAILABLE = False print(f"Warning: Sensor libraries not available: {e}") print("Using mock data for testing.") app = Flask(__name__) DATA_DIR = "data" MAX_HISTORY_POINTS = 100 # Number of historical points to display TEMP_COMP_FACTOR = 2.25 # Tuning factor for temperature compensation # CPU temperature tracking for compensation cpu_temps = [] def get_cpu_temperature(): """Get the temperature of the CPU for compensation""" try: with open("/sys/class/thermal/thermal_zone0/temp", "r") as f: temp = f.read() temp = int(temp) / 1000.0 return temp except Exception as e: # CPU temp not available (e.g., on non-Linux systems) return None def read_current_sensors(): """Read current sensor data""" global cpu_temps if SENSORS_AVAILABLE: try: # Read temperature, pressure, humidity from BME280 raw_temp = bme280_sensor.get_temperature() pressure = bme280_sensor.get_pressure() humidity = bme280_sensor.get_humidity() # Get CPU temperature and calculate compensated temperature cpu_temp = get_cpu_temperature() comp_temp = raw_temp # Default to raw if compensation fails if cpu_temp is not None: # Smooth out with some averaging to decrease jitter cpu_temps.append(cpu_temp) if len(cpu_temps) > 5: cpu_temps = cpu_temps[-5:] # Keep only last 5 readings if len(cpu_temps) > 0: avg_cpu_temp = sum(cpu_temps) / float(len(cpu_temps)) # Compensate temperature using CPU temp comp_temp = raw_temp - ((avg_cpu_temp - raw_temp) / TEMP_COMP_FACTOR) # Read light data from LTR559 lux = ltr559.get_lux() proximity = ltr559.get_proximity() return { 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'temperature_raw': round(raw_temp, 2), 'temperature': round(comp_temp, 2), 'pressure': round(pressure, 2), 'humidity': round(humidity, 2), 'light': round(lux, 2), 'proximity': round(proximity, 2) } except (OSError, IOError, RuntimeError) as e: print(f"Warning: I/O error reading sensors: {e}") print("Falling back to mock data. Check if sensors are properly connected.") return get_mock_data() except Exception as e: print(f"Unexpected error reading sensors: {e}") return get_mock_data() else: return get_mock_data() def get_mock_data(): """Generate mock data for testing when sensors aren't available""" import random raw_temp = 20 + random.uniform(-2, 2) comp_temp = raw_temp - 2 # Mock compensation return { 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'temperature_raw': round(raw_temp, 2), 'temperature': round(comp_temp, 2), 'pressure': round(1013 + random.uniform(-10, 10), 2), 'humidity': round(50 + random.uniform(-10, 10), 2), 'light': round(300 + random.uniform(-50, 50), 2), 'proximity': round(50 + random.uniform(-20, 20), 2) } def get_csv_files(): """Get list of CSV log files, most recent first""" if not os.path.exists(DATA_DIR): return [] files = [f for f in os.listdir(DATA_DIR) if f.startswith('weather_log_') and f.endswith('.csv')] files.sort(reverse=True) return files def read_historical_data(days=1): """Read historical data from CSV files""" data = [] files = get_csv_files()[:days] # Get most recent day(s) for filename in reversed(files): # Oldest first filepath = os.path.join(DATA_DIR, filename) try: with open(filepath, 'r') as csvfile: reader = csv.DictReader(csvfile) for row in reader: data.append({ 'timestamp': row['timestamp'], 'temperature': float(row['temperature_c']), 'pressure': float(row['pressure_hpa']), 'light': float(row['light_lux']) }) except Exception as e: print(f"Error reading {filename}: {e}") # Limit to most recent points if len(data) > MAX_HISTORY_POINTS: data = data[-MAX_HISTORY_POINTS:] return data @app.route('/') def index(): """Render main dashboard page""" return render_template('dashboard.html') @app.route('/api/current') def api_current(): """API endpoint for current sensor readings""" return jsonify(read_current_sensors()) @app.route('/api/history') def api_history(): """API endpoint for historical data""" days = int(request.args.get('days', 1)) return jsonify(read_historical_data(days)) @app.route('/api/stats') def api_stats(): """API endpoint for statistics""" data = read_historical_data(1) if not data: return jsonify({}) temps = [d['temperature'] for d in data] pressures = [d['pressure'] for d in data] lights = [d['light'] for d in data] return jsonify({ 'temperature': { 'min': round(min(temps), 2), 'max': round(max(temps), 2), 'avg': round(sum(temps) / len(temps), 2) }, 'pressure': { 'min': round(min(pressures), 2), 'max': round(max(pressures), 2), 'avg': round(sum(pressures) / len(pressures), 2) }, 'light': { 'min': round(min(lights), 2), 'max': round(max(lights), 2), 'avg': round(sum(lights) / len(lights), 2) } }) if __name__ == '__main__': # Create data directory if it doesn't exist if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) print("Starting Lab Weather Dashboard...") print("Access the dashboard at: http://localhost:5000") print("Or from another device: http://:5000") # Run on all interfaces so it's accessible from other devices # use_reloader=False to prevent input() errors in non-interactive environments app.run(host='0.0.0.0', port=5000, debug=False, use_reloader=False)