update
This commit is contained in:
Your Name
142
SARA/R5/setPDP.py
Normal file
142
SARA/R5/setPDP.py
Normal file
@@ -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'<span style="color: red;font-weight: bold;">[ALERT] ⚠️{total_elapsed_time:.2f}s⚠️</span>')
|
||||
|
||||
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
|
||||
@@ -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('<span style="color: orange;font-weight: bold;">🔄SARA reboot!🔄</span>')
|
||||
command = f'AT+CFUN=15\r'
|
||||
ser_sara.write(command.encode('utf-8'))
|
||||
@@ -638,6 +639,8 @@ try:
|
||||
print(response_SARA_9r)
|
||||
print("</p>")
|
||||
|
||||
reset_uSpot_url = True
|
||||
|
||||
print("Sleep 10 secs before continuing")
|
||||
time.sleep(10)
|
||||
|
||||
@@ -651,14 +654,14 @@ try:
|
||||
print("</p>", end="")
|
||||
|
||||
|
||||
# 2.2 code 1 (HHTP succeded)
|
||||
# 2.2 code 1 (✅✅HHTP / UUHTTPCR succeded✅✅)
|
||||
else:
|
||||
# Si la commande HTTP a réussi
|
||||
print('<span style="font-weight: bold;">✅✅HTTP operation successful.</span>')
|
||||
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('➡️<p class="fw-bold">SEND TO uSPOT SERVERS</p>')
|
||||
|
||||
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>,<internal_name>,<data_size>
|
||||
# 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
|
||||
|
||||
Reference in New Issue
Block a user