update
This commit is contained in:
PaulVua
@@ -22,8 +22,7 @@ parameter = sys.argv[1:] # Exclude the script name
|
||||
#print("Parameters received:")
|
||||
port='/dev/'+parameter[0] # ex: ttyAMA2
|
||||
url = parameter[1] # ex: data.mobileair.fr
|
||||
|
||||
profile_id = 3
|
||||
profile_id = 2
|
||||
|
||||
#get baudrate
|
||||
def load_config(config_file):
|
||||
@@ -43,20 +42,45 @@ config = load_config(config_file)
|
||||
baudrate = config.get('SaraR4_baudrate', 115200)
|
||||
send_uSpot = config.get('send_uSpot', False)
|
||||
|
||||
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2):
|
||||
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(
|
||||
@@ -72,51 +96,105 @@ ser_sara = serial.Serial(
|
||||
try:
|
||||
#step 1: import the certificate
|
||||
print("****")
|
||||
print("Add certificate")
|
||||
with open("/var/www/nebuleair_pro_4g/SARA/SSL/isrgrootx1.der", "rb") as cert_file:
|
||||
with open("/var/www/nebuleair_pro_4g/SARA/SSL/certificate/e6.der", "rb") as cert_file:
|
||||
certificate = cert_file.read()
|
||||
|
||||
size_of_string = len(certificate)
|
||||
|
||||
command = f'AT+USECMNG=0,0,"myCertificate2",{size_of_string}\r'
|
||||
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,"e6",{size_of_string}\r'
|
||||
ser_sara.write((command + '\r').encode('utf-8'))
|
||||
response_SARA_1 = read_complete_response(ser_sara)
|
||||
print("Write certificate metadata")
|
||||
print(response_SARA_1)
|
||||
|
||||
time.sleep(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("Write certificate data")
|
||||
print(response_SARA_2)
|
||||
|
||||
time.sleep(1)
|
||||
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=<profile_id>[,<op_code>[,<param_val>]]
|
||||
|
||||
|
||||
# op_code: 0 -> certificate validation level
|
||||
# param_val : 0 -> Level 0 No validation; 1-> Level 1 Root certificate validation
|
||||
security_profile_id = 0
|
||||
print("\033[0;33mSet the security profile (params)\033[0m")
|
||||
command = f'AT+USECPRF={security_profile_id},0,0\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: 1 -> minimum SSL/TLS version
|
||||
# param_val : 0 -> any; server can use any version for the connection; 1-> LSv1.0;
|
||||
print("\033[0;33mSet the security profile (params)\033[0m")
|
||||
command = f'AT+USECPRF={security_profile_id},1,0\r'
|
||||
ser_sara.write((command + '\r').encode('utf-8'))
|
||||
response_SARA_5bb = read_complete_response(ser_sara, wait_for_line="OK")
|
||||
print(response_SARA_5bb)
|
||||
time.sleep(0.5)
|
||||
|
||||
#op_code: 2 -> cipher suite
|
||||
print("\033[0;33mSet cipher \033[0m")
|
||||
command = f'AT+USECPRF={security_profile_id},2,0\r'
|
||||
ser_sara.write((command + '\r').encode('utf-8'))
|
||||
response_SARA_5cc = read_complete_response(ser_sara, wait_for_line="OK")
|
||||
print(response_SARA_5cc)
|
||||
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,"e6"\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("****")
|
||||
print("SET URL")
|
||||
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)
|
||||
response_SARA_5 = read_complete_response(ser_sara, wait_for_line="OK")
|
||||
print(response_SARA_5)
|
||||
time.sleep(1)
|
||||
|
||||
#step 4: set PORT (op_code = 5)
|
||||
print("****")
|
||||
print("SET PORT")
|
||||
command = f'AT+UHTTP={profile_id},5,443\r'
|
||||
print("\033[0;33mSET PORT\033[0m")
|
||||
port = 443
|
||||
command = f'AT+UHTTP={profile_id},5,{port}\r'
|
||||
ser_sara.write((command + '\r').encode('utf-8'))
|
||||
response_SARA_55 = read_complete_response(ser_sara)
|
||||
response_SARA_55 = read_complete_response(ser_sara, wait_for_line="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("****")
|
||||
print("SET SSL")
|
||||
command = f'AT+UHTTP={profile_id},6,1\r'
|
||||
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)
|
||||
response_SARA_5 = read_complete_response(ser_sara, wait_for_line="OK")
|
||||
print(response_SARA_5)
|
||||
time.sleep(1)
|
||||
|
||||
@@ -162,53 +240,91 @@ try:
|
||||
# 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 + '\r').encode('utf-8'))
|
||||
response_SARA_1 = read_complete_response(ser_sara)
|
||||
print("Open JSON:")
|
||||
ser_sara.write(command.encode('utf-8'))
|
||||
response_SARA_1 = read_complete_response(ser_sara, wait_for_line=">")
|
||||
print(response_SARA_1)
|
||||
time.sleep(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)
|
||||
print("Write to memory:")
|
||||
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("Trigger POST REQUEST")
|
||||
#command = f'AT+UHTTPC={profile_id},1,"/tests/test.php","https.resp"\r'
|
||||
command = f'AT+UHTTPC={profile_id},4,"/nebuleair?token=2AFF6dQk68daFZ","https.resp","sensordata_json.json",4\r'
|
||||
print("\033[0;33mPOST REQUEST\033[0m")
|
||||
#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_received = False
|
||||
while not response_received:
|
||||
response = read_complete_response(ser_sara, timeout=5)
|
||||
print(response)
|
||||
if "+UUHTTPCR" in response:
|
||||
response_received = True
|
||||
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("Read reply from server")
|
||||
print("\033[0;33mREPLY SERVER\033[0m")
|
||||
ser_sara.write(b'AT+URDFILE="https.resp"\r')
|
||||
response_SARA_7 = read_complete_response(ser_sara)
|
||||
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("Empty SARA memory:")
|
||||
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)
|
||||
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
|
||||
0 No error
|
||||
73 Secure socket connect error
|
||||
'''
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
Reference in New Issue
Block a user