PaulVua/nebuleair_pro_4g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

update

Browse Source
This commit is contained in:
PaulVua
2025-02-07 13:31:07 +01:00
parent d98eb48535
commit fb44b57ac1
5 changed files with 88 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
loop/SARA_send_data_v2.py
View File

@@ -218,15 +218,19 @@ ser_sara = serial.Serial(
timeout = 2
)
def read_complete_response(serial_connection, timeout=2, end_of_response_timeout=2, wait_for_line=None, debug=True):
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:
@@ -234,25 +238,29 @@ def read_complete_response(serial_connection, timeout=2, end_of_response_timeout
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:
if debug: print(f"[DEBUG] 🔎Found target line: {wait_for_line}")
break
# 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}")
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.")
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. ⏱️")
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] 🚨 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 response.decode('utf-8', errors='replace') # Return the full response if no target line is found
try:
'''
@@ -297,7 +305,7 @@ try:
# Getting the LTE Signal
print("-> Getting LTE signal <-")
ser_sara.write(b'AT+CSQ\r')
response2 = read_complete_response(ser_sara, wait_for_line="OK")
response2 = read_complete_response(ser_sara, wait_for_lines=["OK"])
print('<p class="text-danger-emphasis">')
print(response2)
print("</p>")
@@ -338,22 +346,22 @@ try:
print("Open JSON:")
command = f'AT+UDWNFILE="sensordata_csv.json",{size_of_string}\r'
ser_sara.write(command.encode('utf-8'))
response_SARA_1 = read_complete_response(ser_sara, wait_for_line=">", debug=False)
response_SARA_1 = read_complete_response(ser_sara, wait_for_lines=[">"], debug=False)
print(response_SARA_1)
time.sleep(1)
#2. Write to shell
print("Write data to memory:")
ser_sara.write(csv_string.encode())
response_SARA_2 = read_complete_response(ser_sara, wait_for_line="OK", debug=False)
response_SARA_2 = read_complete_response(ser_sara, wait_for_lines=["OK"], debug=False)
print(response_SARA_2)
#3. Send to endpoint (with device ID)
print("Send data (POST REQUEST):")
command= f'AT+UHTTPC={aircarto_profile_id},4,"/pro_4G/data.php?sensor_id={device_id}?timestamp=000","server_response.txt","sensordata_csv.json",4\r'
command= f'AT+UHTTPC={aircarto_profile_id},4,"/pro_4G/data.php?sensor_id={device_id}&datetime=000","server_response.txt","sensordata_csv.json",4\r'
ser_sara.write(command.encode('utf-8'))
response_SARA_3 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=120, wait_for_line="+UUHTTPCR")
response_SARA_3 = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=120, wait_for_lines=["+UUHTTPCR", "+CME ERROR"], debug=True)
print('<p class="text-danger-emphasis">')
print(response_SARA_3)
@@ -433,7 +441,7 @@ try:
print("Getting error code (11->Server connection error, 73->Secure socket connect error)")
command = f'AT+UHTTPER={aircarto_profile_id}\r'
ser_sara.write(command.encode('utf-8'))
response_SARA_9 = read_complete_response(ser_sara, wait_for_line="OK", debug=False)
response_SARA_9 = read_complete_response(ser_sara, wait_for_lines=["OK"], debug=False)
print('<p class="text-danger-emphasis">')
print(response_SARA_9)
print("</p>")
@@ -470,22 +478,58 @@ try:
#4. Read reply from server
print("Reply from server:")
ser_sara.write(b'AT+URDFILE="server_response.txt"\r')
response_SARA_4 = read_complete_response(ser_sara, wait_for_line="OK", debug=False)
response_SARA_4 = read_complete_response(ser_sara, wait_for_lines=["OK"], debug=False)
print('<p class="text-success">')
print(response_SARA_4)
print('</p>')
#Si non ne recoit pas de réponse UHTTPCR
#on a peut etre une ERROR de type "+CME ERROR: No connection to phone"
else:
print('<span style="color: red;font-weight: bold;">No UUHTTPCR response</span>')
print("Blink red LED")
# Run LED blinking in a separate thread
led_thread = Thread(target=blink_led, args=(24, 5, 0.5))
led_thread.start()
#Vérification de l'erreur
print("Getting type of error")
# Split the response into lines and search for "+CME ERROR:"
lines2 = response_SARA_3.strip().splitlines()
for line in lines2:
if "+CME ERROR" in line:
error_message = line.split("+CME ERROR:")[1].strip()
print("*****")
print('<span style="color: red;font-weight: bold;">⚠️ATTENTION: CME ERROR⚠️</span>')
print(f"Error type: {error_message}")
print("*****")
# Handle "No connection to phone" error
if error_message == "No connection to phone":
print('<span style="color: orange;font-weight: bold;">📞Try reconnect to network📞</span>')
#IMPORTANT!
# Reconnexion au réseau (AT+COPS)
#command = f'AT+COPS=1,2,{selected_networkID}\r'
command = f'AT+COPS=0\r'
ser_sara.write(command.encode('utf-8'))
responseReconnect = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=120, wait_for_lines=["OK", "+CME ERROR"], debug=True)
print('<p class="text-danger-emphasis">')
print(responseReconnect)
print("</p>")
# Handle "Operation not allowed" error
if error_message == "Operation not allowed":
print('<span style="color: orange;font-weight: bold;">❓Try Resetting the HTTP Profile❓</span>')
command = f'AT+UHTTP={aircarto_profile_id},1,"data.nebuleair.fr"\r'
ser_sara.write(command.encode('utf-8'))
responseResetHTTP_profile = read_complete_response(ser_sara, timeout=5, end_of_response_timeout=5, wait_for_lines=["OK", "+CME ERROR"], debug=True)
print('<p class="text-danger-emphasis">')
print(responseResetHTTP_profile)
print("</p>")
#5. empty json
print("Empty SARA memory:")
ser_sara.write(b'AT+UDELFILE="sensordata_csv.json"\r')
response_SARA_5 = read_complete_response(ser_sara, wait_for_line="OK", debug=False)
response_SARA_5 = read_complete_response(ser_sara, wait_for_lines=["OK"], debug=False)
print(response_SARA_5)