175 lines
6.2 KiB
Python
175 lines
6.2 KiB
Python
'''
|
||
____ _ ____ _
|
||
/ ___| / \ | _ \ / \
|
||
\___ \ / _ \ | |_) | / _ \
|
||
___) / ___ \| _ < / ___ \
|
||
|____/_/ \_\_| \_\/_/ \_\
|
||
|
||
Script that starts at the boot of the RPI
|
||
|
||
/usr/bin/python3 /var/www/nebuleair_pro_4g/SARA/reboot/start.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('<h3>Start reboot python script</h3>')
|
||
|
||
# 1. Set AIRCARTO URL
|
||
print('➡️Set aircarto URL')
|
||
aircarto_profile_id = 0
|
||
aircarto_url="data.nebuleair.fr"
|
||
command = f'AT+UHTTP={aircarto_profile_id},1,"{aircarto_url}"\r'
|
||
ser_sara.write(command.encode('utf-8'))
|
||
response_SARA_1 = read_complete_response(ser_sara, wait_for_lines=["OK"])
|
||
print(response_SARA_1)
|
||
time.sleep(1)
|
||
|
||
#2. Set uSpot URL
|
||
print('➡️Set uSpot URL')
|
||
uSpot_profile_id = 1
|
||
uSpot_url="api-prod.uspot.probesys.net"
|
||
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'
|
||
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)
|
||
|
||
#3. Get localisation (CellLocate)
|
||
mode = 2
|
||
sensor = 2
|
||
response_type = 0
|
||
timeout_s = 2
|
||
accuracy_m = 1
|
||
command = f'AT+ULOC={mode},{sensor},{response_type},{timeout_s},{accuracy_m}\r'
|
||
ser_sara.write((command + '\r').encode('utf-8'))
|
||
response_SARA_3 = read_complete_response(ser_sara, wait_for_lines=["+UULOC"])
|
||
print(response_SARA_3)
|
||
|
||
match = re.search(r"\+UULOC: \d{2}/\d{2}/\d{4},\d{2}:\d{2}:\d{2}\.\d{3},([-+]?\d+\.\d+),([-+]?\d+\.\d+)", response_SARA_3)
|
||
if match:
|
||
latitude = match.group(1)
|
||
longitude = match.group(2)
|
||
print(f"📍 Latitude: {latitude}, Longitude: {longitude}")
|
||
else:
|
||
print("❌ Failed to extract coordinates.")
|
||
|
||
#update config.json
|
||
update_json_key(config_file, "latitude_raw", float(latitude))
|
||
update_json_key(config_file, "longitude_raw", float(longitude))
|
||
|
||
time.sleep(1)
|
||
|
||
except Exception as e:
|
||
print("An error occurred:", e)
|
||
traceback.print_exc() # This prints the full traceback |