import sys import bme680 import uvicorn from fastapi import FastAPI import RPi.GPIO as GPIO import time import threading import pymysql from fastapi.middleware.cors import CORSMiddleware db_user = 'sensor_user' db_password = 'dein_passwort' use_bme680 = False use_proximity = False use_camera = True # --- Database connection setup --- try: # Connect to local MySQL database connection = pymysql.connect( host='localhost', port=3306, user=db_user, password=db_password, database='sensor_db', ) cursor = connection.cursor() # Create table if it doesn't already exist cursor.execute(''' CREATE TABLE IF NOT EXISTS sensor_data ( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, timestamp TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, motion BOOLEAN NOT NULL, temperature FLOAT NOT NULL, humidity FLOAT NOT NULL, pressure FLOAT NOT NULL, gas_resistance FLOAT NOT NULL ) ''') except pymysql.Error as error: print(error) sys.exit(1) # --- GPIO setup for motion sensor (RCWL-0516) --- sensor_pin = 22 GPIO.setmode(GPIO.BCM) GPIO.setup(sensor_pin, GPIO.IN) # --- Global shared state variables --- state = 0 motion_detected = False # Shared state temperature = 0 humidity = 0 pressure = 0 gas_resistance = 0 # ------------------------------------------------------------- # @function proxmity_loop # @description Monitors the motion sensor continuously and updates the shared motion state. # @returns None # ------------------------------------------------------------- def proxmity_loop(): global state, motion_detected try: while True: val = GPIO.input(sensor_pin) if val == 1: if state == 0: print("Motion detected!") motion_detected = True state = 1 else: if state == 1: print("Motion stopped!") motion_detected = False state = 0 time.sleep(0.1) except Exception as e: print(f"Sensor error: {e}") finally: GPIO.cleanup() # ------------------------------------------------------------- # @function sensor_loop # @description Reads data from the BME680 sensor (temperature, humidity, pressure, gas resistance) # and updates global variables in regular intervals. # @returns None # ------------------------------------------------------------- def sensor_loop(): global temperature, humidity, pressure, gas_resistance bme680sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY) bme680sensor.set_humidity_oversample(bme680.OS_2X) bme680sensor.set_pressure_oversample(bme680.OS_4X) bme680sensor.set_temperature_oversample(bme680.OS_8X) bme680sensor.set_filter(bme680.FILTER_SIZE_3) bme680sensor.set_gas_status(bme680.ENABLE_GAS_MEAS) bme680sensor.set_gas_heater_temperature(320) bme680sensor.set_gas_heater_duration(150) bme680sensor.select_gas_heater_profile(0) while True: if bme680sensor.get_sensor_data(): temperature = bme680sensor.data.temperature pressure = bme680sensor.data.pressure humidity = bme680sensor.data.humidity gas_resistance = bme680sensor.data.gas_resistance time.sleep(1) # ------------------------------------------------------------- # @function lineare_regression # @description Performs a simple linear regression between x and y values. # @param {list[float]} x - The independent variable values. # @param {list[float]} y - The dependent variable values. # @param {list[float]} [neue_x] - New x-values for prediction (optional). # @returns {list[float]|None} Returns the predicted y-values or None. # ------------------------------------------------------------- def lineare_regression(x, y, neue_x=None): n = len(x) x_mean = sum(x) / n y_mean = sum(y) / n sum_xy_abweichung = 0 sum_x_quadrat_abweichung = 0 for i in range(n): x_abweichung = x[i] - x_mean y_abweichung = y[i] - y_mean sum_xy_abweichung += x_abweichung * y_abweichung sum_x_quadrat_abweichung += x_abweichung * x_abweichung m = sum_xy_abweichung / sum_x_quadrat_abweichung if sum_x_quadrat_abweichung != 0 else 0 b = y_mean - m * x_mean vorhersagen = None if neue_x: vorhersagen = [round(m * xi + b, 4) for xi in neue_x] return vorhersagen # ------------------------------------------------------------- # @function database_loop # @description Periodically stores the latest sensor readings into the MySQL database. # @returns None # ------------------------------------------------------------- def database_loop(): global temperature, humidity, pressure, gas_resistance, motion_detected time.sleep(10) while True: cursor.execute('INSERT INTO sensor_data(motion, temperature, humidity, pressure, gas_resistance) VALUES (%s, %s, %s, %s, %s) ', ( motion_detected, temperature, humidity, pressure, gas_resistance )) connection.commit() time.sleep(300) # ------------------------------------------------------------- # @function startup_event # @description Initializes background threads for the sensor loops on application startup. # @returns None # ------------------------------------------------------------- app = FastAPI() @app.on_event("startup") def startup_event(): if use_bme680: sensor_thread = threading.Thread(target=sensor_loop, daemon=True) sensor_thread.start() if use_proximity: proxmity_thread = threading.Thread(target=proxmity_loop, daemon=True) proxmity_thread.start() if use_bme680: database_thread = threading.Thread(target=database_loop, daemon=True) database_thread.start() print("Sensor thread started") # --- CORS configuration (allows cross-origin requests) --- app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ------------------------------------------------------------- # @endpoint GET / # @description Basic test endpoint for server health check. # @returns {object} Returns a simple hello message. # ------------------------------------------------------------- @app.get("/") async def root(): return {"message": "Hello World"} # ------------------------------------------------------------- # @endpoint GET /sensor # @description Returns the latest live sensor readings from global variables. # @returns {object} JSON object containing motion, temperature, humidity, pressure, and gas resistance. # ------------------------------------------------------------- @app.get("/sensor") async def sensor(): return { "motion": motion_detected, "temperature": temperature, "humidity": humidity, "pressure": pressure, "gas_resistance": gas_resistance } # ------------------------------------------------------------- # @endpoint GET /get/all # @description Retrieves all stored sensor data from the database. # @returns {list[object]} List of all sensor data records. # ------------------------------------------------------------- @app.get("/get/all") async def get_all(): cursor.execute("SELECT * FROM sensor_data") sensor_data = cursor.fetchall() better_data = [] for row in sensor_data: better_data.append({ "id": row[0], "timestamp": row[1], "motion": row[2], "temperature": row[3], "humidity": row[4], "pressure": row[5], "gas_resistance": row[6] }) return better_data # ------------------------------------------------------------- # @endpoint GET /get/regression # @description Performs linear regression on predefined sample data (demo endpoint). # @returns {list[float]} Predicted y-values for the sample regression. # ------------------------------------------------------------- @app.get("/get/regression") async def getRegression(): countGuest = [13, 11, 16, 21, 14, 52, 27, 18] temps = [19.1, 18.0, 17.0, 16.1, 15.1, 23.1, 21.1, 19.9] return lineare_regression(countGuest, temps, [0, 60, 120]) # ------------------------------------------------------------- # @entrypoint # @description Starts the FastAPI app server using Uvicorn. # ------------------------------------------------------------- if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)