''' Script to set the URL for a HTTP request and trigger the POST Request Ex: /usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 api-prod.uspot.probesys.net /nebuleair?token=2AFF6dQk68daFZ /usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 webhook.site /6bee2237-099a-4ff4-8452-9f4126df7151 /usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 aircarto.fr /tests/test.php /usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 ssl.aircarto.fr /test.php /usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/SSL/test_22.py ttyAMA2 vps.aircarto.fr /test.php First profile id: AT+UHTTP=0,1,"data.nebuleair.fr" Second profile id: AT+UHTTP=1,1,"api-prod.uspot.probesys.net" Third profile id: AT+UHTTP=2,1,"aircarto.fr" ''' import serial import time import sys import json parameter = sys.argv[1:] # Exclude the script name #print("Parameters received:") port='/dev/'+parameter[0] # ex: ttyAMA2 url = parameter[1] # ex: data.mobileair.fr endpoint = parameter[2] profile_id = 3 #get baudrate 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 {} # 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) send_uSpot = config.get('send_uSpot', False) def color_text(text, color): colors = { "red": "\033[31m", "green": "\033[32m", "yellow": "\033[33m", "blue": "\033[34m", "magenta": "\033[35m", "cyan": "\033[36m", "white": "\033[37m", } reset = "\033[0m" return f"{colors.get(color, '')}{text}{reset}" def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2, wait_for_line=None): 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 the specific line if wait_for_line: decoded_response = response.decode('utf-8', errors='replace') if wait_for_line in decoded_response: print(f"[DEBUG] 🔎Found target line: {wait_for_line}") break elif time.time() > end_time: 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 print(f"[DEBUG] ⏱️ elapsed time: {total_elapsed_time:.2f}s. ⏱️") return response.decode('utf-8', errors='replace') ser_sara = serial.Serial( port=port, #USB0 or ttyS0 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 ) try: #step 1: import the certificate print("****") with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/isrgrootx1.der", "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,"isrgrootx1",{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) #check certificate (List all available certificates and private keys) print("\033[0;33mCheck certificate\033[0m") command = f'AT+USECMNG=3\r' ser_sara.write((command + '\r').encode('utf-8')) response_SARA_5b = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_5b) time.sleep(0.5) # ******************************* # SECURITY PROFILE # AT+USECPRF=[,[,]] security_profile_id = 2 # op_code: 0 -> certificate validation level # param_val : 0 -> Level 0 No validation; 1-> Level 1 Root certificate validation print("\033[0;33mSet the security profile (params)\033[0m") certification_level=1 command = f'AT+USECPRF={security_profile_id},0,{certification_level}\r' ser_sara.write((command + '\r').encode('utf-8')) response_SARA_5b = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_5b) time.sleep(0.5) # 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,"isrgrootx1"\r' ser_sara.write((command + '\r').encode('utf-8')) response_SARA_5c = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_5c) time.sleep(0.5) # ************************* # ************************* #step 4: set url (op_code = 1) print("\033[0;33mSET URL\033[0m") command = f'AT+UHTTP={profile_id},1,"{url}"\r' ser_sara.write((command + '\r').encode('utf-8')) response_SARA_5 = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_5) 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={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_line="OK") print(response_SARA_5) time.sleep(1) # Write Data to saraR4 payload_json = { "nebuleairid": "C04F8B8D3A08", "software_version": "ModuleAir-V1-012023", "sensordatavalues": [ { "value_type": "NPM_P0", "value": "2.3" }, { "value_type": "NPM_P0", "value": "3.30" }, { "value_type": "NPM_P1", "value": "9.05" }, { "value_type": "NPM_P2", "value": "20.60" }, { "value_type": "NPM_N1", "value": "49.00" }, { "value_type": "NPM_N10", "value": "49.00" }, { "value_type": "NPM_N25", "value": "49.00" }, { "value_type": "BME280_temperature", "value": "25.82" } ] } # 1. Open sensordata_csv.json (with correct data size) payload_string = json.dumps(payload_json) # Convert dict to JSON string size_of_string = len(payload_string) print("\033[0;33mOPEN JSON\033[0m") command = f'AT+UDWNFILE="sensordata_json.json",{size_of_string}\r' ser_sara.write(command.encode('utf-8')) response_SARA_1 = read_complete_response(ser_sara, wait_for_line=">") print(response_SARA_1) time.sleep(0.5) #2. Write to shell print("\033[0;33mWrite to Memory\033[0m") ser_sara.write(payload_string.encode()) response_SARA_2 = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_2) #step 4: trigger the request (http_command=1 for GET and http_command=1 for POST) print("****") print("\033[0;33mPOST REQUEST\033[0m") #parameter (POST) command = f'AT+UHTTPC={profile_id},4,"{endpoint}","https.resp","sensordata_json.json",4\r' #AIRCARTO #command = f'AT+UHTTPC={profile_id},4,"/tests/test.php","https.resp","sensordata_json.json",4\r' #uSPOT #command = f'AT+UHTTPC={profile_id},4,"/nebuleair?token=2AFF6dQk68daFZ","https.resp","sensordata_json.json",4\r' #AtmoSud #command = f'AT+UHTTPC={profile_id},1,"/","https.resp"\r' #Webhook #command = f'AT+UHTTPC={profile_id},4,"/6bee2237-099a-4ff4-8452-9f4126df7151","https.resp","sensordata_json.json",4\r' ser_sara.write(command.encode('utf-8')) # Wait for the +UUHTTPCR response print("Waiting for +UUHTTPCR response...") response_SARA_3 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=30, wait_for_line="+UUHTTPCR") print("\033[0;34m") print(response_SARA_3) print("\033[0m") if "+UUHTTPCR" in response_SARA_3: print("✅ Received +UUHTTPCR response.") lines = response_SARA_3.strip().splitlines() http_response = lines[-1] # "+UUHTTPCR: 0,4,0" parts = http_response.split(',') # code 0 (HTTP failed) if len(parts) == 3 and parts[-1] == '0': # The third value indicates success print("\033[0;31mATTENTION: HTTP operation failed\033[0m") else: print("\033[0;32m HTTP operation successful!!!\033[0m") #READ REPLY print("****") print("\033[0;33mREPLY SERVER\033[0m") ser_sara.write(b'AT+URDFILE="https.resp"\r') response_SARA_7 = read_complete_response(ser_sara, wait_for_line="OK") print("Reply from server:") print("\033[0;32m") print(response_SARA_7) print("\033[0m") #5. empty json print("\033[0;33mEmpty Memory\033[0m") ser_sara.write(b'AT+UDELFILE="sensordata_json.json"\r') response_SARA_8 = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_8) # Get error code print("\033[0;33mEmpty Memory\033[0m") command = f'AT+UHTTPER={profile_id}\r' ser_sara.write(command.encode('utf-8')) response_SARA_9 = read_complete_response(ser_sara, wait_for_line="OK") print(response_SARA_9) ''' +UHTTPER: profile_id,error_class,error_code error_class 0 OK, no error 3 HTTP Protocol error class 10 Wrong HTTP API USAGE error_code (for error_class 3) 0 No error 11 Server connection error 73 Secure socket connect error ''' except serial.SerialException as e: print(f"Error: {e}") finally: if ser_sara.is_open: ser_sara.close() print("****") #print("Serial closed")