From f1d6f595ac3e7f4b035f20e2a51f8e68c0f5c126 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 14 Mar 2025 09:28:28 +0100 Subject: [PATCH] update --- SARA/R5/setPDP.py | 142 ++++++++++++++++++++++++++++++++++++++ loop/SARA_send_data_v2.py | 83 ++++++++++++++++++++-- 2 files changed, 218 insertions(+), 7 deletions(-) create mode 100644 SARA/R5/setPDP.py diff --git a/SARA/R5/setPDP.py b/SARA/R5/setPDP.py new file mode 100644 index 0000000..bdafc40 --- /dev/null +++ b/SARA/R5/setPDP.py @@ -0,0 +1,142 @@ +''' + ____ _ ____ _ + / ___| / \ | _ \ / \ + \___ \ / _ \ | |_) | / _ \ + ___) / ___ \| _ < / ___ \ + |____/_/ \_\_| \_\/_/ \_\ + +Script to set the PDP context for the SARA R5 + + +/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/R5/setPDP.py + +''' +import serial +import time +import sys +import json +import re + +#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 {} + +#Fonction pour mettre à jour le JSON de configuration +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"💾 updating '{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) +baudrate = config.get('SaraR4_baudrate', 115200) #baudrate du sara R4 +device_id = config.get('deviceID', '').upper() #device ID en maj + +ser_sara = 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 +) + +def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2, wait_for_lines=None, debug=True): + ''' + Fonction très importante !!! + Reads the complete response from a serial connection and waits for specific lines. + ''' + if wait_for_lines is None: + wait_for_lines = [] # Default to an empty list if not provided + + response = bytearray() + serial_connection.timeout = timeout + end_time = time.time() + end_of_response_timeout + start_time = time.time() + + while True: + elapsed_time = time.time() - start_time # Time since function start + 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 + + # Decode and check for any target line + decoded_response = response.decode('utf-8', errors='replace') + for target_line in wait_for_lines: + if target_line in decoded_response: + if debug: + print(f"[DEBUG] 🔎 Found target line: {target_line} (in {elapsed_time:.2f}s)") + return decoded_response # Return response immediately if a target line is found + elif time.time() > end_time: + if debug: + print(f"[DEBUG] Timeout reached. No more data received.") + break + time.sleep(0.1) # Short sleep to prevent busy waiting + + # Final response and debug output + total_elapsed_time = time.time() - start_time + if debug: + print(f"[DEBUG] ⏱️ elapsed time: {total_elapsed_time:.2f}s. ⏱️") + # Check if the elapsed time exceeded 10 seconds + if total_elapsed_time > 10 and debug: + print(f"[ALERT] 🚨 The operation took too long 🚨") + print(f'[ALERT] ⚠️{total_elapsed_time:.2f}s⚠️') + + return response.decode('utf-8', errors='replace') # Return the full response if no target line is found + +try: + print('Start script') + + # 1. Set AIRCARTO URL (profile id = 0) + print('➡️Check SARA R5 connexion') + command = f'ATI0\r' + ser_sara.write(command.encode('utf-8')) + response_SARA_1 = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_1, end="") + time.sleep(1) + + # 2. Activate PDP context 1 + print('➡️Activate PDP context 1') + command = f'AT+CGACT=1,1\r' + ser_sara.write(command.encode('utf-8')) + response_SARA_2 = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_2, end="") + time.sleep(1) + + + +except Exception as e: + print("An error occurred:", e) + traceback.print_exc() # This prints the full traceback \ No newline at end of file diff --git a/loop/SARA_send_data_v2.py b/loop/SARA_send_data_v2.py index 8646b98..5aaf82f 100755 --- a/loop/SARA_send_data_v2.py +++ b/loop/SARA_send_data_v2.py @@ -220,6 +220,7 @@ wind_meter= config.get('windMeter/read.py', False) send_aircarto = config.get('send_aircarto', True) #envoi sur AirCarto (data.nebuleair.fr) send_uSpot = config.get('send_uSpot', False) #envoi sur MicroSpot () +reset_uSpot_url = False selected_networkID = int(config.get('SARA_R4_neworkID', 0)) npm_5channel = config.get('NextPM_5channels', False) #5 canaux du NPM @@ -587,10 +588,9 @@ try: led_thread.start() else: - # 2.Si la réponse contient une réponse HTTP valide - # Extract HTTP response code from the last line - # ATTENTION: lines[-1] renvoie l'avant dernière ligne et il peut y avoir un soucis avec le OK - # rechercher plutot + # 2.Si la réponse contient une réponse UUHTTPCR + # Extract UUHTTPCR response code from the last line + http_response = lines[-1] # "+UUHTTPCR: 0,4,0" parts = http_response.split(',') @@ -629,7 +629,8 @@ try: 73 Secure socket connect error ''' - #Essayer un reboot du SARA R4 + #Reboot du SARA R4 + #ATTENTION: sur le SARA R5 la commande renvoie une erreur print('🔄SARA reboot!🔄') command = f'AT+CFUN=15\r' ser_sara.write(command.encode('utf-8')) @@ -638,6 +639,8 @@ try: print(response_SARA_9r) print("

") + reset_uSpot_url = True + print("Sleep 10 secs before continuing") time.sleep(10) @@ -651,14 +654,14 @@ try: print("

", end="") - # 2.2 code 1 (HHTP succeded) + # 2.2 code 1 (✅✅HHTP / UUHTTPCR succeded✅✅) else: - # Si la commande HTTP a réussi print('✅✅HTTP operation successful.') update_json_key(config_file, "SARA_R4_network_status", "connected") print("Blink blue LED") led_thread = Thread(target=blink_led, args=(23, 5, 0.5)) led_thread.start() + #4. Read reply from server print("Reply from server:") ser_sara.write(b'AT+URDFILE="aircarto_server_response.txt"\r') @@ -797,6 +800,72 @@ try: if send_uSpot: print('➡️

SEND TO uSPOT SERVERS

') + if reset_uSpot_url: + #2. Set uSpot URL (profile id = 1) + print('➡️Set uSpot URL') + uSpot_profile_id = 1 + uSpot_url="api-prod.uspot.probesys.net" + security_profile_id = 1 + + #step 1: import the certificate + print("****") + certificate_name = "e6" + with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.pem", "rb") as cert_file: + certificate = cert_file.read() + size_of_string = len(certificate) + + print("\033[0;33m Import certificate\033[0m") + # AT+USECMNG=0,,, + # type-> 0 -> trusted root CA + command = f'AT+USECMNG=0,0,"{certificate_name}",{size_of_string}\r' + ser_sara.write((command + '\r').encode('utf-8')) + response_SARA_1 = read_complete_response(ser_sara) + print(response_SARA_1) + + time.sleep(0.5) + + print("\033[0;33mAdd certificate\033[0m") + ser_sara.write(certificate) + response_SARA_2 = read_complete_response(ser_sara) + print(response_SARA_2) + + time.sleep(0.5) + + # SECURITY PROFILE + # op_code: 3 -> trusted root certificate internal name + print("\033[0;33mSet the security profile (choose cert)\033[0m") + command = f'AT+USECPRF={security_profile_id},3,"{certificate_name}"\r' + ser_sara.write((command + '\r').encode('utf-8')) + response_SARA_5c = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_5c) + time.sleep(0.5) + + #step 4: set url (op_code = 1) + command = f'AT+UHTTP={uSpot_profile_id},1,"{uSpot_url}"\r' + ser_sara.write(command.encode('utf-8')) + response_SARA_2 = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_2) + time.sleep(1) + + #step 4: set PORT (op_code = 5) + print("set port 443") + command = f'AT+UHTTP={uSpot_profile_id},5,443\r' + ser_sara.write((command + '\r').encode('utf-8')) + response_SARA_55 = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_55) + time.sleep(1) + + #step 4: set url to SSL (op_code = 6) (http_secure = 1 for HTTPS)(USECMNG_PROFILE = 2) + print("\033[0;33mSET SSL\033[0m") + http_secure = 1 + command = f'AT+UHTTP={uSpot_profile_id},6,{http_secure},{security_profile_id}\r' + #command = f'AT+UHTTP={profile_id},6,{http_secure}\r' + + ser_sara.write(command.encode('utf-8')) + response_SARA_5 = read_complete_response(ser_sara, wait_for_lines=["OK"]) + print(response_SARA_5) + time.sleep(1) + # 1. Open sensordata_json.json (with correct data size) print("Open JSON:") payload_string = json.dumps(payload_json) # Convert dict to JSON string