from flask import Flask, Response, request, render_template, jsonify, g import paho.mqtt.client as mqtt from ultralytics import YOLO import cv2 import numpy as np import threading import mysql.connector import json app = Flask(__name__) # Load a model model = YOLO("yolo11n.pt") # pretrained YOLO11n model kobuki_message = "" latest_image = None processed_image = None yolo_results = [] # Lock for thread-safe access to shared variables lock = threading.Lock() # List of class names (example for COCO dataset) yolo_classes = list(model.names.values()) def on_message(client, userdata, message): global kobuki_message, latest_image, processed_image, yolo_results if message.topic == "kobuki/data": kobuki_message = str(message.payload.decode("utf-8")) elif message.topic == "kobuki/cam": with lock: # Lock the shared variables between threads so they can't be accessed at the same time and you cant have half processed images latest_image = np.frombuffer(message.payload, np.uint8) latest_image = cv2.imdecode(latest_image, cv2.IMREAD_COLOR) # Process the image with YOLO results = model(latest_image) yolo_results = [] processed_image = latest_image.copy() # Create a copy for processing for result in results: for box in result.boxes: class_id = int(box.cls.item()) class_name = yolo_classes[class_id] yolo_results.append({ "class": class_name, "confidence": box.conf.item(), "bbox": box.xyxy.tolist() }) # Draw bounding box on the processed image x1, y1, x2, y2 = map(int, box.xyxy[0]) cv2.rectangle(processed_image, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText(processed_image, f"{class_name} {box.conf.item():.2f}", (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2) # Create an MQTT client instance mqtt_client = mqtt.Client() mqtt_client.username_pw_set("server", "serverwachtwoordofzo") mqtt_client.connect("localhost", 1884, 60) mqtt_client.loop_start() mqtt_client.subscribe("kobuki/data") mqtt_client.subscribe("kobuki/cam") mqtt_client.on_message = on_message # this line needs to be under the function definition otherwise it can't find which function it needs to use # Database connectie-functie def get_db(): if 'db' not in g: # 'g' is specifiek voor een request en leeft zolang een request duurt g.db = mysql.connector.connect( host="127.0.0.1", port=3306, user="admin", password="kobuki", database="kobuki" ) return g.db # Sluit de database na elke request @app.teardown_appcontext def close_db(error): db = g.pop('db', None) if db is not None: db.close() @app.route('/') def index(): return render_template('index.html') @app.route('/control', methods=["GET", "POST"]) def control(): if request.authorization and request.authorization.username == 'ishak' and request.authorization.password == 'kobuki': return render_template('control.html') else: return ('Unauthorized', 401, {'WWW-Authenticate': 'Basic realm="Login Required"'}) @app.route('/move', methods=['POST']) def move(): data = request.get_json() direction = data.get("direction") # Verstuur de richting via MQTT if direction: mqtt_client.publish("home/commands", direction) db_connection = get_db() cursor = db_connection.cursor() sql_command = "INSERT INTO command (command) VALUES (%s)" cursor.execute(sql_command, (direction,)) db_connection.commit() cursor.close() db_connection.close() return jsonify({"status": "success", "direction": direction}) @app.route('/data', methods=['GET']) def data(): try: # Parse de JSON-string naar een Python-dictionary data = json.loads(kobuki_message) # Maak een lijst van tuples met de naam en waarde van elk veld sensor_data_tuples = [(name, float(value)) for name, value in data.items() if isinstance(value, (int, float))] # Extra informatie of nested data (zoals "extraInfo" of "gyroData") kun je apart verwerken if "extraInfo" in data: for key, value in data["extraInfo"].items(): sensor_data_tuples.append((f"extraInfo_{key}", float(value))) if "gyroData" in data: for i, gyro in enumerate(data["gyroData"]): for axis, value in gyro.items(): sensor_data_tuples.append((f"gyroData_{i}_{axis}", float(value))) # Database-insert db = get_db() with db.cursor() as cursor: # Zorg dat je tabel `kobuki_data` kolommen heeft: `name` en `value` sql_sensor = "INSERT INTO kobuki_data (name, value) VALUES (%s, %s)" cursor.executemany(sql_sensor, sensor_data_tuples) # Commit en sluit de cursor db.commit() cursor.close() except json.JSONDecodeError as e: print(f"JSON decode error: {e}") except mysql.connector.Error as err: print(f"Database error: {err}") return kobuki_message @app.route('/image') def image(): global processed_image with lock: # Lock the shared variables between threads so they can't be accessed at the same time and you cant have half processed images if processed_image is not None: _, buffer = cv2.imencode('.jpg', processed_image) return Response(buffer.tobytes(), mimetype='image/jpeg') else: return "No image available", 404 @app.route('/yolo_results', methods=['GET']) def yolo_results_endpoint(): global yolo_results return jsonify(yolo_results) def yolo_results_db(): global yolo_results with lock: try: db = get_db() with db.cursor() as cursor: sql_yolo = "INSERT INTO image (class, confidence) VALUES (%s, %s)" yolo_tuples = [(result["class"], result["confidence"]) for result in yolo_results] print(f"YOLO Tuples: {yolo_tuples}") # Debug statement cursor.executemany(sql_yolo, yolo_tuples) db.commit() cursor.close() except mysql.connector.Error as err: print(f"Database error: {err}") except Exception as e: print(f"Unexpected error: {e}") if __name__ == '__main__': app.run(debug=True, port=5000)