diff --git a/SARA/cellLocate/server_conf.py b/SARA/cellLocate/server_conf.py new file mode 100644 index 0000000..6f0c0c1 --- /dev/null +++ b/SARA/cellLocate/server_conf.py @@ -0,0 +1,103 @@ +''' + ____ _ ____ _ + / ___| / \ | _ \ / \ + \___ \ / _ \ | |_) | / _ \ + ___) / ___ \| _ < / ___ \ + |____/_/ \_\_| \_\/_/ \_\ + +Script to Configures the network connection to a Multi GNSS Assistance (MGA) server used also per CellLocate + +/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/cellLocate/server_conf.py ttyAMA2 1 + +AT+UGSRV="cell-live1.services.u-blox.com","cell-live2.services.u-blox.com","XkEKfGqVSbmNE1eZfBZm4Q" + + +''' + +import serial +import time +import sys +import json + +parameter = sys.argv[1:] # Exclude the script name +port='/dev/'+parameter[0] # ex: ttyAMA2 +timeout = float(parameter[1]) # ex:2 + + +#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) + +ser = 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 = timeout +) + +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 + + +#command = f'ATI\r' +command = f'AT+UGSRV="cell-live1.services.u-blox.com","cell-live2.services.u-blox.com","XkEKfGqVSbmNE1eZfBZm4Q"\r' +ser.write((command + '\r').encode('utf-8')) + +response = read_complete_response(ser, wait_for_lines=["+UULOC"]) +print(response) diff --git a/SARA/reboot/start.py b/SARA/reboot/start.py index 4e0e215..8502746 100644 --- a/SARA/reboot/start.py +++ b/SARA/reboot/start.py @@ -121,19 +121,38 @@ try: print('

Start reboot python script

') #check modem status + #Attention: + # SARA R4 response: Manufacturer: u-blox Model: SARA-R410M-02B + # SArA R5 response: SARA-R500S-01B-00 print("⚙️Check SARA Status") command = f'ATI\r' ser_sara.write(command.encode('utf-8')) response_SARA_ATI = read_complete_response(ser_sara, wait_for_lines=["IMEI"]) print(response_SARA_ATI) - match = re.search(r"Model:\s*(.+)", response_SARA_ATI) - model = match.group(1).strip() if match else "Unknown" # Strip unwanted characters - print(f" Model: {model}") + + # Check for SARA model with more robust regex + model = "Unknown" + if "SARA-R410M" in response_SARA_ATI: + model = "SARA-R410M" + print("📱 Detected SARA R4 modem") + elif "SARA-R500" in response_SARA_ATI: + model = "SARA-R500" + print("📱 Detected SARA R5 modem") + else: + # Fallback to regex match if direct string match fails + match = re.search(r"Model:\s*([A-Za-z0-9\-]+)", response_SARA_ATI) + if match: + model = match.group(1).strip() + else: + model = "Unknown" + print("⚠️ Could not identify modem model") + + print(f"🔍 Model: {model}") update_json_key(config_file, "modem_version", model) time.sleep(1) - # 1. Set AIRCARTO URL + # 1. Set AIRCARTO URL (profile id = 0) print('➡️Set aircarto URL') aircarto_profile_id = 0 aircarto_url="data.nebuleair.fr" @@ -143,26 +162,74 @@ try: print(response_SARA_1) time.sleep(1) - #2. Set 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) - print("set port 81") - command = f'AT+UHTTP={uSpot_profile_id},5,81\r' + #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) #3. Get localisation (CellLocate) - mode = 2 - sensor = 2 + mode = 2 #single shot position + sensor = 2 #use cellular CellLocate® location information response_type = 0 timeout_s = 2 accuracy_m = 1