justin/AsiaBackend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

switched to gitea

Browse Source
This commit is contained in:
Justin Eckenweber
2026-05-20 10:29:00 +02:00
commit dc442f29bc
6 changed files with 599 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+268
View File
@@ -0,0 +1,268 @@
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'
# --- 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():
sensor_thread = threading.Thread(target=sensor_loop, daemon=True)
sensor_thread.start()
proxmity_thread = threading.Thread(target=proxmity_loop, daemon=True)
proxmity_thread.start()
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)