213 lines
6.9 KiB
Python
213 lines
6.9 KiB
Python
#!/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://<raspberry-pi-ip>: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)
|