from RPi import GPIO from helpers.RFIDklasse import RFIDReader from helpers.AfstandssensorKlasse import UltrasonicSensor from helpers.SpeakerKlasse import BluetoothMusicPlayer from helpers.MCP3008reader import MCP3008Reader from helpers.klasseknop import Button from helpers.OLEDdisplayKlasse import OLEDDisplay from helpers.acceleroSensorKlasse import PunchSpeedSensor from PIL import Image import spidev import subprocess from subprocess import getoutput import threading import datetime import time from repositories.DataRepository import DataRepository from flask import Flask, jsonify, request from flask_socketio import SocketIO, emit from flask_cors import CORS import signal import sys import queue audio_command_queue = queue.Queue() GPIO.setmode(GPIO.BCM) app = Flask(__name__) app.config['SECRET_KEY'] = 'HELLOTHISISSCERET' # ping interval forces rapid B2F communication socketio = SocketIO(app, cors_allowed_origins="*", async_mode='gevent', ping_interval=0.5) CORS(app) game_Running = 0 game_ID = 0 currentUserID = 0 current_datetime_notSTR = datetime.datetime.now() current_datetime = current_datetime_notSTR.strftime('%Y-%m-%d %H:%M:%S') #OLEDDisplay #Images must start with "path_to_image" path_to_image_MCT = '/home/user/2023-2024-projectone-mct-Maciej-Mitura/backend/MCT.bmp' path_to_image_Geert = '/home/user/2023-2024-projectone-mct-Maciej-Mitura/backend/Geert.bmp' OLED = OLEDDisplay() OLED.lock = threading.Lock() OLED.current_thread = None MAC_Bluetooh = 'BA:89:44:1C:E8:26' speaker = BluetoothMusicPlayer() path_Countdown = "/home/user/2023-2024-projectone-mct-Maciej-Mitura/backend/media/countdown.mp3" path_song = "/home/user/2023-2024-projectone-mct-Maciej-Mitura/backend/helpers/RoxyDaddy.mp3" path_Greatest = "/home/user/2023-2024-projectone-mct-Maciej-Mitura/backend/helpers/Greatest.mp3" def playMusic(path): audio_command_queue.put(("play", path)) print("Queued music for playback:", path) def stopMusic(): audio_command_queue.put(("stop", None)) print("Queued stop music command") def audio_thread(): while True: command, path = audio_command_queue.get() if command == "play": speaker.play(path) elif command == "stop": speaker.stop_music() audio_command_queue.task_done() def getIP(): ips = getoutput("hostname --all-ip-addresses").split() return ips [1] if len(ips) > 1 else ips[0] if ips else 'No IP found' current_IP = str(getIP()) content_array = [path_to_image_MCT, f"ProjectOne - Maciej Mitura - IP: {current_IP}", "Press the joystick to login by RFID", "Log in as guest"] current_index = 0 select_Once_Button = 0 canEnterSecond = 0 def selectByButton(channel): global current_index global game_Running global game_ID global currentUserID global select_Once_Button global current_datetime global canEnterSecond print(f"Button pressed, current index: {current_index}, content: {content_array[current_index]}") if current_index == 3 and select_Once_Button == 0 and game_Running == 0: select_Once_Button =1 gameid = DataRepository.get_last_game_id() #Previous gameID + 1 -> Current gameID tussenstap = gameid["game_id"] game_ID = tussenstap +1 print(f"GameID: {game_ID}") #Make it possible for the LOGGING IN BY RFID and starting the game to run only 1time print("-- Logging in as guest --") #OLED.write_text("Please login by using your RFID chip...") #time.sleep(0.1) #id, text = RFID.read_from_rfid() #time.sleep(0.1) #print(f"RFID read ID: {id}, Text: {text}") #Get userID by username #currentUserID_JSON = DataRepository.get_userID_by_RFID(text) currentUserID = 2 OLED.write_text(f"Welcome guest ") time.sleep(2) OLED.write_text(f"Press the button again to stop the training. Goal? Hit as fast and accurate as possible...") time.sleep(5) OLED.write_text("Ready?") time.sleep(1) playMusic(path_Countdown) OLED.write_text("Starting in 3...") time.sleep(1) OLED.write_text("Starting in 2...") time.sleep(1) OLED.write_text("Starting in 1...") time.sleep(1) OLED.write_text("START - GOOD LUCK ") playMusic(path_song) #Creating new gameID print("Creating new game in backend") print(f"Current user: {currentUserID}") print(f"Current datetime: {current_datetime}") DataRepository.create_game(currentUserID, current_datetime) socketio.emit("B2F_Starting_live_game", {"waarde": "start"}) time.sleep(1) OLED.display_image(path_to_image_MCT) #Toggling game_Running = 1 - game_Running canEnterSecond = 0 print(f"Game running: {game_Running}") if current_index == 3 and game_Running == 1 and canEnterSecond ==1: OLED.write_text("Stopping the game") time.sleep(1) OLED.clear_display() select_Once_Button = 0 game_Running = 1 - game_Running print("Injecting stoptime into database") current_datetime_notSTR = datetime.datetime.now() current_datetime = current_datetime_notSTR.strftime('%Y-%m-%d %H:%M:%S') print(f"Inject time:{ current_datetime}") print(f"Inject gameID: {game_ID}") okeoke = DataRepository.update_game_stoptime(current_datetime, game_ID) socketio.emit("B2F_Stopping_live_game", {"waarde": "stop"}) print(okeoke) stopMusic() OLED.display_image(path_to_image_MCT) current_index = 0 if current_index == 2 and select_Once_Button == 0 and game_Running == 0: select_Once_Button =1 gameid = DataRepository.get_last_game_id() #Previous gameID + 1 -> Current gameID tussenstap = gameid["game_id"] game_ID = tussenstap +1 print(f"GameID: {game_ID}") #Make it possible for the LOGGING IN BY RFID and starting the game to run only 1time print("-- Logging in by RFID --") OLED.write_text("Please login by using your RFID chip...") time.sleep(0.1) id, text = RFID.read_from_rfid() time.sleep(0.1) print(f"RFID read ID: {id}, Text: {text}") #Get userID by username currentUserID_JSON = DataRepository.get_userID_by_RFID(text) currentUserID = currentUserID_JSON["user_id"] OLED.write_text(f"Welcome {text}") time.sleep(2) OLED.write_text(f"Press the button again to stop the training. Goal? Hit as fast and accurate as possible...") time.sleep(5) OLED.write_text("Ready?") playMusic(path_Countdown) time.sleep(1) OLED.write_text("Starting in 3...") time.sleep(1) OLED.write_text("Starting in 2...") time.sleep(1) OLED.write_text("Starting in 1...") time.sleep(1) OLED.write_text("START - GOOD LUCK ") playMusic(path_Greatest) #Creating new gameID print("Creating new game in backend") print(f"Current user: {currentUserID}") print(f"Current datetime: {current_datetime}") DataRepository.create_game(currentUserID, current_datetime) socketio.emit("B2F_Stopping_live_game", {"waarde": "start"}) time.sleep(1) OLED.clear_display() #Toggling game_Running = 1 - game_Running canEnterSecond = 0 print(f"Game running: {game_Running}") if current_index == 2 and game_Running == 1 and canEnterSecond ==1: OLED.write_text("Stopping the game") time.sleep(1) OLED.clear_display() select_Once_Button = 0 game_Running = 1 - game_Running print("Injecting stoptime into database") current_datetime_notSTR = datetime.datetime.now() current_datetime = current_datetime_notSTR.strftime('%Y-%m-%d %H:%M:%S') print(f"Inject time:{ current_datetime}") print(f"Inject gameID: {game_ID}") okeoke = DataRepository.update_game_stoptime(current_datetime, game_ID) socketio.emit("B2F_Stopping_live_game", {"waarde": "stop"}) print(okeoke) stopMusic() OLED.display_image(path_to_image_MCT) current_index = 0 canEnterSecond = 1 time.sleep(1) def display_image(image_path): def display_image_worker(): OLED.display_image(image_path) thread = threading.Thread(target=display_image_worker) thread.start() def display_text(text): def display_text_worker(): with OLED.lock: OLED.write_text(text) if OLED.current_thread and OLED.current_thread.is_alive(): OLED.current_thread.join() OLED.current_thread = threading.Thread(target=display_text_worker) OLED.current_thread.start() def clear_screen(): def clear_screen_worker(): OLED.clear_display() thread = threading.Thread(target=clear_screen_worker) thread.start() turnOffPiButton = Button(12) def TurnOffPi(channel): clear_screen() print(f"Button {channel} has been pressed...") print(f"Shutting down the Pi.....") subprocess.call(["sudo", "shutdown", "now"]) turnOffPiButton.on_release(TurnOffPi) #Define objects/classes MCPreader = MCP3008Reader() punchMeter = PunchSpeedSensor() RFID = RFIDReader() button_joystick = Button(25) button_joystick.on_release(selectByButton) AfstandSens = UltrasonicSensor(13 ,26) stop_threads = threading.Event() def read_PunchMeter_values(): global current_index global game_Running global game_ID global currentUserID while True: if game_Running: DTnoSTR = datetime.datetime.now() curr_DateTi = DTnoSTR.strftime('%Y-%m-%d %H:%M:%S') max_velocity_kmh, duration = punchMeter.measure_punch_speed() max_velo_float = round(max_velocity_kmh,2) duration_float = round(duration,2) if max_velocity_kmh > 0: print(f"Velocity: {max_velo_float}, time: {duration_float}") socketio.emit('B2F_update_MPU_sensor', {'speed': max_velo_float, 'duration':duration_float}) DataRepository.create_new_history_record(6, max_velo_float, curr_DateTi, game_ID) DataRepository.create_new_history_record(7, duration_float, curr_DateTi, game_ID) time.sleep(0.25) # adjust the sleep time as necessary time.sleep(0.01) def ultrasonic_distance_values(): global current_index global game_Running global game_ID global currentUserID measurement_counter = 0 while True: if game_Running: abccc = datetime.datetime.now() deff = abccc.strftime('%Y-%m-%d %H:%M:%S') distance = AfstandSens.get_distance() if distance is not None and distance > 0: measurement_counter += 1 if measurement_counter % 5 == 0: DataRepository.create_new_history_record(5, distance, deff, game_ID) print(f"Added distance: {distance}cm") time.sleep(0.1) def read_pressure_sensors(): global current_index global game_Running global game_ID global currentUserID print("reading from pressure sensors") sensor_1_triggered = False sensor_2_triggered = False sensor_3_triggered = False threshold_value = 350 #Pressure sensor while True: #Sensor 1 #local_Game_Running = game_Running if game_Running: kkkk = datetime.datetime.now() okkkk = kkkk.strftime('%Y-%m-%d %H:%M:%S') sens_Pressure_1 = MCPreader.read_channel(0) if sens_Pressure_1 > threshold_value and not sensor_1_triggered: print(f"Received update for pressure sensor 1: {sens_Pressure_1}") socketio.emit('B2F_Bag_Punched', {'sensorID': 1}) DataRepository.create_new_history_record(1, sens_Pressure_1, okkkk, game_ID) sensor_1_triggered = True elif sens_Pressure_1 <= threshold_value and sensor_1_triggered: sensor_1_triggered = False #Sensor 2 sens_Pressure_2 = MCPreader.read_channel(1) if sens_Pressure_2 > threshold_value and not sensor_2_triggered: print(f"Received update for pressure sensor 2: {sens_Pressure_2}") socketio.emit('B2F_Bag_Punched', {'sensorID': 2}) DataRepository.create_new_history_record(2, sens_Pressure_2, okkkk, game_ID) sensor_2_triggered = True elif sens_Pressure_2 <= threshold_value and sensor_2_triggered: sensor_2_triggered = False #Sensor 3 sens_Pressure_3 = MCPreader.read_channel(2) if sens_Pressure_3 > threshold_value and not sensor_3_triggered: print(f"Received update for pressure sensor 3: {sens_Pressure_3}") socketio.emit('B2F_Bag_Punched', {'sensorID': 3}) DataRepository.create_new_history_record(3, sens_Pressure_3, okkkk, game_ID) sensor_3_triggered = True elif sens_Pressure_3 <= threshold_value and sensor_3_triggered: sensor_3_triggered = False time.sleep(0.01) def read_joystick_values(): global current_index while True: # Check for joystick event sens_Joystick_X = MCPreader.read_channel(4) if sens_Joystick_X > 700: # Move to the right if current_index < len(content_array) - 1: current_index += 1 else: current_index = 0 # Loop back to the start if at the end display_current_content() time.sleep(0.5) elif sens_Joystick_X < 350: # Move to the left if current_index > 0: current_index -= 1 else: current_index = len(content_array) - 1 # Loop back to the end if at the start display_current_content() time.sleep(0.5) def display_current_content(): current_content = content_array[current_index] if current_content.startswith("/home"): display_image(current_content) else: display_text(current_content) def start_threads(): thread_pressure_sensors = threading.Thread(target=read_pressure_sensors, daemon=True) punchMeter_thread = threading.Thread(target=read_PunchMeter_values, daemon=True) joystick_thread = threading.Thread(target=read_joystick_values, daemon=True) ultraSonic_sensor_thread = threading.Thread(target=ultrasonic_distance_values, daemon=True) threading.Thread(target=audio_thread, daemon=True).start() ultraSonic_sensor_thread.start() thread_pressure_sensors.start() punchMeter_thread.start() joystick_thread.start() print("threads started") # API ENDPOINTS @app.route('/') def hallo(): return "Server is running, er zijn momenteel geen API endpoints beschikbaar." @app.route('/api/v1/checkifexists/', methods=["GET", "POST"]) def checkIfUserExists(): if request.method == 'POST': try: gegevens = DataRepository.json_or_formdata(request) data = DataRepository.get_user(gegevens['username'],gegevens['password']) print(data) if data == None: return jsonify(user = False), 200 else: return jsonify(user = data), 200 except ValueError: print("hier") @app.route('/api/v1/createNewUser/', methods=["GET","POST"]) def createNewUser(): if request.method == "POST": gegevens = DataRepository.json_or_formdata(request) data = DataRepository.create_user(gegevens['username'],gegevens['password'], gegevens['type']) #print(gegevens) if data == 0: raise ValueError else: return jsonify(user = data), 201 @app.route('/api/v1/retrieveLastRecords/', methods=["GET", "POST"]) def retrieveLastRecords(): if request.method == "GET": data = DataRepository.retrieve_last_records() if data == 0: raise ValueError else: return jsonify(records = data), 201 # SOCKET IO @socketio.on('connect') def initial_connection(): print('A new client connect') # # Send to the client! # vraag de status op van de lampen uit de DB #status = DataRepository.read_status_lampen() # socketio.emit('B2F_status_lampen', {'lampen': status}) # Beter is het om enkel naar de client te sturen die de verbinding heeft gemaakt. #emit('B2F_status_lampen', {'lampen': status}, broadcast=False) @socketio.on('F2B_create_new_game') def create_new_gameID(): print("Creating new gameID in database") DataRepository.create_game(1, "2024-05-29 14:01:55", "2024-05-29 14:04:21") print("Created successfully") socketio.emit("B2F_confirm_create_new_game", 201) @socketio.on('F2B_check_for_values_RFID') def check_for_user_by_RFID(): print("Checking for user") id, password = RFID.read_from_rfid() print(f"ID: {id}") print(f"Password: {password}") body = { 'id': id, 'password': password } user_exists = DataRepository.check_if_rfid_data_exists(id, password) print(user_exists) if user_exists == 1: socketio.emit("B2F_user_existence_confirmed") socketio.emit("B2F_confirm_create_new_game") else: socketio.emit("B2F_user_existence_declined") socketio.emit('B2F_data_from_chip', body) @socketio.on('F2B_shutdown_pi') def shutting_down(): print('received a shutdown function from webserver') clear_screen() subprocess.call(["sudo", "shutdown", "now"]) def signal_handler(sig, frame): print('Exiting gracefully') stop_threads.set() # Signal threads to stop sys.exit(0) if __name__ == '__main__': signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) try: speaker.connect(MAC_Bluetooh) display_image(path_to_image_MCT) start_threads() #print(f"IP: {current_IP}") print("**** Starting APP ****") socketio.run(app, debug=False, host='0.0.0.0') except KeyboardInterrupt: print('KeyboardInterrupt exception is caught') finally: clear_screen() print("finished")