import io import sys from time import sleep 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 from picamera2 import Picamera2 from starlette.responses import StreamingResponse 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 ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS recordings ( id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, timestamp TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, file_path TEXT 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) # --- Camera Pins camera_pin = 17 GPIO.setmode(GPIO.BCM) GPIO.setup(camera_pin, GPIO.IN) # --- Global shared state variables --- state = 0 motion_detected = False # Shared state temperature = 0 humidity = 0 pressure = 0 gas_resistance = 0 is_recording = False picam2 = None def camera_loop(): global is_recording, picam2 try: while True and not is_recording: val = GPIO.input(camera_pin) if val == 1: is_recording = True print("Start recording!") timestamp = time.strftime("%Y%m%d_%H%M%S") filename = f"recordings/video_{timestamp}.mp4" picam2.start_and_record_video(filename, duration=10) cursor.execute( "INSERT INTO recordings (file_path) VALUES (%s)", filename ) connection.commit() print("Recorded video of 10 seconds, motion detected!!!!") is_recording = False print(val) except Exception as e: print(f"Camera error: {e}") finally: GPIO.cleanup() # ------------------------------------------------------------- # @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() def init_camera(): global picam2 try: camera = Picamera2() # capture_file(..., format="jpeg") kodiert die Frames selbst als JPEG. # Der Kamerastream bleibt deshalb ein normaler RGB-Stream. camera_config = camera.create_preview_configuration(main={"format": "RGB888", "size": (640, 480)}) camera.configure(camera_config) #camera.start() picam2 = camera #print("Kamera bereit für das Streaming!") except Exception as e: picam2 = None print(f"Kamera konnte nicht initialisiert werden: {e}") def capture_jpeg_frame(camera): stream = io.BytesIO() camera.capture_file(stream, format="jpeg") return stream.getvalue() # ------------------------------------------------------------- # @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() if use_camera: init_camera() camera_thread = threading.Thread(target=camera_loop, daemon=True) camera_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]) def gen_frames(): global picam2 while True: camera = picam2 if camera is not None: try: frame_bytes = capture_jpeg_frame(camera) # MJPEG-Boundary-Stream zusammensetzen yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n') except Exception as e: print(f"Fehler beim Frame-Grabbing: {e}") time.sleep(1) time.sleep(0.04) # ~25 FPS # Der neue FastAPI-Endpoint für dein Frontend @app.get("/video_feed") async def video_feed(): return StreamingResponse(gen_frames(), media_type="multipart/x-mixed-replace; boundary=frame") @app.get("/recordings") async def getRecordings(): cursor.execute("SELECT * FROM recordings") sensor_data = cursor.fetchall() formated_data = [] for row in sensor_data: formated_data.append({ "id": row[0], "timestamp": row[1], "file_path": row[2], }) return formated_data # ------------------------------------------------------------- # @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)