""" Main loop to gather data from sensor: * NPM * I2C BME280 * Noise sensor and send it to AirCarto servers via SARA R4 MQTT requests """ import board import json import serial import time import busio import RPi.GPIO as GPIO from adafruit_bme280 import basic as adafruit_bme280 # Record the start time of the script start_time = time.time() url="data.nebuleair.fr" # Set up GPIO mode (for Blue LED: network status) GPIO.setwarnings(False) GPIO.setmode(GPIO.BCM) # Use Broadcom pin numbering GPIO.setup(23, GPIO.OUT) # Set GPIO23 as an output pin # Add yellow color to the output yellow_color = "\033[33m" # ANSI escape code for yellow red_color = "\033[31m" # ANSI escape code for red reset_color = "\033[0m" # Reset color to default green_color = "\033[32m" # Green #get data from config def load_config(config_file): try: with open(config_file, 'r') as file: config_data = json.load(file) return config_data except Exception as e: print(f"Error loading config file: {e}") return {} def update_json_key(file_path, key, value): """ Updates a specific key in a JSON file with a new value. :param file_path: Path to the JSON file. :param key: The key to update in the JSON file. :param value: The new value to assign to the key. """ try: # Load the existing data with open(file_path, "r") as file: data = json.load(file) # Check if the key exists in the JSON file if key in data: data[key] = value # Update the key with the new value else: print(f"Key '{key}' not found in the JSON file.") return # Write the updated data back to the file with open(file_path, "w") as file: json.dump(data, file, indent=2) # Use indent for pretty printing print(f"Successfully updated '{key}' to '{value}'.") except Exception as e: print(f"Error updating the JSON file: {e}") # Define the config file path config_file = '/var/www/nebuleair_pro_4g/config.json' # Load the configuration data config = load_config(config_file) # Access the shared variables baudrate = config.get('SaraR4_baudrate', 115200) #baudrate du sara R4 device_id = config.get('deviceID', '').upper() #device ID en maj need_to_log = config.get('loop_log', False) #inscription des logs bme_280_config = config.get('i2c_BME', False) #présence du BME280 i2C_sound_config = config.get('i2C_sound', False) #présence du BME280 ser = serial.Serial( port='/dev/ttyAMA2', baudrate=baudrate, #115200 ou 9600 parity=serial.PARITY_NONE, #PARITY_NONE, PARITY_EVEN or PARITY_ODD stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 2 ) ser_NPM = serial.Serial( port='/dev/ttyAMA5', baudrate=115200, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout = 1 ) def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2): response = bytearray() serial_connection.timeout = timeout end_time = time.time() + end_of_response_timeout while True: if serial_connection.in_waiting > 0: data = serial_connection.read(serial_connection.in_waiting) response.extend(data) end_time = time.time() + end_of_response_timeout # Reset timeout on new data elif time.time() > end_time: break time.sleep(0.1) # Short sleep to prevent busy waiting # Decode the response and filter out empty lines decoded_response = response.decode('utf-8') non_empty_lines = "\n".join(line for line in decoded_response.splitlines() if line.strip()) # Add yellow color to the output colored_output = f"{yellow_color}{non_empty_lines}\n{reset_color}" return colored_output # Open and read the JSON file try: # Send the command to request data (e.g., data for 60 seconds) ser_NPM.write(b'\x81\x12\x6D') # Read the response byte_data = ser_NPM.readline() #if npm is disconnected byte_data is empty # Extract the state byte and PM data from the response state_byte = int.from_bytes(byte_data[2:3], byteorder='big') state_bits = [int(bit) for bit in bin(state_byte)[2:].zfill(8)] PM1 = int.from_bytes(byte_data[9:11], byteorder='big') / 10 PM25 = int.from_bytes(byte_data[11:13], byteorder='big') / 10 PM10 = int.from_bytes(byte_data[13:15], byteorder='big') / 10 # Create a dictionary with the parsed data data = { 'sondeID': device_id, 'PM1': PM1, 'PM25': PM25, 'PM10': PM10 } message = f"{data['PM1']},{data['PM25']},{data['PM10']}" if bme_280_config: #on récupère les infos du BME280 et on les ajoute au message i2c = busio.I2C(board.SCL, board.SDA) bme280 = adafruit_bme280.Adafruit_BME280_I2C(i2c, address=0x76) bme280.sea_level_pressure = 1013.25 # Update this value for your location data['temp'] = round(bme280.temperature, 2) data['hum'] = round(bme280.humidity, 2) data['press'] = round(bme280.pressure, 2) message += f",{data['temp']},{data['hum']},{data['press']}" if i2C_sound_config: #on récupère les infos de sound_metermoving et on les ajoute au message file_path_data_noise = "/var/www/nebuleair_pro_4g/sound_meter/moving_avg_minute.txt" # Read the file and extract the numbers try: with open(file_path_data_noise, "r") as file: content = file.read().strip() avg_noise, max_noise, min_noise = map(int, content.split()) # Append the variables to the JSON and to the message data['avg_noise'] = avg_noise data['max_noise'] = max_noise data['min_noise'] = min_noise #get BME280 data (SAFE: it returns none if the key do not exist) message = f"{data.get('PM1', '')},{data.get('PM25', '')},{data.get('PM10', '')},{data.get('temp', '')},{data.get('hum', '')},{data.get('press', '')},{avg_noise},{max_noise},{min_noise}" print(message) # Display the message or send it further except FileNotFoundError: print(f"Error: File {file_path} not found.") except ValueError: print("Error: File content is not valid numbers.") # Print the content of the JSON file if need_to_log: print("Data from sensors:") print(json.dumps(data, indent=4)) # Pretty print the JSON data #Write Data to saraR4 #1. MQTT profile configuration # Note: you need to logout first to change the config print("") #print("1.PROFILE CONFIG") #print(" 1.A. READ CONFIG") #command = f'AT+UMQTT?\r' #ser.write((command + '\r').encode('utf-8')) #response_SARA_1 = read_complete_response(ser) #if need_to_log: # print(response_SARA_1) # La config s'efface à chaque redémarrage! need_to_update_config = False if need_to_update_config: print("1.B. SET CONFIG") #command = f'AT+UMQTT=1,1883\r' #MQTT local TCP port number command = f'AT+UMQTT=2,"aircarto.fr"\r' #MQTT server name #command = f'AT+UMQTT=3,"193.252.54.10"\r' # MQTT server IP address #command = f'AT+UMQTT=12,1\r' # MQTT clean session ser.write((command + '\r').encode('utf-8')) response_SARA_1 = read_complete_response(ser) if need_to_log: print(response_SARA_1) lines = response_SARA_1.strip().splitlines() for line in lines: if line.startswith("+UMQTT"): # Split the line by commas and get the last number parts = line.split(',') last_number = parts[-1].strip() # Get the last part and strip any whitespace if int(last_number) == 1: print(f"{green_color}MQTT profile configuration SUCCEDED{reset_color}") else: print(f"{green_color}ERROR: MQTT profile configuration fail{reset_color}") #2. MQTT login need_to_update_login = False if need_to_update_login: print("") print("2.MQTT LOGIN") #command = f'AT+UMQTTC=1\r' #MQTT login command = f'AT+UMQTTC=0\r' #MQTT logout ser.write((command + '\r').encode('utf-8')) response_SARA_2 = read_complete_response(ser, 8, 8) if need_to_log: print(response_SARA_2) lines = response_SARA_2.strip().splitlines() for line in lines: if line.startswith("+UMQTTC"): parts = line.split(',') first_number = parts[0].replace("+UMQTTC:", "").strip() last_number = parts[-1].strip() # Get the last part and strip any whitespace #print(f"Last number: {last_number}") if int(first_number) == 0: print("MQTT logout command ->") if int(first_number) == 1: print("MQTT login command ->") if int(last_number) == 1: print(f"{green_color}SUCCESS{reset_color}") else: print(f"{red_color}FAIL{reset_color}") if line.startswith("+UUMQTTC"): parts = line.split(',') first_number = parts[0].replace("+UUMQTTC:", "").strip() last_number = parts[-1].strip() # Get the last part and strip any whitespace if int(first_number) == 1: print("MQTT login result ->") if int(last_number) == 0: print(f"{green_color}connection accepted{reset_color}") if int(last_number) == 1: print(f"{green_color}the server does not support the level of the MQTT protocol requested by the Client{reset_color}") if int(last_number) == 2: print(f"{green_color} the client identifier is correct UTF-8 but not allowed by the Server{reset_color}") if int(last_number) == 3: print(f"{green_color} the network connection has been made but the MQTT service is unavailable{reset_color}") #3. MQTT publish print("") print("3.MQTT PUBLISH") command = f'AT+UMQTTC=2,0,0,"nebuleair/pro/{device_id}/data","{message}"\r' ser.write((command + '\r').encode('utf-8')) response_SARA_3 = read_complete_response(ser) if need_to_log: print(response_SARA_3) lines = response_SARA_3.strip().splitlines() for line in lines: if line.startswith("+UMQTTC"): parts = line.split(',') first_number = parts[0].replace("+UMQTTC:", "").strip() last_number = parts[-1].strip() # Get the last part and strip any whitespace if int(first_number) == 2: print("MQTT Publish ->") if int(last_number) == 1: print(f"{green_color}SUCCESS{reset_color}") else: print(f"{red_color}FAIL{reset_color}") # Calculate and print the elapsed time elapsed_time = time.time() - start_time if need_to_log: print("") print(f"Elapsed time: {elapsed_time:.2f} seconds") print("----------------------------------------") print("----------------------------------------") except Exception as e: print(f"Error reading the JSON file: {e}")